Replace paskia.db.logging with kanta's built-in logging (kanta 0.7.0)

The vendored db/logging module duplicated what kanta now provides:
diff formatting, UUID-to-label resolution via logfmt callbacks, unsafe
character filtering and value truncation. Censoring of oidc.key material
moves into the format_log_uuid logfmt callback in db.lifecycle, taking
care to hide only the value, not the 'key' path component itself.

frontend/auth/admin/AdminApp.vue

@@ -736,10 +736,7 @@ async function refreshUserDetail() {
   }
 }
 
-async function onUserNameSaved() {
-  await refreshUserDetail()
-  authStore.showMessage('User renamed', 'success', 1500)
-}
+
 
 async function submitDialog() {
   if (!dialog.value.type || dialog.value.busy) return
@@ -824,7 +821,7 @@ async function submitDialog() {
       apiJson(`/auth/api/admin/users/${user.uuid}/info`, { method: 'PATCH', body: { display_name: name } })
         .then(() => {
           authStore.showMessage(`User renamed to "${name}".`, 'success', 2500)
-          onUserNameSaved()
+          refreshUserDetail()
         })
         .catch(e => {
           authStore.showMessage(e.message || 'Failed to update user name', 'error')
@@ -1003,9 +1000,7 @@ async function submitDialog() {
                   :show-reg-modal="showRegModal"
                   :navigation-disabled="hasActiveModal"
                   @generate-user-registration-link="generateUserRegistrationLink"
-                  @go-overview="goOverview"
                   @open-org="openOrg"
-                  @on-user-name-saved="onUserNameSaved"
                   @refresh-user-detail="refreshUserDetail"
                   @edit-user-name="editUserName"
                   @close-reg-modal="showRegModal = false"

frontend/src/admin/AdminUserDetail.vue

@@ -18,7 +18,7 @@ const props = defineProps({
   navigationDisabled: { type: Boolean, default: false }
 })
 
-const emit = defineEmits(['generateUserRegistrationLink', 'goOverview', 'openOrg', 'onUserNameSaved', 'closeRegModal', 'editUserName', 'refreshUserDetail', 'navigateOut', 'deleteUser'])
+const emit = defineEmits(['generateUserRegistrationLink', 'openOrg', 'closeRegModal', 'editUserName', 'refreshUserDetail', 'navigateOut', 'deleteUser'])
 
 const authStore = useAuthStore()
 const terminatingSessions = ref({})
@@ -69,11 +69,13 @@ async function handleDelete(credential) {
   try {
     const data = await apiJson(`/auth/api/admin/users/${props.selectedUser.uuid}/credentials/${credential.credential}`, { method: 'DELETE' })
     if (data.status === 'ok') {
-      emit('onUserNameSaved') // Reuse to refresh user detail
+      emit('refreshUserDetail')
+      authStore.showMessage('Passkey removed', 'success', 2500)
     } else {
-      console.error('Failed to delete credential', data)
+      authStore.showMessage(data.detail || 'Failed to remove passkey', 'error')
     }
   } catch (err) {
+    authStore.showMessage(err.message || 'Failed to remove passkey', 'error')
     console.error('Delete credential error', err)
   }
 }
@@ -90,7 +92,7 @@ async function handleTerminateSession(session) {
         location.reload()
         return
       }
-      emit('refreshUserDetail') // Refresh without showing rename message
+      emit('refreshUserDetail')
       authStore.showMessage('Session terminated', 'success', 2500)
     } else {
       authStore.showMessage(data.detail || 'Failed to terminate session', 'error')
@@ -227,7 +229,6 @@ const adminPictureTitle = computed(() => {
         :org-display-name="userDetail.org.display_name"
         :role-name="userDetail.role.display_name"
         :update-endpoint="`/auth/api/admin/users/${selectedUser.uuid}/info`"
-        @saved="$emit('onUserNameSaved')"
         @avatar-click="openPictureDialog"
         @edit="handleEditName"
       >
@@ -287,7 +288,7 @@ const adminPictureTitle = computed(() => {
     <RegistrationLinkModal
       v-if="showRegModal"
       :endpoint="`/auth/api/admin/users/${selectedUser.uuid}/create-link`"
-      :user-name="userDetail?.display_name || selectedUser.display_name"
+      :user-name="userDetail?.user?.display_name || selectedUser.display_name"
       @close="$emit('closeRegModal')"
       @copied="onLinkCopied"
     />

paskia/__main__.py

@@ -1,16 +1,20 @@
 import argparse
+import asyncio
 import logging
 import os
 import sys
+from pathlib import Path
 
 import msgspec
 from fastapi_vue import server
 from fastapi_vue.hostutil import parse_endpoints
+from kanta import Kanta
 
 from paskia._version import __version__
-from paskia.db.jsonl import load_readonly
 from paskia.db.paths import db_file_path
+from paskia.db.structs import DB, Config
 from paskia.util import startupbox
+from paskia.util.constants import DEFAULT_PORT, DEVMODE
 from paskia.util.hostutil import (
     normalize_auth_host_and_origins,
     normalize_origin,
@@ -18,9 +22,6 @@ from paskia.util.hostutil import (
 )
 from paskia.util.runtime import RuntimeConfig
 
-DEFAULT_PORT = 4401
-DEVMODE = os.getenv("PASKIA_DEV") == "1"
-
 EPILOG = """\
 Example:
   paskia --rp-id example.com --rp-name "Example Corporation" --auth-host auth.example.com
@@ -50,6 +51,36 @@ def add_common_options(p: argparse.ArgumentParser) -> None:
     )
 
 
+def _load_stored_config(db_path: Path, *, rp_id: str) -> Config:
+    """Load the stored Config from disk using Kanta in read-only mode.
+
+    This must not depend on PASKIA_CONFIG or the global lifecycle Kanta.
+    If the database file does not exist, a default config is returned.
+    """
+    if not db_path.exists():
+        return Config(rp_id=rp_id)
+
+    kanta = Kanta(
+        str(db_path),
+        DB(config=Config(rp_id=rp_id)),
+        migrations="paskia.db.migrations",
+    )
+    kanta.ctx.rp_id = rp_id
+
+    async def _read() -> Config:
+        await kanta.open(readonly=True)
+        try:
+            return kanta.data.config
+        finally:
+            await kanta.close()
+
+    try:
+        return asyncio.run(_read())
+    except Exception as e:
+        logging.exception("Failed to load database")
+        raise SystemExit(f"{e}") from e
+
+
 def main():
     # Configure logging to remove the "ERROR:root:" prefix
     logging.basicConfig(level=logging.INFO, format="%(message)s", force=True)
@@ -75,10 +106,12 @@ def main():
 
     args = parser.parse_args()
 
-    # Load stored config (read-only, no writes, no global state)
+    # Load stored config using a local read-only Kanta instance.
+    # This happens before PASKIA_CONFIG is set, so we must not import
+    # modules that initialize the global database lifecycle.
     db_path = db_file_path(rp_id=args.rp_id, create_root=True)
     try:
-        config = load_readonly(str(db_path), rp_id=args.rp_id).config
+        config = _load_stored_config(db_path, rp_id=args.rp_id)
     except SystemExit as e:
         print(f"🛑 Paskia {__version__} could not load")
         sys.exit(str(e))

paskia/bootstrap.py

@@ -4,13 +4,17 @@ Bootstrap module for passkey authentication system.
 This module handles initial system setup when a new database is created,
 including creating default admin user, organization, permissions, and
 generating a reset link for initial admin setup.
+
+The actual database seeding is performed by the module-level kanta bootstrap
+callback defined in :mod:`paskia.db.bootstrap` and registered during
+:func:`paskia.db.lifecycle.init`.
 """
 
 import logging
 
 from paskia import authsession, db
+from paskia.db.bootstrap import log_reset_link
 from paskia.db.structs import Config
-from paskia.util import hostutil
 
 logger = logging.getLogger(__name__)
 
@@ -27,37 +31,10 @@ def _configure_logger() -> None:
 
 _configure_logger()
 
-# Shared log message template for admin reset links
-ADMIN_RESET_MESSAGE = """
-👤 Admin  %s
-   - Use this link to register a Passkey for the admin user!
-"""
-
 
 def _log_reset_link(passphrase: str, message: str | None = None) -> str:
     """Log a reset link message and return the URL."""
-    reset_link = hostutil.reset_link_url(passphrase)
-    if message:
-        logger.info(message)
-    logger.info(ADMIN_RESET_MESSAGE, reset_link)
-    return reset_link
-
-
-async def bootstrap_system(config: Config | None = None) -> None:
-    """
-    Bootstrap the entire system with default data.
-
-    Uses db.bootstrap() which performs all operations in a single transaction.
-    The transaction log will show a single "bootstrap" action with all changes.
-
-    Args:
-        config: Configuration to store (rp_id, rp_name, origins, etc.)
-    """
-    # Call the single-transaction bootstrap function
-    reset_passphrase = db.bootstrap(config=config)
-
-    # Log the reset link (this is separate from the transaction log)
-    _log_reset_link(reset_passphrase, "✅ Bootstrap completed!")
+    return log_reset_link(passphrase, message)
 
 
 async def check_admin_credentials() -> bool:
@@ -114,22 +91,18 @@ async def check_admin_credentials() -> bool:
 
 async def bootstrap_if_needed(config: Config | None = None) -> bool:
     """
-    Check if system needs bootstrapping and perform it if necessary.
+    Check if admin needs credentials and create a reset link if needed.
+
+    Database bootstrapping itself is now handled automatically during
+    ``db.init()`` via the registered kanta bootstrap callback. This function
+    remains as a post-init hook for credential checks.
 
     Args:
-        config: Configuration to store during bootstrap (rp_id, rp_name, origins, etc.)
+        config: Kept for backwards compatibility; config is now applied during
+            ``db.init()``.
 
     Returns:
-        bool: True if bootstrapping was performed, False if system was already set up
+        bool: Always returns False (bootstrapping is performed during init).
     """
-    # Check if the admin permission exists - if it does, system is already bootstrapped
-    if any(p.scope == "auth:admin" for p in db.data().permissions.values()):
-        # Permission exists, system is already bootstrapped
-        # Check if admin needs credentials (only for already-bootstrapped systems)
-        await check_admin_credentials()
-        return False
-
-    # No admin permission found, need to bootstrap
-    # Bootstrap creates the admin user AND the reset link, so no need to check credentials after
-    await bootstrap_system(config=config)
-    return True
+    await check_admin_credentials()
+    return False

paskia/db/__init__.py

@@ -19,15 +19,7 @@ Usage:
 """
 
 import paskia.db.operations as operations
-from paskia.db.background import (
-    start_background,
-    start_cleanup,
-    stop_background,
-    stop_cleanup,
-)
 from paskia.db.bootstrap import bootstrap
-from paskia.db.jsonl import load_readonly
-from paskia.db.lifecycle import cleanup_expired, init
 from paskia.db.operations import (
     add_permission_to_org,
     add_permission_to_role,
@@ -101,19 +93,11 @@ __all__ = [
     "User",
     # Instance
     "data",
-    "init",
-    "load_readonly",
-    # Background
-    "start_background",
-    "stop_background",
-    "start_cleanup",
-    "stop_cleanup",
     # Read ops
     # Write ops
     "add_permission_to_org",
     "add_permission_to_role",
     "bootstrap",
-    "cleanup_expired",
     "create_credential",
     "create_credential_session",
     "create_org",

paskia/db/background.py

@@ -7,12 +7,7 @@ companion task that periodically cleans up expired sessions/tokens.
 
 import asyncio
 import logging
-import os
-import signal
 
-from kanta.exceptions import DatabaseError
-
-import paskia.db.operations as _ops
 from paskia.db.lifecycle import cleanup_expired
 
 CLEANUP_INTERVAL = 1  # Expired item cleanup
@@ -21,24 +16,6 @@ _logger = logging.getLogger(__name__)
 _background_task: asyncio.Task | None = None
 
 
-def _sigterm_on_error(error: DatabaseError) -> None:
-    """Exit the server when a database write fails."""
-    _logger.error("Fatal database error: %s", error)
-    os.kill(os.getpid(), signal.SIGTERM)
-
-
-async def flush() -> None:
-    """Write all pending database changes to disk."""
-    store = _ops._store
-    if store is None:
-        _logger.warning("flush() called but _store is None")
-        return
-    try:
-        await store.flush()
-    except DatabaseError as e:
-        _sigterm_on_error(e)
-
-
 async def _background_loop():
     """Background task that periodically cleans up expired items."""
     # Run cleanup immediately on startup to clear old expired items
@@ -87,7 +64,7 @@ async def start_background():
 
 
 async def stop_background():
-    """Stop the background cleanup task and close kanta."""
+    """Stop the background cleanup task."""
     global _background_task
     if _background_task:
         _background_task.cancel()
@@ -96,12 +73,6 @@ async def stop_background():
         except asyncio.CancelledError:
             pass
         _background_task = None
-    store = _ops._store
-    if store is not None:
-        try:
-            await store.close()
-        except DatabaseError as e:
-            _sigterm_on_error(e)
 
 
 # Aliases for backwards compatibility

paskia/db/bootstrap.py

@@ -2,24 +2,60 @@
 Bootstrap operations for initial system setup.
 """
 
+import logging
+import sys
 from datetime import UTC, datetime
 
 import uuid7
 
-import paskia.db.operations as _ops
 from paskia.authsession import reset_expires
-from paskia.db.structs import Config, Org, Permission, ResetToken, Role, User
+from paskia.db.structs import DB, Config, Org, Permission, ResetToken, Role, User
 from paskia.util.crypto import secret_key
+from paskia.util.hostutil import reset_link_url
+
+_reset_link_logger = logging.getLogger("paskia.reset_link")
+
+
+def _configure_reset_link_logger() -> None:
+    if _reset_link_logger.handlers:
+        return
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(logging.Formatter("%(message)s"))
+    _reset_link_logger.addHandler(handler)
+    _reset_link_logger.setLevel(logging.INFO)
+    _reset_link_logger.propagate = False
+
+
+_configure_reset_link_logger()
+
+ADMIN_RESET_MESSAGE = """
+👤 Admin  %s
+   - Use this link to register a Passkey for the admin user!
+"""
+
+
+def log_reset_link(passphrase: str, message: str | None = None) -> str:
+    """Log a reset link message and return the URL."""
+    reset_link = reset_link_url(passphrase)
+    if message:
+        _reset_link_logger.info(message)
+    _reset_link_logger.info(ADMIN_RESET_MESSAGE, reset_link)
+    return reset_link
 
 
 def bootstrap(
+    data: "DB",
     org_name: str = "Organization",
     admin_name: str = "Admin",
     reset_passphrase: str | None = None,
     reset_expiry: datetime | None = None,
     config: Config | None = None,
 ) -> str:
-    """Bootstrap the entire system in a single transaction.
+    """Bootstrap the entire system by seeding an empty database.
+
+    This is intended to be called from a ``@kanta.bootstrap`` callback during
+    ``kanta.open()``. It mutates the provided root ``data`` object directly;
+    kanta queues the resulting state as the initial "bootstrap" change record.
 
     Creates:
     - auth:admin permission (Master Admin)
@@ -29,10 +65,8 @@ def bootstrap(
     - Reset token for admin registration
     - Config (if provided)
 
-    This is the only way to create a new database file.
-    All data is created atomically - if any step fails, nothing is written.
-
     Args:
+        data: The live root database object (usually a ``DB`` instance).
         org_name: Display name for the organization (default: "Organization")
         admin_name: Display name for the admin user (default: "Admin")
         reset_passphrase: Passphrase for the reset token (generated if not provided)
@@ -44,7 +78,7 @@ def bootstrap(
     """
 
     # Check if system is already bootstrapped
-    for p in _ops._db.permissions.values():
+    for p in data.permissions.values():
         if p.scope == "auth:admin":
             raise ValueError(
                 "System already bootstrapped (auth:admin permission exists)"
@@ -62,65 +96,66 @@ def bootstrap(
     if reset_expiry is None:
         reset_expiry = reset_expires()
 
-    with _ops._db.transaction("bootstrap"):
-        # Create auth:admin permission
-        perm_admin = Permission(
-            scope="auth:admin",
-            display_name="Master Admin",
-            orgs={org_uuid: True},  # Grant to org
-        )
-        perm_admin.uuid = perm_admin_uuid
-        perm_admin.store()
+    # Create auth:admin permission
+    perm_admin = Permission(
+        scope="auth:admin",
+        display_name="Master Admin",
+        orgs={org_uuid: True},  # Grant to org
+    )
+    perm_admin.uuid = perm_admin_uuid
 
-        # Create auth:org:admin permission
-        perm_org_admin = Permission(
-            scope="auth:org:admin",
-            display_name="Org Admin",
-            orgs={org_uuid: True},  # Grant to org
-        )
-        perm_org_admin.uuid = perm_org_admin_uuid
-        perm_org_admin.store()
+    # Create auth:org:admin permission
+    perm_org_admin = Permission(
+        scope="auth:org:admin",
+        display_name="Org Admin",
+        orgs={org_uuid: True},  # Grant to org
+    )
+    perm_org_admin.uuid = perm_org_admin_uuid
 
-        # Create organization
-        new_org = Org.create(display_name=org_name)
-        new_org.uuid = org_uuid
-        new_org.store()
+    # Create organization
+    new_org = Org.create(display_name=org_name)
+    new_org.uuid = org_uuid
 
-        # Create Administration role with both permissions
-        admin_role = Role(
-            org_uuid=org_uuid,
-            display_name="Administration",
-            permissions={perm_admin_uuid: True, perm_org_admin_uuid: True},
-        )
-        admin_role.uuid = role_uuid
-        admin_role.store()
+    # Create Administration role with both permissions
+    admin_role = Role(
+        org_uuid=org_uuid,
+        display_name="Administration",
+        permissions={perm_admin_uuid: True, perm_org_admin_uuid: True},
+    )
+    admin_role.uuid = role_uuid
 
-        # Create admin user
-        admin_user = User(
-            display_name=admin_name,
-            role_uuid=role_uuid,
-            created_at=now,
-            last_seen=None,
-            visits=0,
-            theme="",
-        )
-        admin_user.uuid = user_uuid
-        admin_user.store()
+    # Create admin user
+    admin_user = User(
+        display_name=admin_name,
+        role_uuid=role_uuid,
+        created_at=now,
+        last_seen=None,
+        visits=0,
+        theme="",
+    )
+    admin_user.uuid = user_uuid
 
-        # Create reset token
-        reset_token, reset_passphrase = ResetToken.create(
-            user=user_uuid,
-            expiry=reset_expiry,
-            token_type="admin bootstrap",
-            passphrase=reset_passphrase,
-        )
-        reset_token.store()
+    # Create reset token
+    reset_token, reset_passphrase = ResetToken.create(
+        user=user_uuid,
+        expiry=reset_expiry,
+        token_type="admin bootstrap",
+        passphrase=reset_passphrase,
+    )
 
-        # Set config if provided
-        if config is not None:
-            _ops._db.config = config
+    # Set config if provided
+    if config is not None:
+        data.config = config
 
-        # Generate OIDC signing key
-        _ops._db.oidc.key = secret_key()
+    # Generate OIDC signing key
+    data.oidc.key = secret_key()
+
+    # Store all bootstrapped objects in the live data object
+    data.permissions[perm_admin_uuid] = perm_admin
+    data.permissions[perm_org_admin_uuid] = perm_org_admin
+    data.orgs[org_uuid] = new_org
+    data.roles[role_uuid] = admin_role
+    data.users[user_uuid] = admin_user
+    data.reset_tokens[reset_token.key] = reset_token
 
     return reset_passphrase

paskia/db/jsonl.py

@@ -1,52 +0,0 @@
-"""
-JSONL read-only loader using kanta.
-"""
-
-import logging
-from pathlib import Path
-
-import msgspec
-from kanta import replay as replay_jsonl
-from kanta.migrate import MigrationRegistry
-
-from paskia.db.migrations import MigrationCtx
-from paskia.db.structs import DB, Config
-
-_logger = logging.getLogger(__name__)
-
-
-def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
-    """Replay JSONL and apply migrations to produce a DB, without writing anything.
-
-    This is suitable for reading settings before the server starts.
-    Migrations are applied in-memory only; nothing is queued or flushed.
-    """
-    path = Path(db_path)
-    if not path.exists():
-        return DB(config=Config(rp_id=rp_id))
-
-    try:
-        content = path.read_bytes()
-        rr = replay_jsonl(content)
-        data_dict = rr.state
-        version = rr.version
-
-        if not data_dict:
-            return DB(config=Config(rp_id=rp_id))
-
-        # Apply migrations in-memory (no persistence)
-        registry = MigrationRegistry.from_module("paskia.db.migrations")
-        version = registry.apply(
-            data_dict, version, MigrationCtx(rp_id=rp_id), silent=True
-        )
-
-        # Decode to msgspec struct
-        return msgspec.json.decode(msgspec.json.encode(data_dict), type=DB)
-    except OSError as e:
-        _logger.exception("Failed to load database")
-        raise SystemExit(f"{e}")
-    except (ValueError, msgspec.DecodeError) as e:
-        raise SystemExit(f"{e}")
-    except Exception as e:
-        _logger.exception("Unexpected error loading database")
-        raise SystemExit(f"{e}")

paskia/db/lifecycle.py

@@ -2,10 +2,14 @@
 Database lifecycle: initialization and maintenance.
 """
 
+import asyncio
 import logging
 import os
 import signal
 from datetime import UTC, datetime
+from pathlib import Path
+from typing import Annotated, Any, Optional
+from uuid import UUID
 
 from kanta import Kanta
 from kanta.exceptions import DatabaseError
@@ -13,58 +17,142 @@ from kanta.exceptions import DatabaseError
 import paskia.db.operations as _ops
 from paskia import oidc_notify
 from paskia.authsession import EXPIRES
-from paskia.db.migrations import MigrationCtx
+from paskia.db.bootstrap import bootstrap, log_reset_link
 from paskia.db.paths import db_file_path
 from paskia.db.structs import DB
+from paskia.util.runtime import config as runtime_config
 
-_logger = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 
-def _fatal_error(error: DatabaseError) -> None:
+runtime = runtime_config()
+if runtime is None:
+    raise RuntimeError("PASKIA_CONFIG must be defined before importing db.lifecycle")
+
+kanta = Kanta(
+    str(db_file_path(rp_id=runtime.config.rp_id, create_root=False)),
+    _ops._db,
+    migrations="paskia.db.migrations",
+)
+kanta.ctx.rp_id = runtime.config.rp_id
+_ops._db._store = kanta
+
+
+def _lookup_uuid_in_state(state: dict | None, uuid_str: str) -> str | None:
+    """Resolve UUID to label from serialized state dict."""
+    if not state:
+        return None
+
+    # Display-name based entities.
+    for bucket in ("users", "orgs", "roles", "permissions"):
+        entity = state.get(bucket, {}).get(uuid_str)
+        if isinstance(entity, dict):
+            display_name = entity.get("display_name")
+            if isinstance(display_name, str) and display_name:
+                return display_name
+
+    # OIDC clients use "name" instead of "display_name".
+    client = state.get("oidc", {}).get("clients", {}).get(uuid_str)
+    if isinstance(client, dict):
+        name = client.get("name")
+        if isinstance(name, str) and name:
+            return name
+
+    return None
+
+
+def _resolve_uuid_label(
+    uuid_str: str,
+    *,
+    previous: dict | None = None,
+    current: dict | None = None,
+) -> str | None:
+    """Resolve known entity UUIDs to human-readable labels."""
+    # Prefer previous state so deletions/renames still show a useful label.
+    label = _lookup_uuid_in_state(previous, uuid_str)
+    if label:
+        return label
+    label = _lookup_uuid_in_state(current, uuid_str)
+    if label:
+        return label
+
+    try:
+        uid = UUID(uuid_str)
+    except ValueError:
+        return None
+
+    if uid in _ops._db.users:
+        return _ops._db.users[uid].display_name
+    if uid in _ops._db.orgs:
+        return _ops._db.orgs[uid].display_name
+    if uid in _ops._db.roles:
+        return _ops._db.roles[uid].display_name
+    if uid in _ops._db.permissions:
+        return _ops._db.permissions[uid].display_name
+    if uid in _ops._db.oidc.clients:
+        return _ops._db.oidc.clients[uid].name
+    return None
+
+
+@kanta.logfmt
+def format_log_uuid(
+    value: Any,
+    path: str,
+    previous: Annotated[dict, "pre"] | None = None,
+    current: Annotated[dict, "post"] | None = None,
+) -> Optional[str]:  # noqa: UP045
+    """Format UUID values/keys/actor labels and censor secrets in transaction logs."""
+    # Censor sensitive OIDC key material regardless of value type, but only
+    # when formatting the value: path components are passed with the component
+    # itself as value and must stay visible ("oidc.key = <hidden>").
+    if (path == "oidc.key" or path.endswith(".oidc.key")) and value != "key":
+        return "<hidden>"
+
+    if not isinstance(value, str):
+        return None
+
+    # Works for transaction actor metadata ($user), values, and path components.
+    return _resolve_uuid_label(value, previous=previous, current=current)
+
+
+@kanta.fatal_error
+def terminate(error: DatabaseError) -> None:
     """Fatal error callback: terminate the process on background write failures."""
-    _logger.error("Fatal database error: %s", error)
+    logger.error("Fatal database error: %s", error)
     os.kill(os.getpid(), signal.SIGTERM)
 
 
-async def init(rp_id: str, *args, **kwargs):
-    """Load database from JSONL file using kanta."""
-    if _ops._store is not None:
-        _logger.debug("Database already initialized, skipping reload")
-        return
-    db_path = db_file_path(rp_id=rp_id, create_root=True)
-    db = DB()
-    kanta = Kanta(
-        str(db_path),
-        db,
-        migrations="paskia.db.migrations",
-        migration_ctx=MigrationCtx(rp_id=rp_id),
-        fatal_error=_fatal_error,
-    )
+@kanta.bootstrap
+def bootstrap_db(data: DB) -> None:
+    reset_passphrase = bootstrap(data, config=runtime.config)
+    log_reset_link(reset_passphrase, "✅ Bootstrap completed!")
+
+
+async def init():
+    """Load database from JSONL file using kanta.
+
+    If the database file is empty, the configured bootstrap callback seeds it
+    with default permissions, organization, role, admin user and a reset token.
+    """
+    rootpath = Path(kanta.filename).parent
     try:
+        await asyncio.to_thread(rootpath.mkdir, parents=True, exist_ok=True)
         await kanta.open()
-    except DatabaseError as e:
+    except Exception as e:
         raise SystemExit(f"{e}") from e
-    _ops._store = kanta
-    _ops._db = db
-    _ops._db._store = kanta
-    # Request a snapshot after successful startup
-    kanta.request_snapshot()
 
 
 def cleanup_expired() -> int:
     """Remove expired sessions and reset tokens. Returns count removed."""
     now = datetime.now(UTC)
-    count = 0
     limit = now - EXPIRES
     expired_sessions = [k for k, s in _ops._db.sessions.items() if s.validated < limit]
     if expired_sessions:
         oidc_notify.schedule_notifications(expired_sessions)
-    with _ops._db.transaction("expiry"):
+    with kanta.transaction("expiry"):
         for k in expired_sessions:
             del _ops._db.sessions[k]
-            count += 1
         expired_tokens = [k for k, t in _ops._db.reset_tokens.items() if t.expiry < now]
         for k in expired_tokens:
             del _ops._db.reset_tokens[k]
-            count += 1
-    return count
+    return len(expired_sessions) + len(expired_tokens)

paskia/db/logging.py

@@ -1,470 +0,0 @@
-"""
-Database change logging with pretty-printed diffs.
-
-Provides a logger for JSONL database changes that formats diffs
-in a human-readable path.notation style with color coding.
-
-UUIDs are replaced with display names where available, or the full UUID string
-for types without display names.
-"""
-
-import logging
-import re
-import sys
-from typing import TYPE_CHECKING, Any
-from uuid import UUID
-
-from kanta.logging import configure_logging as configure_kanta_logging
-
-if TYPE_CHECKING:
-    from paskia.db.structs import DB
-
-logger = logging.getLogger("paskia.db")
-
-# UUID regex pattern (8-4-4-4-12 hex format)
-_UUID_PATTERN = re.compile(
-    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
-)
-
-# Pattern to match control characters and bidirectional overrides
-_UNSAFE_CHARS = re.compile(
-    r"[\x00-\x1f\x7f-\x9f"  # C0 and C1 control characters
-    r"\u200e\u200f"  # LRM, RLM
-    r"\u202a-\u202e"  # LRE, RLE, PDF, LRO, RLO
-    r"\u2066-\u2069"  # LRI, RLI, FSI, PDI
-    r"]"
-)
-
-# ANSI color codes (matching FastAPI logging style)
-_RESET = "\033[0m"
-_SEP = "\033[38;5;242m"  # Dark grey for separators (like host/timing in access log)
-_PATH_PREFIX = "\033[38;5;242m"  # Dark grey for path prefix (like host in access log)
-_PATH_FINAL = "\033[38;5;250m"  # Default for final element (like path in access log)
-_DELETE = "\033[1;31m"  # Red for deletions
-_ADD = "\033[0;32m"  # Green for additions
-_ACTION = "\033[1;34m"  # Bold blue for action name
-_USER = "\033[0;34m"  # Blue for user display
-
-
-def _is_uuid(value: str) -> bool:
-    """Check if a string is a UUID."""
-    return bool(_UUID_PATTERN.match(value))
-
-
-class UuidResolver:
-    """Resolve UUIDs to display names or short suffixes.
-
-    Uses the previous state for lookups to show the name before any changes.
-    """
-
-    def __init__(self, db: "DB | None" = None, previous: dict | None = None):
-        self._db = db
-        self._previous = previous
-
-    def resolve(self, uuid_str: str) -> str:
-        """Resolve a UUID to its display name or the full UUID string."""
-        display = self._get_display_name(uuid_str)
-        if display:
-            return display
-        return uuid_str
-
-    def _get_display_name(self, uuid_str: str) -> str | None:
-        """Look up display name for a UUID.
-
-        First checks the previous state (to show names before changes),
-        then falls back to the current database.
-        """
-        # Try previous state first (for showing name before a change)
-        name = self._lookup_in_previous(uuid_str)
-        if name:
-            return name
-
-        # Fall back to current database
-        return self._lookup_in_db(uuid_str)
-
-    def _lookup_in_previous(self, uuid_str: str) -> str | None:
-        """Look up display name in the previous state dict."""
-        if not self._previous:
-            return None
-
-        # Check users
-        if "users" in self._previous and uuid_str in self._previous["users"]:
-            user_data = self._previous["users"][uuid_str]
-            if isinstance(user_data, dict) and "display_name" in user_data:
-                return user_data["display_name"]
-
-        # Check orgs
-        if "orgs" in self._previous and uuid_str in self._previous["orgs"]:
-            org_data = self._previous["orgs"][uuid_str]
-            if isinstance(org_data, dict) and "display_name" in org_data:
-                return org_data["display_name"]
-
-        # Check roles
-        if "roles" in self._previous and uuid_str in self._previous["roles"]:
-            role_data = self._previous["roles"][uuid_str]
-            if isinstance(role_data, dict) and "display_name" in role_data:
-                return role_data["display_name"]
-
-        # Check permissions
-        if (
-            "permissions" in self._previous
-            and uuid_str in self._previous["permissions"]
-        ):
-            perm_data = self._previous["permissions"][uuid_str]
-            if isinstance(perm_data, dict) and "display_name" in perm_data:
-                return perm_data["display_name"]
-
-        return None
-
-    def _lookup_in_db(self, uuid_str: str) -> str | None:
-        """Look up display name in the current database."""
-        if not self._db:
-            return None
-
-        try:
-            uuid_obj = UUID(uuid_str)
-        except ValueError:
-            return None
-
-        # Check users
-        if uuid_obj in self._db.users:
-            return self._db.users[uuid_obj].display_name
-
-        # Check orgs
-        if uuid_obj in self._db.orgs:
-            return self._db.orgs[uuid_obj].display_name
-
-        # Check roles
-        if uuid_obj in self._db.roles:
-            return self._db.roles[uuid_obj].display_name
-
-        # Check permissions
-        if uuid_obj in self._db.permissions:
-            return self._db.permissions[uuid_obj].display_name
-
-        return None
-
-
-def _format_value(
-    value: Any,
-    max_len: int = 60,
-    resolver: UuidResolver | None = None,
-) -> str:
-    """Format a value for display, truncating if needed.
-
-    If resolver is provided, UUIDs are replaced with display names or short suffixes.
-    """
-    if value is None:
-        return "null"
-
-    if isinstance(value, bool):
-        return "true" if value else "false"
-
-    if isinstance(value, (int, float)):
-        return str(value)
-
-    if isinstance(value, str):
-        # Check if it's a UUID and resolve to display name
-        if resolver and _is_uuid(value):
-            return resolver.resolve(value)
-        # Filter out control characters and bidirectional overrides
-        value = _UNSAFE_CHARS.sub("", value)
-        # Truncate long strings
-        if len(value) > max_len:
-            return value[: max_len - 3] + "..."
-        return value
-
-    if isinstance(value, dict):
-        if not value:
-            return "{}"
-        # Check if all values are True - render as set-like {key1, key2}
-        all_true = all(v is True for v in value.values())
-        parts = []
-        for k, v in value.items():
-            # Replace UUID keys with display names
-            key_display = resolver.resolve(k) if resolver and _is_uuid(k) else k
-            if all_true:
-                parts.append(key_display)
-            else:
-                val_display = _format_value(v, max_len=30, resolver=resolver)
-                parts.append(f"{key_display}: {val_display}")
-        return "{" + ", ".join(parts) + "}"
-
-    if isinstance(value, list):
-        if not value:
-            return "[]"
-        parts = [_format_value(v, max_len=30, resolver=resolver) for v in value]
-        return "[" + ", ".join(parts) + "]"
-
-    # Fallback for other types
-    text = str(value)
-    if len(text) > max_len:
-        text = text[: max_len - 3] + "..."
-    return text
-
-
-def _format_path(path: list[str], resolver: UuidResolver | None = None) -> str:
-    """Format a path as dot notation with prefix in dark grey, final in default.
-
-    If resolver is provided, UUIDs in the path are replaced with display names.
-    """
-    if not path:
-        return ""
-
-    # Replace UUIDs in path with display names
-    if resolver:
-        path = [resolver.resolve(p) if _is_uuid(p) else p for p in path]
-
-    if len(path) == 1:
-        return f"{_PATH_FINAL}{path[0]}{_RESET}"
-    prefix = ".".join(path[:-1])
-    final = path[-1]
-    return f"{_PATH_PREFIX}{prefix}.{_RESET}{_PATH_FINAL}{final}{_RESET}"
-
-
-def _get_nested(data: dict | None, path: list[str]) -> Any:
-    """Get a nested value from a dict by path, or None if not found."""
-    if data is None:
-        return None
-    current = data
-    for key in path:
-        if not isinstance(current, dict) or key not in current:
-            return None
-        current = current[key]
-    return current
-
-
-def _collect_changes(
-    diff: dict,
-    path: list[str],
-    changes: list[tuple[str, list[str], Any]],
-    previous: dict | None,
-) -> None:
-    """
-    Recursively collect changes from a diff into a flat list.
-
-    Each change is a tuple of (change_type, path, new_value).
-    change_type is one of: 'add', 'update', 'delete'
-    """
-    if not isinstance(diff, dict):
-        # Leaf value - check if it existed before
-        existed = _get_nested(previous, path) is not None
-        changes.append(("update" if existed else "add", path, diff))
-        return
-
-    for key, value in diff.items():
-        if key == "$delete":
-            # $delete contains a list of keys to delete
-            if isinstance(value, list):
-                for deleted_key in value:
-                    changes.append(("delete", path + [str(deleted_key)], None))
-            else:
-                changes.append(("delete", path + [str(value)], None))
-
-        elif key == "$replace":
-            # $replace replaces the entire collection at this path
-            # We need to track what was added and what was deleted
-            old_collection = _get_nested(previous, path)
-            old_keys = (
-                set(old_collection.keys())
-                if isinstance(old_collection, dict)
-                else set()
-            )
-            new_keys = set(value.keys()) if isinstance(value, dict) else set()
-
-            # Items that existed before but not in new = deleted
-            for deleted_key in old_keys - new_keys:
-                changes.append(("delete", path + [str(deleted_key)], None))
-
-            # Items in new collection
-            if isinstance(value, dict):
-                for rkey, rval in value.items():
-                    existed = rkey in old_keys
-                    changes.append(
-                        ("update" if existed else "add", path + [str(rkey)], rval)
-                    )
-            elif value or not old_keys:
-                # Non-dict replacement or empty replacement with nothing before
-                changes.append(
-                    ("update" if old_collection is not None else "add", path, value)
-                )
-
-        elif key.startswith("$"):
-            # Other special operations (future-proofing)
-            changes.append(("add", path, {key: value}))
-
-        else:
-            # Regular nested key - check if this item existed before
-            new_path = path + [str(key)]
-            existed = _get_nested(previous, new_path) is not None
-            if existed:
-                # Item exists - recurse to show specific field changes
-                _collect_changes(value, new_path, changes, previous)
-            else:
-                # New item - record as add with full value, don't recurse
-                changes.append(("add", new_path, value))
-
-
-def _format_change_lines(
-    change_type: str,
-    path: list[str],
-    value: Any,
-    resolver: UuidResolver | None = None,
-) -> list[str]:
-    """Format a single change as one or more lines.
-
-    If resolver is provided, UUIDs are replaced with display names.
-    """
-
-    # Helper to format a value, checking for censored paths
-    def fmt_value(v: Any, child_path: list[str]) -> str:
-        if child_path[-2:] == ["oidc", "key"]:
-            return f"{_SEP}<hidden>{_RESET}"
-        return _format_value(v, resolver=resolver)
-
-    # Helper to format path with UUID replacement
-    def fmt_path(p: list[str]) -> list[str]:
-        if resolver:
-            return [resolver.resolve(x) if _is_uuid(x) else x for x in p]
-        return p
-
-    formatted_path = fmt_path(path)
-
-    if change_type == "delete":
-        if len(formatted_path) == 1:
-            return [f"  {_DELETE}{formatted_path[0]} ✗{_RESET}"]
-        prefix = ".".join(formatted_path[:-1])
-        final = formatted_path[-1]
-        return [f"  {_PATH_PREFIX}{prefix}.{_RESET}{_DELETE}{final} ✗{_RESET}"]
-
-    if change_type == "add":
-        # New item being created - only final element in green
-        # For dict values, show children on separate indented lines
-        if isinstance(value, dict) and value:
-            lines = []
-            # First line: path with green final element and grey =
-            if len(formatted_path) == 1:
-                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET}")
-            else:
-                prefix = ".".join(formatted_path[:-1])
-                final = formatted_path[-1]
-                lines.append(
-                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}"
-                )
-            # Child lines: indented key: value, with aligned values
-            # Format keys (may contain UUIDs)
-            formatted_items = []
-            for k, v in value.items():
-                k_display = resolver.resolve(k) if resolver and _is_uuid(k) else k
-                v_str = fmt_value(v, path + [k])
-                formatted_items.append((k_display, v_str))
-            max_key_len = max(len(k) for k, _ in formatted_items)
-            field_width = max(max_key_len, 12)  # minimum 12 chars
-            for k_display, v_str in formatted_items:
-                padding = " " * (field_width - len(k_display))
-                lines.append(f"    {k_display}{_SEP}:{_RESET}{padding} {v_str}")
-            return lines
-        else:
-            value_str = fmt_value(value, path)
-            if len(formatted_path) == 1:
-                return [
-                    f"  {_ADD}{formatted_path[0]}{_RESET} {_SEP}={_RESET} {value_str}"
-                ]
-            prefix = ".".join(formatted_path[:-1])
-            final = formatted_path[-1]
-            return [
-                f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET} {value_str}"
-            ]
-
-    # update: Existing item being updated - normal path colors
-    value_str = fmt_value(value, path)
-    path_str = _format_path(path, resolver=resolver)
-    return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
-
-
-def format_diff(
-    diff: dict, previous: dict | None = None, db: "DB | None" = None
-) -> list[str]:
-    """
-    Format a JSON diff as human-readable lines.
-
-    Args:
-        diff: The JSON diff dict
-        previous: The previous state dict (for determining add vs update)
-        db: Optional database for looking up display names
-
-    Returns a list of formatted lines (without newlines).
-    UUIDs are replaced with display names (using previous state for lookups).
-    """
-    changes: list[tuple[str, list[str], Any]] = []
-    _collect_changes(diff, [], changes, previous)
-
-    if not changes:
-        return []
-
-    # Create resolver for UUID replacement (uses previous state for lookups)
-    resolver = UuidResolver(db, previous)
-
-    # Format each change
-    lines = []
-    for change_type, path, value in changes:
-        lines.extend(_format_change_lines(change_type, path, value, resolver))
-
-    return lines
-
-
-def format_action_header(action: str, user_display: str | None = None) -> str:
-    """Format the action header line."""
-    action_str = f"{_ACTION}{action}{_RESET}"
-    if user_display:
-        user_str = f"{_USER}{user_display}{_RESET}"
-        return f"{action_str} by {user_str}"
-    return action_str
-
-
-def log_change(
-    action: str,
-    diff: dict,
-    user_display: str | None = None,
-    previous: dict | None = None,
-    db: "DB | None" = None,
-) -> None:
-    """
-    Log a database change with pretty-printed diff.
-
-    UUIDs are replaced with display names for readability. For types without
-    display names, the full UUID string is used.
-
-    Args:
-        action: The action name (e.g., "login", "admin:delete_user")
-        diff: The JSON diff dict
-        user_display: Optional display name of the user who performed the action
-        previous: The previous state dict (for determining add vs update)
-        db: Optional database for looking up display names
-    """
-    header = format_action_header(action, user_display)
-    diff_lines = format_diff(diff, previous, db)
-
-    if not diff_lines:
-        logger.info(header)
-        return
-
-    if len(diff_lines) == 1:
-        # Single change - combine on one line
-        logger.info(f"{header}{diff_lines[0]}")
-    else:
-        # Multiple changes - header on its own line, then changes
-        logger.info(header)
-        for line in diff_lines:
-            logger.info(line)
-
-
-def configure_db_logging() -> None:
-    """Configure the database logger to output to stderr without prefix."""
-    handler = logging.StreamHandler(sys.stderr)
-    handler.setFormatter(logging.Formatter("%(message)s"))
-    logger.addHandler(handler)
-    logger.setLevel(logging.INFO)
-    logger.propagate = False
-    # Kanta logs changes through its own logger; wire it to the same output.
-    configure_kanta_logging()

paskia/db/migrations.py

@@ -7,35 +7,30 @@ Each migration should be idempotent and only run when needed.
 
 import base64
 
+from kanta import Kanta
+
 from paskia.util.crypto import secret_key
 
 
-class MigrationCtx:
-    """Context passed to each migration function."""
-
-    def __init__(self, rp_id: str):
-        self.rp_id = rp_id
-
-
-def migrate_v1(d: dict, ctx: MigrationCtx) -> None:
+def migrate_v1(d: dict) -> None:
     """Remove Org.created_at fields."""
     for org_data in d["orgs"].values():
         org_data.pop("created_at", None)
 
 
-def migrate_v2(d: dict, ctx: MigrationCtx) -> None:
+def migrate_v2(d: dict, kanta: Kanta) -> None:
     """Add config field if missing."""
     if "config" not in d:
-        d["config"] = {"rp_id": ctx.rp_id}
+        d["config"] = {"rp_id": kanta.ctx.rp_id}
 
 
-def migrate_v3(d: dict, ctx: MigrationCtx) -> None:
+def migrate_v3(d: dict) -> None:
     """Ensure all users have visits field."""
     for user_data in d["users"].values():
         user_data.setdefault("visits", 0)
 
 
-def migrate_v4(d: dict, ctx: MigrationCtx) -> None:
+def migrate_v4(d: dict) -> None:
     """OpenID Connect support and hardened session keys."""
     # Session keys changed to hashes, drop old sessions
     d["sessions"] = {}
@@ -46,7 +41,7 @@ def migrate_v4(d: dict, ctx: MigrationCtx) -> None:
     }
 
 
-def migrate_v5(d: dict, ctx: MigrationCtx) -> None:
+def migrate_v5(d: dict) -> None:
     """Convert config.listen from str to list[str] if needed."""
     listen = d["config"].get("listen")
     if listen and isinstance(listen, str):

paskia/db/operations.py

@@ -12,7 +12,6 @@ from datetime import UTC, datetime, timedelta
 from uuid import UUID
 
 import uuid7
-from kanta import Kanta
 
 from paskia import oidc_notify
 from paskia.config import SESSION_LIFETIME
@@ -39,7 +38,26 @@ _UNSET = object()
 
 # Global database instance (empty until init() loads data)
 _db = DB(config=Config(rp_id="uninitialized.invalid"))
-_store: Kanta[DB] | None = None
+
+
+def _store():
+    """Return active Kanta instance for the current DB object."""
+    store = _db._store
+    if store is None:
+        raise RuntimeError("Kanta store is not initialized")
+    return store
+
+
+def _transaction(
+    action: str,
+    ctx: SessionContext | None = None,
+    *,
+    user: str | None = None,
+    mtime: bool | datetime = True,
+):
+    """Create a Kanta transaction with minimal metadata mapping."""
+    user_id = str(ctx.user.uuid) if ctx else user
+    return _store().transaction(action, user=user_id, mtime=mtime)
 
 
 def is_username_taken(username: str, exclude_uuid: UUID | None = None) -> bool:
@@ -60,7 +78,7 @@ def is_username_taken(username: str, exclude_uuid: UUID | None = None) -> bool:
 
 def update_config(config: Config) -> None:
     """Update the stored configuration."""
-    with _db.transaction("update_config"):
+    with _transaction("update_config"):
         _db.config = config
 
 
@@ -68,7 +86,7 @@ def create_permission(perm: Permission, *, ctx: SessionContext | None = None) ->
     """Create a new permission."""
     if perm.uuid in _db.permissions:
         raise ValueError(f"Permission {perm.uuid} already exists")
-    with _db.transaction("admin:create_permission", ctx):
+    with _transaction("admin:create_permission", ctx):
         perm.store()
 
 
@@ -86,7 +104,7 @@ def update_permission(
     """
     if uuid not in _db.permissions:
         raise ValueError(f"Permission {uuid} not found")
-    with _db.transaction("admin:update_permission", ctx):
+    with _transaction("admin:update_permission", ctx):
         _db.permissions[uuid].scope = scope
         _db.permissions[uuid].display_name = display_name
         _db.permissions[uuid].domain = domain
@@ -96,7 +114,7 @@ def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
     """Delete a permission and remove it from all roles."""
     if uuid not in _db.permissions:
         raise ValueError(f"Permission {uuid} not found")
-    with _db.transaction("admin:delete_permission", ctx):
+    with _transaction("admin:delete_permission", ctx):
         _db.permissions[uuid].delete()
 
 
@@ -108,7 +126,7 @@ def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
     if org.uuid in _db.orgs:
         raise ValueError(f"Organization {org.uuid} already exists")
     now = datetime.now(UTC)
-    with _db.transaction("admin:create_org", ctx):
+    with _transaction("admin:create_org", ctx):
         new_org = Org.create(display_name=org.display_name, created_at=now)
         new_org.uuid = org.uuid
         new_org.store()
@@ -140,7 +158,7 @@ def update_org_name(
     """Update organization display name."""
     if uuid not in _db.orgs:
         raise ValueError(f"Organization {uuid} not found")
-    with _db.transaction("admin:update_org_name", ctx):
+    with _transaction("admin:update_org_name", ctx):
         _db.orgs[uuid].display_name = display_name
 
 
@@ -148,7 +166,7 @@ def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
     """Delete organization and all its roles/users."""
     if uuid not in _db.orgs:
         raise ValueError(f"Organization {uuid} not found")
-    with _db.transaction("admin:delete_org", ctx):
+    with _transaction("admin:delete_org", ctx):
         _db.orgs[uuid].delete()
 
 
@@ -165,7 +183,7 @@ def add_permission_to_org(
     if permission_uuid not in _db.permissions:
         raise ValueError(f"Permission {permission_uuid} not found")
 
-    with _db.transaction("admin:add_permission_to_org", ctx):
+    with _transaction("admin:add_permission_to_org", ctx):
         _db.permissions[permission_uuid].orgs[org_uuid] = True
 
 
@@ -182,7 +200,7 @@ def remove_permission_from_org(
     if permission_uuid not in _db.permissions:
         return  # Permission not found, silently return
 
-    with _db.transaction("admin:remove_permission_from_org", ctx):
+    with _transaction("admin:remove_permission_from_org", ctx):
         _db.permissions[permission_uuid].orgs.pop(org_uuid, None)
 
 
@@ -192,7 +210,7 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
         raise ValueError(f"Role {role.uuid} already exists")
     if role.org_uuid not in _db.orgs:
         raise ValueError(f"Organization {role.org_uuid} not found")
-    with _db.transaction("admin:create_role", ctx):
+    with _transaction("admin:create_role", ctx):
         role.store()
 
 
@@ -205,7 +223,7 @@ def update_role_name(
     """Update role display name."""
     if uuid not in _db.roles:
         raise ValueError(f"Role {uuid} not found")
-    with _db.transaction("admin:update_role_name", ctx):
+    with _transaction("admin:update_role_name", ctx):
         _db.roles[uuid].display_name = display_name
 
 
@@ -220,7 +238,7 @@ def add_permission_to_role(
         raise ValueError(f"Role {role_uuid} not found")
     if permission_uuid not in _db.permissions:
         raise ValueError(f"Permission {permission_uuid} not found")
-    with _db.transaction("admin:add_permission_to_role", ctx):
+    with _transaction("admin:add_permission_to_role", ctx):
         _db.roles[role_uuid].permissions[permission_uuid] = True
 
 
@@ -233,7 +251,7 @@ def remove_permission_from_role(
     """Remove permission from role by UUID."""
     if role_uuid not in _db.roles:
         raise ValueError(f"Role {role_uuid} not found")
-    with _db.transaction("admin:remove_permission_from_role", ctx):
+    with _transaction("admin:remove_permission_from_role", ctx):
         _db.roles[role_uuid].permissions.pop(permission_uuid, None)
 
 
@@ -245,7 +263,7 @@ def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
     role = _db.roles[uuid]
     if role.users:
         raise ValueError(f"Cannot delete role {uuid}: users still assigned")
-    with _db.transaction("admin:delete_role", ctx):
+    with _transaction("admin:delete_role", ctx):
         _db.roles[uuid].delete()
 
 
@@ -255,7 +273,7 @@ def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None:
         raise ValueError(f"User {new_user.uuid} already exists")
     if new_user.role_uuid not in _db.roles:
         raise ValueError(f"Role {new_user.role_uuid} not found")
-    with _db.transaction("admin:create_user", ctx):
+    with _transaction("admin:create_user", ctx):
         new_user.store()
 
 
@@ -282,7 +300,7 @@ def update_user_display_name(
     if not display_name:
         raise ValueError("Display name cannot be empty")
     user = _db.users[uuid]
-    with _db.transaction("update_user_display_name", ctx):
+    with _transaction("update_user_display_name", ctx):
         user.display_name = display_name
         # Auto-fill preferred_username if not already set
         if user.preferred_username is None:
@@ -356,7 +374,7 @@ def update_user_info(
         elif len(telephone) > 32:
             raise ValueError("telephone too long")
 
-    with _db.transaction("update_user_info", ctx):
+    with _transaction("update_user_info", ctx):
         if display_name is not _UNSET:
             user.display_name = display_name
         if theme is not _UNSET:
@@ -380,7 +398,7 @@ def update_user_role(
         raise ValueError(f"User {uuid} not found")
     if role_uuid not in _db.roles:
         raise ValueError(f"Role {role_uuid} not found")
-    with _db.transaction("admin:update_user_role", ctx):
+    with _transaction("admin:update_user_role", ctx):
         _db.users[uuid].role_uuid = role_uuid
 
 
@@ -388,7 +406,7 @@ def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
     """Delete user and their credentials/sessions."""
     if uuid not in _db.users:
         raise ValueError(f"User {uuid} not found")
-    with _db.transaction("admin:delete_user", ctx):
+    with _transaction("admin:delete_user", ctx):
         _db.users[uuid].delete()
 
 
@@ -398,7 +416,7 @@ def create_credential(cred: Credential, *, ctx: SessionContext | None = None) ->
         raise ValueError(f"Credential {cred.uuid} already exists")
     if cred.user_uuid not in _db.users:
         raise ValueError(f"User {cred.user_uuid} not found")
-    with _db.transaction("create_credential", ctx):
+    with _transaction("create_credential", ctx):
         cred.store()
 
 
@@ -412,7 +430,7 @@ def update_credential_sign_count(
     """Update credential sign count and last_used."""
     if uuid not in _db.credentials:
         raise ValueError(f"Credential {uuid} not found")
-    with _db.transaction("update_credential_sign_count", ctx):
+    with _transaction("update_credential_sign_count", ctx):
         _db.credentials[uuid].sign_count = sign_count
         if last_used:
             _db.credentials[uuid].last_used = last_used
@@ -434,7 +452,7 @@ def delete_credential(
     if user_uuid is not None:
         if cred.user_uuid != user_uuid:
             raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}")
-    with _db.transaction("delete_credential", ctx):
+    with _transaction("delete_credential", ctx):
         cred.delete()
 
 
@@ -450,7 +468,7 @@ def update_session(
     """Update session metadata."""
     if key not in _db.sessions:
         raise ValueError("Session not found")
-    with _db.transaction("update_session", ctx):
+    with _transaction("update_session", ctx):
         s = _db.sessions[key]
         if host is not None:
             s.host = host
@@ -480,7 +498,7 @@ def delete_session(
         raise ValueError("Session not found")
 
     oidc_notify.schedule_notifications([key])
-    with _db.transaction(action, ctx):
+    with _transaction(action, ctx):
         _db.sessions[key].delete()
 
 
@@ -499,7 +517,7 @@ def delete_sessions_for_user(
 
     keys = [s.key for s in user.sessions]
     oidc_notify.schedule_notifications(keys)
-    with _db.transaction("admin:delete_sessions_for_user", ctx):
+    with _transaction("admin:delete_sessions_for_user", ctx):
         for sess in user.sessions:
             sess.delete()
 
@@ -530,7 +548,7 @@ def create_reset_token(
     )
     if token.key in _db.reset_tokens:
         raise ValueError("Reset token already exists")
-    with _db.transaction("create_reset_token", ctx, user=user):
+    with _transaction("create_reset_token", ctx, user=user):
         token.store()
     return passphrase
 
@@ -539,7 +557,7 @@ def delete_reset_token(key: bytes, *, ctx: SessionContext | None = None) -> None
     """Delete a reset token."""
     if key not in _db.reset_tokens:
         raise ValueError("Reset token not found")
-    with _db.transaction("delete_reset_token", ctx):
+    with _transaction("delete_reset_token", ctx):
         _db.reset_tokens[key].delete()
 
 
@@ -588,7 +606,7 @@ def login(
         validated=now,
     )
     user_str = str(user_uuid)
-    with _db.transaction("login", user=user_str):
+    with _transaction("login", user=user_str):
         session.store(now)
         # Update credential
         _db.credentials[credential_uuid].sign_count = sign_count
@@ -615,7 +633,7 @@ def oidc_login(
     """
     now = datetime.now(UTC)
     user_str = str(session.user_uuid)
-    with _db.transaction("oidc_login", user=user_str):
+    with _transaction("oidc_login", user=user_str):
         session.store(now)
         # Update credential
         _db.credentials[credential_uuid].sign_count = sign_count
@@ -661,7 +679,7 @@ def create_credential_session(
         validated=now,
     )
     user_str = str(user_uuid)
-    with _db.transaction("create_credential_session", user=user_str):
+    with _transaction("create_credential_session", user=user_str):
         # Update display name if provided
         if display_name:
             _db.users[user_uuid].display_name = display_name
@@ -694,7 +712,7 @@ def create_oid_client(client: Client, *, ctx: SessionContext | None = None) -> N
     """Create a new OIDC client."""
     if client.uuid in _db.oidc.clients:
         raise ValueError(f"OIDC client {client.uuid} already exists")
-    with _db.transaction("admin:create_oid_client", ctx):
+    with _transaction("admin:create_oid_client", ctx):
         _db.oidc.clients[client.uuid] = client
 
 
@@ -735,7 +753,7 @@ def update_oid_client(
         else client.backchannel_logout_uri
     )
 
-    with _db.transaction("admin:update_oid_client", ctx):
+    with _transaction("admin:update_oid_client", ctx):
         # Create updated client with new values
         updated_client = Client(
             client_secret_hash=secret_hash
@@ -761,7 +779,7 @@ def reset_oid_client_secret(
     if client_uuid not in _db.oidc.clients:
         raise ValueError(f"OIDC client {client_uuid} not found")
     client = _db.oidc.clients[client_uuid]
-    with _db.transaction("admin:reset_oid_client_secret", ctx):
+    with _transaction("admin:reset_oid_client_secret", ctx):
         updated = Client(
             client_secret_hash=new_secret_hash,
             name=client.name,
@@ -776,5 +794,5 @@ def delete_oid_client(client_uuid: UUID, *, ctx: SessionContext | None = None) -
     """Delete an OIDC client."""
     if client_uuid not in _db.oidc.clients:
         raise ValueError(f"OIDC client {client_uuid} not found")
-    with _db.transaction("admin:delete_oid_client", ctx):
+    with _transaction("admin:delete_oid_client", ctx):
         del _db.oidc.clients[client_uuid]

paskia/db/structs.py

@@ -3,13 +3,13 @@ from __future__ import annotations
 import hashlib
 import secrets
 from datetime import UTC, datetime
+from typing import Any
 from uuid import UUID
 
 import msgspec
 import uuid7
 
 from paskia import db
-from paskia.db.logging import UuidResolver
 from paskia.util import passphrase as passphrase_util
 from paskia.util.crypto import hash_secret
 
@@ -631,8 +631,8 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
     oidc: OIDC = msgspec.field(default_factory=lambda: OIDC())
 
     def __post_init__(self):
-        # Store reference for persistence (not serialized)
-        self._store = None
+        # Optional store reference for non-global DB instances (e.g. tests).
+        self._store: Any | None = None
         # Set the key fields on all stored objects
         for uuid, perm in self.permissions.items():
             perm.uuid = uuid
@@ -652,23 +652,6 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
         for uuid, client in self.oidc.clients.items():
             client.uuid = uuid
 
-    def transaction(self, action, ctx=None, *, user=None):
-        """Wrap writes in transaction. Delegates to Kanta."""
-        user_id = str(ctx.user.uuid) if ctx else user
-        user_display = None
-        if user_id:
-            try:
-                user_uuid = UUID(user_id)
-                if user_uuid in self.users:
-                    user_display = self.users[user_uuid].display_name
-            except (ValueError, KeyError):
-                user_display = user_id
-        previous_state = msgspec.to_builtins(self)
-        resolver = UuidResolver(self, previous_state).resolve
-        return self._store.transaction(
-            action, user=user_id, user_display=user_display, resolver=resolver
-        )
-
     def session_ctx(
         self, session_secret: str, host: str | None = None
     ) -> SessionContext | None:

paskia/fastapi/__init__.py

@@ -1,3 +0,0 @@
-from paskia.fastapi.mainapp import app
-
-__all__ = ["app"]

paskia/fastapi/admin/permissions.py

@@ -28,7 +28,7 @@ def _validate_permission_domain(domain: str | None) -> None:
     except ValueError:
         pass
 
-    rp_id = passkey.instance.rp_id
+    rp_id = passkey.rp_id
     if domain == rp_id or domain.endswith(f".{rp_id}"):
         return
     raise ValueError(

paskia/fastapi/admin/server_config.py

@@ -22,7 +22,7 @@ async def admin_get_server_config(
 ):
     """Get current server configuration (master admin only)."""
     await authz.verify(auth, ["auth:admin"], host=request.headers.get("host"))
-    pk = passkey.instance
+    pk = passkey
     config = db.data().config
     return {
         "rp_name": pk.rp_name,
@@ -46,7 +46,7 @@ async def admin_update_server_config(
         auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m"
     )
     config = db.data().config
-    pk = passkey.instance
+    pk = passkey
 
     rp_name = payload.get("rp_name", "").strip() or None
     auth_host = payload.get("auth_host", "").strip() or None

paskia/fastapi/api.py

@@ -256,7 +256,7 @@ async def forward_authentication(
 
 @app.get("/settings")
 async def get_settings():
-    pk = global_passkey.instance
+    pk = global_passkey
     base_path = hostutil.ui_base_path()
     return MsgspecResponse(
         ApiSettings(

paskia/fastapi/mainapp.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import os
 from contextlib import asynccontextmanager
@@ -6,13 +7,12 @@ from pathlib import Path
 import msgspec
 from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse, RedirectResponse
+from kanta.logging import configure_logging as configure_kanta_logging
 
-from paskia import authcode, db, globals
-from paskia.__main__ import DEVMODE
+from paskia import authcode, db, remoteauth
 from paskia.bootstrap import bootstrap_if_needed
-from paskia.db import start_background, stop_background
-from paskia.db.background import flush
-from paskia.db.logging import configure_db_logging
+from paskia.db.background import start_background, stop_background
+from paskia.db.lifecycle import kanta
 from paskia.fastapi import admin, api, auth_host, oid, ws
 from paskia.fastapi.admin.adminapp import adminapp
 
@@ -21,11 +21,12 @@ from paskia.fastapi.front import frontend
 from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging
 from paskia.fastapi.session import AUTH_COOKIE
 from paskia.util import hostutil, passphrase, vitedev
+from paskia.util.constants import DEVMODE
 from paskia.util.runtime import RuntimeConfig
 
 # Configure custom logging
 configure_access_logging()
-configure_db_logging()
+configure_kanta_logging()
 
 _access_logger = logging.getLogger("paskia.access")
 
@@ -43,34 +44,33 @@ async def lifespan(app: FastAPI):  # pragma: no cover - startup path
     """
     runtime = msgspec.json.decode(os.environ["PASKIA_CONFIG"], type=RuntimeConfig)
 
-    try:
-        await globals.init(
-            rp_id=runtime.config.rp_id,
-            rp_name=runtime.config.rp_name,
-            origins=runtime.config.origins,
-            bootstrap=False,
-        )
-    except ValueError as e:
-        logging.error(f"⚠️ {e}")
-        # Re-raise to fail fast
-        raise
+    await asyncio.to_thread(
+        Path(kanta.filename).parent.mkdir, parents=True, exist_ok=True
+    )
+    async with kanta:
+        try:
+            await remoteauth.init()
+            await authcode.start()
+        except ValueError as e:
+            logging.error(f"⚠️ {e}")
+            # Re-raise to fail fast
+            raise
 
-    # Bootstrap and persist config now that the full DB is loaded
-    await bootstrap_if_needed(config=runtime.config)
-    if runtime.save:
-        db.update_config(runtime.config)
-    await flush()
+        # Bootstrap and persist config now that the full DB is loaded
+        await bootstrap_if_needed(config=runtime.config)
+        if runtime.save:
+            db.update_config(runtime.config)
 
-    # Restore uvicorn info logging (suppressed during startup in dev mode)
-    # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages
-    if app.debug:
-        logging.getLogger("uvicorn").setLevel(logging.INFO)
-    logging.getLogger("uvicorn.error").setLevel(logging.WARNING)
-    await frontend.load()
-    await start_background()
-    yield
-    await stop_background()
-    await authcode.stop()
+        # Restore uvicorn info logging (suppressed during startup in dev mode)
+        # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages
+        if app.debug:
+            logging.getLogger("uvicorn").setLevel(logging.INFO)
+        logging.getLogger("uvicorn.error").setLevel(logging.WARNING)
+        await frontend.load()
+        await start_background()
+        yield
+        await stop_background()
+        await authcode.stop()
 
 
 app = FastAPI(

paskia/fastapi/ws.py

@@ -58,7 +58,7 @@ async def websocket_register_add(
     if reset is not None:
         if not passphrase.is_well_formed(reset):
             raise ValueError(
-                f"The reset link for {passkey.instance.rp_name} is invalid or has expired"
+                f"The reset link for {passkey.rp_name} is invalid or has expired"
             )
         s = get_reset(reset)
         user_uuid = s.user_uuid

paskia/fastapi/wschat.py

@@ -23,14 +23,14 @@ async def register_chat(
     credential_ids: list[bytes] | None = None,
 ):
     """Run WebAuthn registration flow and return the verified credential."""
-    options, challenge = passkey.instance.reg_generate_options(
+    options, challenge = passkey.reg_generate_options(
         user_id=user_uuid,
         user_name=user_name,
         credential_ids=credential_ids,
     )
     await ws.send_json({"optionsJSON": options})
     response = await ws.receive_json()
-    return passkey.instance.reg_verify(response, challenge, user_uuid, origin=origin)
+    return passkey.reg_verify(response, challenge, user_uuid, origin=origin)
 
 
 async def authenticate_chat(
@@ -43,11 +43,9 @@ async def authenticate_chat(
         tuple of (credential, new_sign_count) where new_sign_count comes from WebAuthn verification
     """
     origin = validate_origin(ws)
-    options, challenge = passkey.instance.auth_generate_options(
-        credential_ids=credential_ids
-    )
+    options, challenge = passkey.auth_generate_options(credential_ids=credential_ids)
     await ws.send_json({"optionsJSON": options})
-    authcred = passkey.instance.auth_parse(await ws.receive_json())
+    authcred = passkey.auth_parse(await ws.receive_json())
 
     cred = next(
         (
@@ -58,11 +56,9 @@ async def authenticate_chat(
         None,
     )
     if not cred:
-        raise ValueError(
-            f"This passkey is no longer registered with {passkey.instance.rp_name}"
-        )
+        raise ValueError(f"This passkey is no longer registered with {passkey.rp_name}")
 
-    verification = passkey.instance.auth_verify(authcred, challenge, cred, origin)
+    verification = passkey.auth_verify(authcred, challenge, cred, origin)
     return cred, verification.new_sign_count
 
 
@@ -94,7 +90,7 @@ async def authenticate_and_login(
     if not normalized_host:
         raise ValueError("Host required for session creation")
     hostname = normalized_host.split(":")[0]
-    rp_id = passkey.instance.rp_id
+    rp_id = passkey.rp_id
     if not (hostname == rp_id or hostname.endswith(f".{rp_id}")):
         raise ValueError(f"Host must be the same as or a subdomain of {rp_id}")
     metadata = infodict(ws, "auth")

paskia/fastapi/wsutil.py

@@ -96,4 +96,4 @@ def validate_origin(ws: WebSocket) -> str:
     origin = ws.headers.get("origin")
     if not origin:
         raise ValueError("Origin header is required for WebSocket connections")
-    return passkey.instance.validate_origin(origin)
+    return passkey.validate_origin(origin)

paskia/globals.py

@@ -1,71 +1,20 @@
-from typing import Generic, TypeVar
+"""Global Passkey instance configured from PASKIA_CONFIG.
+
+The Passkey instance is created at import time using the runtime configuration
+passed via the ``PASKIA_CONFIG`` environment variable.  Other runtime setup
+(remote auth, auth codes, bootstrap checks) is performed explicitly by the
+FastAPI lifespan once the database is open.
+"""
 
-from paskia import authcode, db, remoteauth
-from paskia.bootstrap import bootstrap_if_needed
 from paskia.sansio import Passkey
+from paskia.util import runtime
 
-T = TypeVar("T")
+runtime = runtime.config()
+if runtime is None:
+    raise RuntimeError("PASKIA_CONFIG must be defined before importing paskia.globals")
 
-
-class Manager(Generic[T]):
-    """Generic manager for global instances."""
-
-    def __init__(self, name: str):
-        self._instance: T | None = None
-        self._name = name
-
-    @property
-    def instance(self) -> T:
-        if self._instance is None:
-            raise RuntimeError(
-                f"{self._name} not initialized. Call globals.init() first."
-            )
-        return self._instance
-
-    @instance.setter
-    def instance(self, instance: T) -> None:
-        self._instance = instance
-
-
-async def init(
-    rp_id: str = "localhost",
-    rp_name: str | None = None,
-    origins: list[str] | None = None,
-    *,
-    bootstrap: bool = True,
-) -> None:
-    """Initialize global passkey + database.
-
-    If bootstrap=True (default) the system bootstrap_if_needed() will be invoked.
-    In FastAPI lifespan we call with bootstrap=False to avoid duplicate bootstrapping
-    since the CLI performs it once before servers start.
-
-    Database configuration:
-        Set PASKIA_DB environment variable to specify the JSONL database file path.
-        Default: {rp_id}.paskiadb
-    """
-
-    # Initialize passkey instance with provided parameters
-    passkey.instance = Passkey(
-        rp_id=rp_id,
-        rp_name=rp_name or rp_id,
-        origins=origins,
-    )
-
-    # Initialize database
-    await db.init(rp_id=rp_id)
-
-    # Initialize remote auth manager
-    await remoteauth.init()
-
-    # Initialize auth code manager
-    await authcode.start()
-
-    if bootstrap:
-        # Bootstrap system if needed
-
-        await bootstrap_if_needed()
-
-
-# Global instances
-passkey = Manager[Passkey]("Passkey")
+passkey = Passkey(
+    rp_id=runtime.config.rp_id,
+    rp_name=runtime.config.rp_name,
+    origins=runtime.config.origins,
+)

paskia/oidc_notify.py

@@ -13,7 +13,7 @@ import httpx
 
 from paskia import db
 from paskia.util import oidjwt
-from paskia.util.runtime import _load_config
+from paskia.util.runtime import config as runtime_config
 
 _logger = logging.getLogger(__name__)
 
@@ -23,8 +23,8 @@ _TIMEOUT = httpx.Timeout(10.0, connect=5.0)
 
 def _issuer() -> str:
     """Derive issuer URL from config (same base as discovery document)."""
-    cfg = _load_config()
-    return cfg.get("site_url", "https://localhost")
+    cfg = runtime_config()
+    return cfg.site_url if cfg else "https://localhost"
 
 
 def _collect_oidc_sessions(

paskia/util/constants.py

@@ -0,0 +1,6 @@
+"""Small, dependency-free constants shared by CLI and server modules."""
+
+import os
+
+DEFAULT_PORT = 4401
+DEVMODE = os.getenv("PASKIA_DEV") == "1"

paskia/util/hostutil.py

@@ -2,11 +2,12 @@
 
 from urllib.parse import urlparse, urlsplit
 
-from paskia.util.runtime import _load_config
+from paskia.util.runtime import clear_config_cache
+from paskia.util.runtime import config as runtime_config
 
 
 def _cfg():
-    return _load_config()
+    return runtime_config()
 
 
 def is_root_mode() -> bool:
@@ -105,7 +106,7 @@ def normalize_auth_host_and_origins(
 
 
 def reload_config() -> None:
-    _load_config.cache_clear()
+    clear_config_cache()
 
 
 def normalize_host(raw_host: str | None) -> str | None:

paskia/util/oidjwt.py

@@ -29,11 +29,14 @@ def _load_or_generate_key() -> None:
     global _private_key, _public_key, _kid
 
     data = db.data()
+    store = data._store
+    if store is None:
+        raise RuntimeError("Kanta store is not initialized")
     if data.oidc.key is not None:
         _private_key = public_key_from_secret(data.oidc.key)
     else:
         raw_key = secret_key()
-        with data.transaction("oidc_key"):
+        with store.transaction("oidc_key"):
             data.oidc.key = raw_key
         _private_key = public_key_from_secret(raw_key)
 

paskia/util/runtime.py

@@ -31,21 +31,37 @@ def _load_config() -> "RuntimeConfig | None":
     return msgspec.json.decode(config_json.encode(), type=RuntimeConfig)
 
 
+def config() -> "RuntimeConfig | None":
+    """Return cached runtime config loaded from PASKIA_CONFIG."""
+    return _load_config()
+
+
+def clear_config_cache() -> None:
+    """Clear cached runtime config; next config() call reloads from env."""
+    _load_config.cache_clear()
+
+
 def update_runtime_config(new_config: Config) -> None:
     """Update the runtime configuration with a new Config and refresh the cache."""
-    current_runtime = _load_config()
+    current_runtime = config()
     if not current_runtime:
         return  # No runtime config to update
 
     # Recompute site_url and site_path based on new config
-    site_path = "/" if new_config.auth_host else "/auth/"
+    old_auth_host = current_runtime.config.auth_host
     if new_config.auth_host:
-        site_url = new_config.auth_host
-    elif new_config.origins:
-        site_url = new_config.origins[0]
+        site_url, site_path = new_config.auth_host, "/"
     else:
-        # Keep current site_url if no auth_host and no origins
-        site_url = current_runtime.site_url
+        site_path = "/auth/"
+        # Never derive site_url from a just-removed auth host
+        origins = [o for o in (new_config.origins or []) if o != old_auth_host]
+        if origins:
+            site_url = origins[0]
+        elif current_runtime.site_url != old_auth_host:
+            # Keep current site_url if it wasn't derived from the removed auth host
+            site_url = current_runtime.site_url
+        else:
+            site_url = f"https://{new_config.rp_id}"
 
     new_runtime = RuntimeConfig(
         config=new_config,
@@ -56,4 +72,4 @@ def update_runtime_config(new_config: Config) -> None:
     os.environ["PASKIA_CONFIG"] = msgspec.json.encode(new_runtime).decode()
 
     # Clear the cache so next access loads the updated config
-    _load_config.cache_clear()
+    clear_config_cache()

paskia/util/startupbox.py

@@ -10,6 +10,7 @@ from typing import TYPE_CHECKING
 from fastapi_vue.hostutil import parse_endpoints
 
 from paskia._version import __version__
+from paskia.util.constants import DEFAULT_PORT, DEVMODE
 from paskia.util.hostutil import format_endpoint
 
 if TYPE_CHECKING:
@@ -73,16 +74,13 @@ def print_startup_config(runtime: RuntimeConfig) -> None:
     if runtime.config.auth_host:
         lines.append(line(f"Auth Host:      {runtime.config.auth_host}"))
 
-    from paskia.__main__ import DEFAULT_PORT as P  # noqa: PLC0415 - circular
-    from paskia.__main__ import DEVMODE  # noqa: PLC0415 - circular
-
     # Show frontend URL if in dev mode
     if DEVMODE:
         lines.append(line(f"Dev Frontend:   {os.environ.get('PASKIA_VITE_URL')}"))
 
     # Format listen endpoints (dev mode only uses the first endpoint)
 
-    endpoints = list(parse_endpoints(runtime.config.listen, P))
+    endpoints = list(parse_endpoints(runtime.config.listen, DEFAULT_PORT))
     if DEVMODE:
         endpoints = endpoints[:1]  # server.run reload=True uses only one
     parts = [format_endpoint(ep) for ep in endpoints]

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
     "msgspec>=0.20.0",
     "fastapi-vue>=1.1.0",
     "ua-parser[regex]>=1.0.1",
-    "kanta>=0.1.1",
+    "kanta>=0.7.0",
 ]
 [dependency-groups]
 dev = [

tests/conftest.py

@@ -12,6 +12,7 @@ in the database to test authenticated endpoints.
 from __future__ import annotations
 
 import asyncio
+import json
 import os
 import secrets
 import tempfile
@@ -24,6 +25,20 @@ import pytest
 import pytest_asyncio
 from kanta import Kanta
 
+# Keep runtime initialization invariant aligned with production:
+# db.lifecycle requires PASKIA_CONFIG at import time.
+os.environ.setdefault(
+    "PASKIA_CONFIG",
+    json.dumps(
+        {
+            "config": {"rp_id": "localhost", "rp_name": "localhost"},
+            "site_url": "http://localhost:4401",
+            "site_path": "/auth/",
+            "save": False,
+        }
+    ),
+)
+
 import paskia.db.operations as ops_db
 from paskia import globals as paskia_globals
 from paskia.authsession import reset_expires
@@ -34,13 +49,12 @@ from paskia.db import (
     Permission,
     Role,
     User,
-    bootstrap,
     create_credential,
     create_reset_token,
     create_role,
     create_user,
 )
-from paskia.db.migrations import MigrationCtx
+from paskia.db.bootstrap import bootstrap
 from paskia.db.operations import DB
 from paskia.db.structs import Session
 from paskia.fastapi.mainapp import app
@@ -61,7 +75,7 @@ def event_loop():
 async def test_db() -> AsyncGenerator[DB, None]:
     """Create a temporary JSONL database for testing using kanta.
 
-    Uses bootstrap() to properly initialize the database with:
+    Uses a kanta bootstrap callback to properly initialize the database with:
     - auth:admin and auth:org:admin permissions
     - A default organization with Administration role
     - An admin user with the Administration role
@@ -72,34 +86,46 @@ async def test_db() -> AsyncGenerator[DB, None]:
             f.name,
             db,
             migrations="paskia.db.migrations",
-            migration_ctx=MigrationCtx(rp_id="test.example.com"),
         )
+        kanta.ctx.rp_id = "test.example.com"
+
+        # Register bootstrap callback so kanta seeds the empty DB during open()
+        @kanta.bootstrap(action="bootstrap")
+        def bootstrap_test_db(data: DB) -> None:
+            bootstrap(
+                data,
+                org_name="Test Organization",
+                admin_name="Test Admin",
+            )
+
         await kanta.open()
-        ops_db._store = kanta
         ops_db._db = db
         ops_db._db._store = kanta
-        # Bootstrap creates the initial permissions, org, role, and admin user
-        bootstrap(
-            org_name="Test Organization",
-            admin_name="Test Admin",
-        )
         yield ops_db._db
         await kanta.close()
         ops_db._db = None
-        ops_db._store = None
 
 
 @pytest_asyncio.fixture(scope="function")
 async def passkey_instance() -> Passkey:
-    """Initialize a passkey instance for testing."""
+    """Override the module-level passkey instance for testing."""
     pk = Passkey(
         rp_id="localhost",
         rp_name="Test RP",
         origins=["http://localhost:4401"],
     )
-    paskia_globals.passkey._instance = pk
+    original = {
+        "rp_id": paskia_globals.passkey.rp_id,
+        "rp_name": paskia_globals.passkey.rp_name,
+        "allowed_origins": paskia_globals.passkey.allowed_origins,
+    }
+    paskia_globals.passkey.rp_id = pk.rp_id
+    paskia_globals.passkey.rp_name = pk.rp_name
+    paskia_globals.passkey.allowed_origins = pk.allowed_origins
     yield pk
-    paskia_globals.passkey._instance = None
+    paskia_globals.passkey.rp_id = original["rp_id"]
+    paskia_globals.passkey.rp_name = original["rp_name"]
+    paskia_globals.passkey.allowed_origins = original["allowed_origins"]
 
 
 @pytest_asyncio.fixture(scope="function")
@@ -286,7 +312,10 @@ def create_test_session(
     )
     if session.key in ops_db._db.sessions:
         raise ValueError("Session already exists")
-    with ops_db._db.transaction("create_test_session"):
+    store = ops_db._db._store
+    if store is None:
+        raise RuntimeError("Test DB store is not initialized")
+    with store.transaction("create_test_session"):
         session.store(now)
     return session.key, token
 

tests/test_admin.py

@@ -37,7 +37,10 @@ from paskia.db import (
     create_user,
 )
 from paskia.db.operations import DB
+from paskia.util import hostutil
 from paskia.util.crypto import hash_secret
+from paskia.util.runtime import clear_config_cache
+from paskia.util.runtime import config as runtime_config
 from tests.conftest import auth_headers, create_test_image_bytes, create_test_session
 
 # -------------------- Additional Fixtures --------------------
@@ -1789,3 +1792,115 @@ class TestOrgAdminAuthExceptions:
             headers={**auth_headers(regular_session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 403
+
+
+class TestServerConfig:
+    """Tests for GET/PATCH /auth/api/admin/server-config/ runtime updates."""
+
+    @pytest.fixture(scope="function")
+    def restore_runtime_config(self):
+        """Restore PASKIA_CONFIG env and cache after a test mutates runtime."""
+        original = os.environ["PASKIA_CONFIG"]
+        yield
+        os.environ["PASKIA_CONFIG"] = original
+        clear_config_cache()
+
+    async def _set_auth_host(self, client, session_token, test_user, test_credential):
+        """Configure an auth host via PATCH, as the admin UI would."""
+        r = await client.patch(
+            "/auth/api/admin/server-config/",
+            json={
+                "rp_name": "",
+                "auth_host": "auth.localhost",
+                "origins": ["auth.localhost", "localhost"],
+            },
+            headers={**auth_headers(session_token), "Host": "localhost:4401"},
+        )
+        assert r.status_code == 200, r.text
+        assert db.data().config.auth_host == "https://auth.localhost"
+        assert hostutil.dedicated_auth_host() == "auth.localhost"
+        assert hostutil.auth_site_url() == "https://auth.localhost/"
+        # Session for requests coming from the auth host (sessions are host-bound)
+        _, token = create_test_session(
+            test_user.uuid, test_credential.uuid, host="auth.localhost"
+        )
+        return {**auth_headers(token), "Host": "auth.localhost"}
+
+    @pytest.mark.asyncio
+    async def test_remove_auth_host_updates_runtime(
+        self,
+        client: httpx.AsyncClient,
+        session_token: str,
+        test_user,
+        test_credential,
+        restore_runtime_config,
+    ):
+        """Removing auth_host must clear it from runtime config and URLs."""
+        headers = await self._set_auth_host(
+            client, session_token, test_user, test_credential
+        )
+
+        # The dialog still lists the old auth host among origins, so it is sent back
+        r = await client.patch(
+            "/auth/api/admin/server-config/",
+            json={
+                "rp_name": "",
+                "auth_host": "",
+                "origins": ["auth.localhost", "localhost"],
+            },
+            headers=headers,
+        )
+        assert r.status_code == 200, r.text
+        assert db.data().config.auth_host is None
+
+        rt = runtime_config()
+        assert rt.config.auth_host is None
+        assert rt.site_path == "/auth/"
+        assert "auth.localhost" not in rt.site_url
+        assert hostutil.dedicated_auth_host() is None
+        assert "auth.localhost" not in hostutil.auth_site_url()
+
+        # GET and settings reflect the cleared state
+        r = await client.get(
+            "/auth/api/admin/server-config/",
+            headers={**auth_headers(session_token), "Host": "localhost:4401"},
+        )
+        assert r.json()["auth_host"] == ""
+        r = await client.get("/auth/api/settings")
+        assert r.json()["auth_host"] is None
+        assert r.json()["ui_base_path"] == "/auth/"
+
+        # Middleware no longer redirects to the removed auth host
+        r = await client.get(
+            "/auth/admin",
+            headers={**auth_headers(session_token), "Host": "localhost:4401"},
+            follow_redirects=False,
+        )
+        assert "auth.localhost" not in r.headers.get("location", "")
+
+    @pytest.mark.asyncio
+    async def test_remove_auth_host_without_origins_falls_back_to_rp_id(
+        self,
+        client: httpx.AsyncClient,
+        session_token: str,
+        test_user,
+        test_credential,
+        restore_runtime_config,
+    ):
+        """With no origins left, site_url must not keep the removed auth host."""
+        headers = await self._set_auth_host(
+            client, session_token, test_user, test_credential
+        )
+
+        r = await client.patch(
+            "/auth/api/admin/server-config/",
+            json={"rp_name": "", "auth_host": "", "origins": []},
+            headers=headers,
+        )
+        assert r.status_code == 200, r.text
+
+        rt = runtime_config()
+        assert rt.config.auth_host is None
+        assert rt.site_path == "/auth/"
+        assert "auth.localhost" not in rt.site_url
+        assert "auth.localhost" not in hostutil.auth_site_url()

tests/test_api.py

@@ -426,7 +426,10 @@ class TestOidcUserInfoEndpoint:
             redirect_uris=["https://client.example/callback"],
             client_secret="topsecret",
         )
-        with test_db.transaction("create_test_oidc_client"):
+        store = test_db._store
+        if store is None:
+            raise RuntimeError("Test DB store is not initialized")
+        with store.transaction("create_test_oidc_client"):
             test_db.oidc.clients[oidc_client.uuid] = oidc_client
 
         access_token = oidjwt.create_access_token(

tests/test_cli.py

@@ -0,0 +1,154 @@
+"""Tests for the CLI entry point in paskia/__main__.py."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+import pytest
+from kanta import Kanta
+
+from paskia.__main__ import main
+from paskia.db.structs import DB, Config
+from paskia.util.runtime import clear_config_cache
+from paskia.util.runtime import config as runtime_config
+
+
+@pytest.fixture
+def cli_run(monkeypatch):
+    """Run the CLI main() with the given args and return the RuntimeConfig."""
+
+    def _run(*args: str, db_root: str | None = None) -> Any:
+        env = os.environ.copy()
+        if db_root is not None:
+            env["PASKIA_DB"] = db_root
+        monkeypatch.setattr(os, "environ", env)
+
+        monkeypatch.setattr(sys, "argv", ["paskia", *args])
+        monkeypatch.setattr("fastapi_vue.server.run", lambda *_args, **_kw: None)
+        monkeypatch.setattr(
+            "paskia.util.startupbox.print_startup_config", lambda _rt: None
+        )
+        monkeypatch.setattr("logging.basicConfig", lambda **_kw: None)
+
+        clear_config_cache()
+        main()
+        runtime = runtime_config()
+        clear_config_cache()
+        return runtime
+
+    return _run
+
+
+async def _write_config(db_path: Path, config: Config) -> None:
+    """Write a Config into a JSONL database file using Kanta.
+
+    The initial root uses a different rp_id so the stored diff includes the
+    target rp_id (required because Config omits defaults when diffing).
+    """
+    kanta = Kanta(
+        str(db_path),
+        DB(config=Config(rp_id="uninitialized.invalid")),
+        migrations="paskia.db.migrations",
+    )
+    kanta.ctx.rp_id = config.rp_id
+    await kanta.open()
+    with kanta.transaction("test:write_config"):
+        kanta.data.config = config
+    await kanta.close()
+
+
+def write_config(db_path: Path, config: Config) -> None:
+    """Synchronous wrapper for _write_config."""
+    asyncio.run(_write_config(db_path, config))
+
+
+def test_cli_defaults(cli_run):
+    with tempfile.TemporaryDirectory() as tmp:
+        runtime = cli_run("--rp-id", "localhost", db_root=tmp)
+
+    assert runtime.config.rp_id == "localhost"
+    assert runtime.config.rp_name is None
+    assert runtime.config.auth_host is None
+    assert runtime.config.origins is None
+    assert runtime.site_url == "http://localhost:4401"
+    assert runtime.site_path == "/auth/"
+    assert runtime.save is False
+
+
+def test_cli_explicit_options(cli_run):
+    runtime = cli_run(
+        "--rp-id",
+        "example.com",
+        "--rp-name",
+        "Example Corp",
+        "--auth-host",
+        "auth.example.com",
+        "--origin",
+        "https://app.example.com",
+    )
+
+    assert runtime.config.rp_id == "example.com"
+    assert runtime.config.rp_name == "Example Corp"
+    assert runtime.config.auth_host == "https://auth.example.com"
+    assert runtime.config.origins == [
+        "https://auth.example.com",
+        "https://app.example.com",
+    ]
+    assert runtime.site_url == "https://auth.example.com"
+    assert runtime.site_path == "/"
+
+
+def test_cli_loads_stored_config(cli_run):
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = Path(tmp) / "main.db"
+        write_config(
+            db_path,
+            Config(
+                rp_id="example.com",
+                rp_name="Stored Name",
+                origins=["https://stored.example.com"],
+            ),
+        )
+        runtime = cli_run("--rp-id", "example.com", db_root=tmp)
+
+    assert runtime.config.rp_name == "Stored Name"
+    assert runtime.config.origins == ["https://stored.example.com"]
+    assert runtime.site_url == "https://stored.example.com"
+
+
+def test_cli_overrides_stored_config(cli_run):
+    with tempfile.TemporaryDirectory() as tmp:
+        db_path = Path(tmp) / "main.db"
+        write_config(db_path, Config(rp_id="example.com", rp_name="Stored Name"))
+        runtime = cli_run(
+            "--rp-id", "example.com", "--rp-name", "Overridden", db_root=tmp
+        )
+
+    assert runtime.config.rp_name == "Overridden"
+
+
+def test_cli_save_flag(cli_run):
+    runtime = cli_run("--save")
+    assert runtime.save is True
+
+
+def test_cli_invalid_auth_host(cli_run):
+    with pytest.raises(SystemExit):
+        cli_run("--rp-id", "example.com", "--auth-host", "notsub.example.org")
+
+
+def test_cli_help():
+    result = subprocess.run(
+        [sys.executable, "-m", "paskia", "--help"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0
+    assert "Paskia authentication server" in result.stdout