Avoid storing IP and User Agent on OpenID Connect token renewals; preserves the user's information from authentication.

paskia/__main__.py

@@ -1,16 +1,16 @@
 import argparse
-import json
 import logging
 import os
 from urllib.parse import urlparse
 
+import msgspec
 from fastapi_vue import server
 from fastapi_vue.hostutil import parse_endpoints
 
-from paskia.config import PaskiaConfig
 from paskia.db.jsonl import load_readonly
 from paskia.util import startupbox
 from paskia.util.hostutil import normalize_origin
+from paskia.util.runtime import RuntimeConfig
 
 DEFAULT_PORT = 4401
 DEVMODE = os.getenv("PASKIA_DEV") == "1"
@@ -51,7 +51,6 @@ def add_common_options(p: argparse.ArgumentParser) -> None:
         "--origin",
         action="append",
         dest="origins",
-        default=[],
         metavar="URL",
         help="Allowed origin URL(s). May be specified multiple times. If any are specified, only those origins are permitted for WebSocket authentication.",
     )
@@ -91,106 +90,66 @@ def main():
 
     args = parser.parse_args()
 
-    # Handle clearing options
+    # Load stored config (read-only, no writes, no global state)
-    if getattr(args, "auth_host", None) == "":
-        args.auth_host = None
-    if getattr(args, "rp_name", None) == "":
-        args.rp_name = None
-    if getattr(args, "listen", None) == "":
-        args.listen = None
-
-    # Read-only load to get stored config (no writes, no global state)
     db_path = os.environ.get("PASKIA_DB", f"{args.rp_id}.paskiadb")
-    stored_db = load_readonly(db_path, rp_id=args.rp_id)
+    config = load_readonly(db_path, rp_id=args.rp_id).config
-    stored_config = stored_db.config
 
-    # Apply defaults from stored config
+    # Override stored config with CLI args, or clear with empty string
-    if args.rp_name is None and stored_config.rp_name is not None:
+    if args.rp_name is not None:
-        args.rp_name = stored_config.rp_name
+        config.rp_name = args.rp_name or None
-    if args.origins is None and stored_config.origins is not None:
+    if args.auth_host is not None:
-        args.origins = stored_config.origins
+        config.auth_host = args.auth_host or None
-    if args.auth_host is None and stored_config.auth_host is not None:
+    if args.origins is not None:
-        args.auth_host = stored_config.auth_host
+        config.origins = None if args.origins == [""] else args.origins
-    if args.listen is None and stored_config.listen is not None:
+    if args.listen is not None:
-        args.listen = stored_config.listen
+        config.listen = None if args.listen == [""] else args.listen
-
-    # Parse first endpoint for config display and site_url
-    ep = next(iter(parse_endpoints(args.listen, DEFAULT_PORT)), {})
-    host, port, uds = ep.get("host"), ep.get("port"), ep.get("uds")
 
     # Process and normalize auth_host
-    if args.auth_host:
+    if config.auth_host:
-        if "://" not in args.auth_host:
+        if "://" not in config.auth_host:
-            args.auth_host = f"https://{args.auth_host}"
+            config.auth_host = f"https://{config.auth_host}"
-        args.auth_host = args.auth_host.rstrip("/")
+        config.auth_host = config.auth_host.rstrip("/")
-        validate_auth_host(args.auth_host, args.rp_id)
+        validate_auth_host(config.auth_host, config.rp_id)
-        if args.origins:
+        if config.origins:
-            args.origins.insert(0, args.auth_host)  # Ensure first in origins
+            config.origins.insert(0, config.auth_host)  # Ensure first in origins
 
-    # Normalize, strip trailing slashes, and deduplicate while preserving order
+    # Normalize and deduplicate while preserving order
-    origins = list({normalize_origin(o).rstrip("/"): ... for o in (args.origins)})
+    if config.origins:
+        config.origins = list({normalize_origin(o): ... for o in config.origins})
 
-    # Compute site_url and site_path for reset links
+    # Parse first endpoint for site_url fallback
-    # Priority: auth_host > first configured origin > PASKIA_VITE_URL (devserver) > http://localhost:port > https://rp_id
+    ep = next(iter(parse_endpoints(config.listen, DEFAULT_PORT)), {})
+    port = ep.get("port")
+
+    # Compute site_url and site_path
+    # Priority: auth_host > origins[0] > PASKIA_VITE_URL > http://localhost:port > https://rp_id
     site_path = "/auth/"
-    if args.auth_host:
+    if config.auth_host:
-        site_url = args.auth_host
+        site_url, site_path = config.auth_host, "/"
-        site_path = "/"
+    elif config.origins:
-    elif origins:
+        site_url = config.origins[0]
-        # Find localhost origin if rp_id is localhost, else use first origin
-        localhost_origin = (
-            next((o for o in origins if "://localhost" in o), None)
-            if args.rp_id == "localhost"
-            else None
-        )
-        site_url = localhost_origin or origins[0]
     elif vite_url := os.environ.get("PASKIA_VITE_URL"):
         site_url = vite_url.rstrip("/")  # Devserver
-    elif args.rp_id == "localhost" and port:
+    elif config.rp_id == "localhost" and port:
         site_url = f"http://localhost:{port}"  # Backend directly if we can
     else:
-        site_url = f"https://{args.rp_id}"  # Assume external reverse proxy
+        site_url = f"https://{config.rp_id}"  # Assume external reverse proxy
 
-    # Build runtime configuration
+    # Build runtime configuration for the server
-    config = PaskiaConfig(
+    runtime = RuntimeConfig(
-        rp_id=args.rp_id,
+        config=config,
-        rp_name=args.rp_name or None,
-        origins=origins or None,
-        auth_host=args.auth_host or None,
         site_url=site_url,
         site_path=site_path,
-        host=host,
+        save=args.save,
-        port=port,
-        uds=uds,
     )
+    startupbox.print_startup_config(runtime)
+    os.environ["PASKIA_CONFIG"] = msgspec.json.encode(runtime).decode()
 
-    # Export configuration via single JSON env variable for worker processes
+    # Run the server (spawns processes in dev mode)
-    # Include cli_config and save flag so lifespan can handle bootstrap/persistence
-    cli_config = {
-        "rp_id": args.rp_id,
-        "rp_name": args.rp_name,
-        "origins": args.origins,
-        "auth_host": args.auth_host,
-        "listen": args.listen,
-    }
-    config_json = {
-        "rp_id": config.rp_id,
-        "rp_name": config.rp_name,
-        "origins": config.origins,
-        "auth_host": config.auth_host,
-        "site_url": config.site_url,
-        "site_path": config.site_path,
-        "save": args.save,
-        "cli_config": cli_config,
-    }
-    os.environ["PASKIA_CONFIG"] = json.dumps(config_json)
-
-    startupbox.print_startup_config(config)
-
     dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {}
     server.run(
         "paskia.fastapi.mainapp:app",
-        listen=args.listen,
+        listen=config.listen,
         default_port=DEFAULT_PORT,
         log_level="warning",
         access_log=False,

paskia/authsession.py

@@ -23,6 +23,11 @@ if TYPE_CHECKING:
 EXPIRES = SESSION_LIFETIME
 
 
+def session_ctx(auth: str, host: str | None = None):
+    """Get session context with normalized host."""
+    return db.data().session_ctx(auth, hostutil.normalize_host(host))
+
+
 def expires() -> datetime:
     return datetime.now(UTC) + EXPIRES
 
@@ -42,7 +47,7 @@ def get_reset(token: str) -> "ResetToken":
 
 def delete_credential(credential_uuid: UUID, auth: str, host: str | None = None):
     """Delete a specific credential for the current user."""
-    ctx = db.data().session_ctx(auth, hostutil.normalize_host(host))
+    ctx = session_ctx(auth, host)
     if not ctx:
         raise ValueError("Session expired")
     db.delete_credential(credential_uuid, ctx.user.uuid)

paskia/config.py

@@ -1,4 +1,3 @@
-from dataclasses import dataclass
 from datetime import timedelta
 
 # Shared configuration constants for session management.
@@ -6,19 +5,3 @@ SESSION_LIFETIME = timedelta(hours=24)
 
 # Lifetime for reset links created by admins
 RESET_LIFETIME = timedelta(days=14)
-
-
-@dataclass
-class PaskiaConfig:
-    """Runtime configuration for the Paskia authentication server."""
-
-    rp_id: str
-    rp_name: str | None
-    origins: list[str] | None
-    auth_host: str | None
-    site_url: str  # Base URL without trailing path (e.g. https://example.com)
-    site_path: str  # Path to auth UI: "/" if auth_host, else "/auth/"
-    # Listen address (one of host:port or uds)
-    host: str | None = None
-    port: int | None = None
-    uds: str | None = None

paskia/db/background.py

@@ -21,7 +21,7 @@ _background_task: asyncio.Task | None = None
 
 async def flush() -> None:
     """Write all pending database changes to disk."""
-    store = _ops._store
+    store = _ops._db._store
     if store is None:
         _logger.warning("flush() called but _store is None")
         return
@@ -48,6 +48,10 @@ async def _background_loop():
                 cleanup_expired()
                 await flush()  # Flush cleanup changes
                 last_cleanup = now
+
+            # Conditionally write a snapshot to speed up future startups
+            if _ops._db._store is not None:
+                _ops._db._store.maybe_snapshot()
         except asyncio.CancelledError:
             # Final flush before exit
             await flush()
@@ -99,7 +103,7 @@ async def stop_background():
         except asyncio.CancelledError:
             pass
         _background_task = None
-    _ops._store.close()
+    _ops._db._store.close()
 
 
 # Aliases for backwards compatibility

paskia/db/jsonl.py

@@ -25,12 +25,56 @@ from paskia.db.migrations import (
     apply_all_migrations,
     apply_migrations_readonly,
 )
+from paskia.db.snapshot import SnapshotState
 from paskia.db.structs import DB, Config, SessionContext
 
 _logger = logging.getLogger(__name__)
 
-# Default database path
+
-DB_PATH_DEFAULT = "paskia.jsonl"
+class ReplayResult(msgspec.Struct, frozen=False):
+    """Return value of _replay_from_data"""
+
+    state: dict
+    v: int = 0
+    ts: datetime | None = None
+    snapts: datetime | None = None
+    changes: int = 0
+
+
+class DatabaseError(Exception):
+    """Exception raised for database loading errors."""
+
+    pass
+
+
+def _replay_from_data(data: bytes, db_path: str) -> ReplayResult:
+    """Replay database state from file data, using the last snapshot if available."""
+    resolved_path = str(Path(db_path).resolve())
+    result = ReplayResult(state={})
+
+    # Find and apply the last snapshot
+    snap, start_offset = SnapshotState.load(data)
+    if snap:
+        result.state = snap.state
+        result.v = snap.v
+        result.snapts = snap.ts
+
+    # Replay change records after the snapshot
+    lines = data[start_offset:].split(b"\n")
+    for line_num, raw in enumerate(lines, start=1):  # 1-based line numbering
+        line = raw.strip()
+        if not line:
+            continue
+        try:
+            change = msgspec.json.decode(line, type=ChangeRecord)
+        except msgspec.DecodeError as e:
+            raise DatabaseError(f"{resolved_path}:{line_num}: {e}")
+        result.state = jsondiff.patch(result.state, change.diff, marshal=True)
+        result.v = change.v
+        result.ts = change.ts
+        result.changes += 1
+
+    return result
 
 
 def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
@@ -43,25 +87,20 @@ def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
     if not path.exists():
         return DB(config=Config(rp_id=rp_id))
 
-    data_dict: dict = {}
-    version = 0
     try:
         with open(path, "rb") as f:
             content = f.read()
-        for line_num, line in enumerate(content.split(b"\n"), 1):
+        r = _replay_from_data(content, str(path.resolve()))
-            line = line.strip()
+        data_dict = r.state
-            if not line:
+        version = r.v
-                continue
-            try:
-                change = msgspec.json.decode(line)
-                data_dict = jsondiff.patch(data_dict, change["diff"], marshal=True)
-                version = change.get("v", 0)
-            except Exception as e:
-                raise ValueError(f"Error parsing line {line_num}: {e}")
     except OSError as e:
-        raise SystemExit(f"Failed to load database: {e}")
+        _logger.exception("Failed to load database")
-    except (ValueError, msgspec.DecodeError) as e:
+        raise SystemExit(f"{e}")
-        raise SystemExit(f"Failed to load database: {e}")
+    except (ValueError, msgspec.DecodeError, DatabaseError) as e:
+        raise SystemExit(f"{e}")
+    except Exception as e:
+        _logger.exception("Unexpected error loading database")
+        raise SystemExit(f"{e}")
 
     if not data_dict:
         return DB(config=Config(rp_id=rp_id))
@@ -74,45 +113,16 @@ def load_readonly(db_path: str, *, rp_id: str = "localhost") -> DB:
     return db
 
 
-class _ChangeRecord(msgspec.Struct, omit_defaults=True):
+class ChangeRecord(msgspec.Struct, omit_defaults=True, kw_only=True):
-    """A single change record in the JSONL file."""
+    ts: datetime = msgspec.field(default_factory=lambda: datetime.now(UTC))
-
-    ts: datetime
     a: str  # action - describes the operation (e.g., "migrate", "login", "create_user")
-    v: int  # schema version after this change
+    v: int = 0  # schema version after this change
     u: str | None = None  # user UUID who performed the action (None for system)
-    diff: dict = {}
+    diff: dict
-
-
-# msgspec encoder for change records
-_change_encoder = msgspec.json.Encoder()
 
 
 def compute_diff(previous: dict, current: dict) -> dict | None:
-    """Compute JSON diff between two states.
+    return jsondiff.diff(previous, current, marshal=True) or None
-
-    Args:
-        previous: Previous state (JSON-compatible dict)
-        current: Current state (JSON-compatible dict)
-
-    Returns:
-        The diff, or None if no changes
-    """
-    diff = jsondiff.diff(previous, current, marshal=True)
-    return diff if diff else None
-
-
-def create_change_record(
-    action: str, version: int, diff: dict, user: str | None = None
-) -> _ChangeRecord:
-    """Create a change record for persistence."""
-    return _ChangeRecord(
-        ts=datetime.now(UTC),
-        a=action,
-        v=version,
-        u=user,
-        diff=diff,
-    )
 
 
 # Actions that are allowed to create a new database file
@@ -122,18 +132,19 @@ _BOOTSTRAP_ACTIONS = frozenset({"bootstrap"})
 class JsonlStore:
     """JSONL persistence layer for a DB instance."""
 
-    def __init__(self, db: DB, db_path: str = DB_PATH_DEFAULT):
+    def __init__(self, db: DB, db_path: str):
         self.db: DB = db
         self.db_path = Path(db_path)
         self._file = LockedFile()
         self._flush_failed = False
-        self._previous_builtins: dict[str, Any] = {}
+        self._statedict: dict[str, Any] = {}
-        self._pending_changes: deque[_ChangeRecord] = deque()
+        self._pending_changes: deque[ChangeRecord] = deque()
         self._current_action: str = "system"
         self._current_user: str | None = None
         self._in_transaction: bool = False
         self._transaction_snapshot: dict[str, Any] | None = None
-        self._current_version: int = DBVER  # Schema version for new databases
+        self._v: int = DBVER  # Schema version for new databases
+        self._snapshot = SnapshotState()
 
     async def load(
         self, db_path: str | None = None, *, rp_id: str = "localhost"
@@ -148,56 +159,49 @@ class JsonlStore:
         # Open with exclusive write lock and read contents — single threadpool call
         content = await asyncio.to_thread(self._file.open_and_read, self.db_path)
 
-        # Replay change log to reconstruct state
+        # Replay change log to reconstruct state (snapshot-accelerated)
-        data_dict: dict = {}
         try:
-            for line_num, line in enumerate(content.split(b"\n"), 1):
+            r = _replay_from_data(content, str(self.db_path.resolve()))
-                line = line.strip()
+            statedict = r.state
-                if not line:
+            self._v = r.v
-                    continue
+            self._snapshot.ts = r.snapts
-                try:
+            self._snapshot.changes = r.changes
-                    change = msgspec.json.decode(line)
+        except (OSError, ValueError, msgspec.DecodeError, DatabaseError) as e:
-                    data_dict = jsondiff.patch(data_dict, change["diff"], marshal=True)
+            raise SystemExit(f"{e}")
-                    self._current_version = change.get("v", 0)
+        except Exception as e:
-                except Exception as e:
+            _logger.exception("Unexpected error loading database")
-                    raise ValueError(f"Error parsing line {line_num}: {e}")
+            raise SystemExit(f"{e}")
-        except OSError as e:
-            raise SystemExit(f"Failed to load database: {e}")
-        except (ValueError, msgspec.DecodeError) as e:
-            raise SystemExit(f"Failed to load database: {e}")
 
-        if not data_dict:
+        if not statedict:
             return
 
         # Set previous state for diffing (will be updated by _queue_change)
-        self._previous_builtins = copy.deepcopy(data_dict)
+        self._statedict = copy.deepcopy(statedict)
 
         # Callback to persist each migration
         async def persist_migration(
             action: str, new_version: int, current: dict
         ) -> None:
-            self._current_version = new_version
+            self._v = new_version
             self._queue_change(action, new_version, current)
 
         # Apply schema migrations one at a time
         await apply_all_migrations(
-            data_dict,
+            statedict,
-            self._current_version,
+            self._v,
             persist_migration,
             MigrationCtx(rp_id=rp_id),
         )
 
         # Decode to msgspec struct
         decoder = msgspec.json.Decoder(DB)
-        self.db = decoder.decode(msgspec.json.encode(data_dict))
+        self.db = decoder.decode(msgspec.json.encode(statedict))
         self.db._store = self
 
         # Normalize via msgspec round-trip (handles omit_defaults etc.)
         # This ensures _previous_builtins matches what msgspec would produce
         normalized_dict = msgspec.to_builtins(self.db)
-        await persist_migration(
+        await persist_migration("migrate:msgspec", self._v, normalized_dict)
-            "migrate:msgspec", self._current_version, normalized_dict
-        )
 
     def _queue_change(
         self, action: str, version: int, current: dict, user: str | None = None
@@ -210,10 +214,17 @@ class JsonlStore:
             current: The current state as a plain dict
             user: Optional user UUID who performed the action
         """
-        diff = compute_diff(self._previous_builtins, current)
+        diff = compute_diff(self._statedict, current)
         if not diff:
             return
-        self._pending_changes.append(create_change_record(action, version, diff, user))
+        self._pending_changes.append(
+            ChangeRecord(
+                a=action,
+                v=version,
+                u=user,
+                diff=diff,
+            )
+        )
 
         # Log the change with user display name if available
         user_display = None
@@ -225,8 +236,8 @@ class JsonlStore:
             except (ValueError, KeyError):
                 user_display = user
 
-        log_change(action, diff, user_display, self._previous_builtins, self.db)
+        log_change(action, diff, user_display, self._statedict, self.db)
-        self._previous_builtins = copy.deepcopy(current)
+        self._statedict = copy.deepcopy(current)
 
     @contextmanager
     def transaction(
@@ -248,13 +259,13 @@ class JsonlStore:
 
         # Check for out-of-transaction modifications
         current_state = msgspec.to_builtins(self.db)
-        if current_state != self._previous_builtins:
+        if current_state != self._statedict:
             # Allow bootstrap to create a new database from empty state
             is_bootstrap = action in _BOOTSTRAP_ACTIONS
-            if is_bootstrap and not self._previous_builtins:
+            if is_bootstrap and not self._statedict:
                 pass  # Expected: creating database from scratch
             else:
-                diff = compute_diff(self._previous_builtins, current_state)
+                diff = compute_diff(self._statedict, current_state)
                 diff_json = msgspec.json.encode(diff).decode()
                 _logger.critical(
                     "Database state modified outside of transaction! "
@@ -275,7 +286,7 @@ class JsonlStore:
             yield
             current = msgspec.to_builtins(self.db)
             self._queue_change(
-                self._current_action, self._current_version, current, self._current_user
+                self._current_action, self._v, current, self._current_user
             )
         except Exception:
             # Rollback on error: restore from snapshot
@@ -318,18 +329,23 @@ class JsonlStore:
         changes_to_write = list(self._pending_changes)
 
         try:
-            lines = [_change_encoder.encode(change) for change in changes_to_write]
+            lines = [msgspec.json.encode(change) for change in changes_to_write]
             if not lines:
                 self._pending_changes.clear()
                 return
 
             await asyncio.to_thread(self._file.write, b"\n".join(lines) + b"\n")
+            self._snapshot.record_lines(len(lines))
             self._pending_changes.clear()
         except OSError as e:
             _logger.error("Failed to flush database: %s", e)
             self._flush_failed = True
             os.kill(os.getpid(), signal.SIGTERM)
 
+    def maybe_snapshot(self) -> None:
+        """Write a snapshot if conditions are met."""
+        self._snapshot.maybe_write(self._file, self._v, self._statedict)
+
     def close(self) -> None:
         """Release the file lock and close the file."""
         self._file.close()

paskia/db/lifecycle.py

@@ -9,20 +9,23 @@ from datetime import UTC, datetime
 import paskia.db.operations as _ops
 from paskia import oidc_notify
 from paskia.authsession import EXPIRES
+from paskia.db.jsonl import JsonlStore
 
 _logger = logging.getLogger(__name__)
 
 
-async def init(rp_id: str = "localhost", *args, **kwargs):
+async def init(rp_id: str, *args, **kwargs):
     """Load database from JSONL file."""
-    if _ops._initialized:
+    if _ops._db._store:
         _logger.debug("Database already initialized, skipping reload")
         return
-    default_path = f"{rp_id}.paskiadb"
+    db_path = os.environ.get("PASKIA_DB", f"{rp_id}.paskiadb")
-    db_path = os.environ.get("PASKIA_DB", default_path)
+    store = JsonlStore(_ops._db, db_path)
-    await _ops._store.load(db_path, rp_id=rp_id)
+    await store.load(db_path, rp_id=rp_id)
-    _ops._db = _ops._store.db
+    _ops._db = store.db
-    _ops._initialized = True
+    _ops._db._store = store
+    # Request a snapshot after successful startup
+    store._snapshot.request_force()
 
 
 def cleanup_expired() -> int:

paskia/db/logging.py

@@ -35,9 +35,9 @@ _UNSAFE_CHARS = re.compile(
 
 # ANSI color codes (matching FastAPI logging style)
 _RESET = "\033[0m"
-_DIM = "\033[2m"
+_SEP = "\033[38;5;242m"  # Dark grey for separators (like host/timing in access log)
-_PATH_PREFIX = "\033[1;30m"  # Dark grey for path prefix (like host in access log)
+_PATH_PREFIX = "\033[38;5;242m"  # Dark grey for path prefix (like host in access log)
-_PATH_FINAL = "\033[0m"  # Default for final element (like path 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
@@ -317,7 +317,7 @@ def _format_change_lines(
     # 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"{_DIM}<hidden>{_RESET}"
+            return f"{_SEP}<hidden>{_RESET}"
         return _format_value(v, resolver=resolver)
 
     # Helper to format path with UUID replacement
@@ -342,12 +342,12 @@ def _format_change_lines(
             lines = []
             # First line: path with green final element and grey =
             if len(formatted_path) == 1:
-                lines.append(f"  {_ADD}{formatted_path[0]}{_RESET} {_DIM}={_RESET}")
+                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} {_DIM}={_RESET}"
+                    f"  {_PATH_PREFIX}{prefix}.{_RESET}{_ADD}{final}{_RESET} {_SEP}={_RESET}"
                 )
             # Child lines: indented key: value, with aligned values
             # Format keys (may contain UUIDs)
@@ -360,24 +360,24 @@ def _format_change_lines(
             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}{_DIM}:{_RESET}{padding} {v_str}")
+                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} {_DIM}={_RESET} {value_str}"
+                    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} {_DIM}={_RESET} {value_str}"
+                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} {_DIM}={_RESET} {value_str}"]
+    return [f"  {path_str} {_SEP}={_RESET} {value_str}"]
 
 
 def format_diff(

paskia/db/operations.py

@@ -15,9 +15,6 @@ import uuid7
 
 from paskia import oidc_notify
 from paskia.config import SESSION_LIFETIME
-from paskia.db.jsonl import (
-    JsonlStore,
-)
 from paskia.db.structs import (
     DB,
     Client,
@@ -41,9 +38,6 @@ _UNSET = object()
 
 # Global database instance (empty until init() loads data)
 _db = DB(config=Config(rp_id="uninitialized.invalid"))
-_store = JsonlStore(_db)
-_db._store = _store
-_initialized = False
 
 
 def is_username_taken(username: str, exclude_uuid: UUID | None = None) -> bool:

paskia/db/snapshot.py

@@ -0,0 +1,88 @@
+"""
+Snapshot handling for JSONL database persistence.
+"""
+
+import logging
+from datetime import UTC, datetime
+from typing import Any
+
+import msgspec
+
+_logger = logging.getLogger(__name__)
+
+LINEPREFIX = b"SNAPSHOT "
+MINDIFFS = 100
+
+
+class Snapshot(msgspec.Struct):
+    """Snapshot data structure for database persistence."""
+
+    ts: datetime
+    v: int
+    state: dict[str, Any]
+
+
+class SnapshotState:
+    """Tracks snapshot timing and line counts for a database file."""
+
+    def __init__(self) -> None:
+        self.ts: datetime | None = None
+        self.changes: int = 0
+        self._force_pending: bool = False
+
+    def request_force(self) -> None:
+        """Request a forced snapshot on the next maybe_write call."""
+        self._force_pending = True
+
+    def record_lines(self, count: int) -> None:
+        self.changes += count
+
+    def maybe_write(self, file, version: int, state: dict) -> None:
+        """Write a snapshot if conditions are met (enough changes, and Sunday UTC or forced)."""
+        if self.changes < MINDIFFS:
+            return
+        force = self._force_pending
+        now = datetime.now(UTC)
+        if not force and now.weekday() != 6:  # 6 = Sunday
+            return
+        sunday_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
+        if not force and self.ts is not None and self.ts >= sunday_midnight:
+            return
+        if not file.is_open:
+            return
+        try:
+            self._write(file, version, state, now)
+            self._force_pending = False
+        except Exception as exc:
+            _logger.error("snapshot: failed to write snapshot: %r", exc)
+
+    def _write(self, file, version: int, state: dict, now: datetime) -> None:
+        """Write a snapshot and update internal state."""
+        data = msgspec.json.encode(Snapshot(ts=now, v=version, state=state))
+        file.write(LINEPREFIX + data + b"\n")
+        self.changes = 0
+        self.ts = now
+
+    @staticmethod
+    def load(data: bytes) -> tuple[Snapshot | None, int]:
+        """Find and parse the last snapshot in file data.
+
+        Returns (snapshot, replay_offset) where replay_offset is the byte
+        position to start replaying change records from. If no valid snapshot
+        is found, returns (None, 0).
+        """
+        marker = b"\n" + LINEPREFIX
+        pos = data.rfind(marker)
+        if pos != -1:
+            pos += 1  # skip the newline
+        elif data.startswith(LINEPREFIX):
+            pos = 0
+        else:
+            return None, 0
+
+        end = data.find(b"\n", pos)
+        if end == -1:
+            raise ValueError("Incomplete snapshot line at end of file")
+
+        snap = msgspec.json.decode(data[pos + len(LINEPREFIX) : end], type=Snapshot)
+        return snap, end + 1

paskia/db/structs.py

@@ -9,7 +9,6 @@ import msgspec
 import uuid7
 
 from paskia import db
-from paskia.util import hostutil
 from paskia.util import passphrase as passphrase_util
 from paskia.util.crypto import hash_secret
 
@@ -601,14 +600,14 @@ class OIDC(msgspec.Struct, dict=True):
     key: bytes | None = None
 
 
-class Config(msgspec.Struct, frozen=True, dict=True, omit_defaults=True):
+class Config(msgspec.Struct, omit_defaults=True):
     """Stored configuration for the instance."""
 
     rp_id: str
     rp_name: str | None = None
-    origins: list[str] | None = None
     auth_host: str | None = None
-    listen: str | None = None
+    origins: list[str] | None = None
+    listen: list[str] | None = None
 
 
 # -------------------------------------------------------------------------
@@ -679,10 +678,8 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
         if s.client_uuid is not None:
             return None
 
-        # Normalize host for comparison (stored hosts are already normalized)
-        normalized_input = hostutil.normalize_host(host)
-
         # Validate host matches (sessions are always created with a host)
+        normalized_input = host
         if s.host != normalized_input:
             # Session bound to different host
             return None

paskia/fastapi/api.py

@@ -15,7 +15,7 @@ from fastapi.security import HTTPBearer
 
 from paskia import authcode, db
 from paskia._version import __version__
-from paskia.authsession import EXPIRES, get_reset
+from paskia.authsession import EXPIRES, get_reset, session_ctx
 from paskia.fastapi import authz, session, user
 from paskia.fastapi.response import MsgspecResponse
 from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME, get_client_ip
@@ -202,7 +202,7 @@ async def api_user_info(
             detail="Authentication required",
             mode="login",
         )
-    ctx = db.data().session_ctx(auth, request.headers.get("host"))
+    ctx = session_ctx(auth, request.headers.get("host"))
     if not ctx:
         raise authz.AuthException(
             status_code=401,
@@ -249,7 +249,7 @@ async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE):
     if not auth:
         return {"message": "Already logged out"}
     host = request.headers.get("host")
-    ctx = db.data().session_ctx(auth, host)
+    ctx = session_ctx(auth, host)
     if not ctx:
         return {"message": "Already logged out"}
     with suppress(Exception):
@@ -282,7 +282,7 @@ async def api_set_session(
     secret = a.session_key
 
     # Verify the session exists
-    ctx = db.data().session_ctx(secret, host)
+    ctx = session_ctx(secret, host)
     if not ctx:
         raise HTTPException(401, f"Session not found on {host}")
 

paskia/fastapi/logging.py

@@ -115,25 +115,16 @@ def format_access_log(
     client: str, status: int, method: str, host: str, path: str, duration_ms: float
 ) -> str:
     """Format access log line with colors and aligned fields."""
-    use_color = sys.stderr.isatty()
-
     # Format components with fixed widths for alignment
     ip = format_client_ip(client).ljust(19)  # IPv6 network max 19 chars
     timing = f"{duration_ms:.0f}ms"
     method_padded = method.ljust(7)  # Longest method is OPTIONS (7)
 
-    if use_color:
+    status_str = f"{status_color(status)}{status}{_RESET}"
-        status_str = f"{status_color(status)}{status}{_RESET}"
+    timing_str = f"{_TIMING}{timing}{_RESET}"
-        timing_str = f"{_TIMING}{timing}{_RESET}"
+    method_str = f"{method_color(method)}{method_padded}{_RESET}"
-        method_str = f"{method_color(method)}{method_padded}{_RESET}"
+    host_str = f"{_HOST}{host}{_RESET}"
-        host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
-        path_str = f"{_PATH}{path}{_RESET}"
-    else:
-        status_str = str(status)
-        timing_str = timing
-        method_str = method_padded
-        host_str = host
-        path_str = path
 
     # Format: "IP STATUS METHOD host path TIMING"
     return f"{ip} {status_str} {method_str} {host_str}{path_str} {timing_str}"
@@ -153,7 +144,6 @@ def _next_ws_id() -> int:
 
 def log_ws_open(ws) -> int:
     """Log WebSocket connection open. Returns connection ID for use in close."""
-    use_color = sys.stderr.isatty()
     ws_id = _next_ws_id()
 
     client = ws.client.host if ws.client else "-"
@@ -169,19 +159,11 @@ def log_ws_open(ws) -> int:
     origin_host = origin.split("://", 1)[-1] if origin else None
     show_origin = origin_host and origin_host != host
 
-    if use_color:
+    # 🔌 aligned with status (takes ~2 char width), ID aligned with method
-        # 🔌 aligned with status (takes ~2 char width), ID aligned with method
+    prefix = f"🔌  {_WS_OPEN}{id_str}{_RESET}"
-        prefix = f"🔌  {_WS_OPEN}{id_str}{_RESET}"
+    host_str = f"{_HOST}{host}{_RESET}"
-        host_str = f"{_HOST}{host}{_RESET}"
+    path_str = f"{_PATH}{path}{_RESET}"
-        path_str = f"{_PATH}{path}{_RESET}"
+    origin_str = f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else ""
-        origin_str = (
-            f" {_RESET}from {_HOST}{origin_host}{_RESET}" if show_origin else ""
-        )
-    else:
-        prefix = f"WS+ {id_str}"
-        host_str = host
-        path_str = path
-        origin_str = f" from {origin_host}" if show_origin else ""
 
     logger.info(f"{ip} {prefix} {host_str}{path_str}{origin_str}")
     return ws_id
@@ -209,8 +191,6 @@ WS_CLOSE_CODES = {
 
 def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None:
     """Log WebSocket connection close with duration and status."""
-    use_color = sys.stderr.isatty()
-
     id_str = f"{ws_id:02d}".ljust(7)  # Align with method field (7 chars)
     timing = f"{duration * 1000:.0f}ms"
 
@@ -220,15 +200,10 @@ def log_ws_close(ws_id: int, close_code: int | None, duration: float) -> None:
     else:
         status = WS_CLOSE_CODES.get(close_code, f"code {close_code}")
 
-    if use_color:
+    # 🔌 aligned with status, ID aligned with method
-        # 🔌 aligned with status, ID aligned with method
+    prefix = f"🔌  {_WS_CLOSE}{id_str}{_RESET}"
-        prefix = f"🔌  {_WS_CLOSE}{id_str}{_RESET}"
+    status_str = f"{_WS_STATUS}{status}{_RESET}"
-        status_str = f"{_WS_STATUS}{status}{_RESET}"
+    timing_str = f"{_TIMING}{timing}{_RESET}"
-        timing_str = f"{_TIMING}{timing}{_RESET}"
-    else:
-        prefix = f"WS- {id_str}"
-        status_str = status
-        timing_str = timing
 
     logger.info(f"{' ' * 19} {prefix} {status_str} {timing_str}")
 

paskia/fastapi/mainapp.py

@@ -1,9 +1,9 @@
-import json
 import logging
 import os
 from contextlib import asynccontextmanager
 from pathlib import Path
 
+import msgspec
 from fastapi import FastAPI, HTTPException, Request, Response
 from fastapi.responses import FileResponse, RedirectResponse
 
@@ -13,7 +13,6 @@ 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.structs import Config
 from paskia.fastapi import admin, api, auth_host, oid, ws
 
 # Import frontend instance
@@ -21,6 +20,7 @@ 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.runtime import RuntimeConfig
 
 # Configure custom logging
 configure_access_logging()
@@ -40,13 +40,13 @@ async def lifespan(app: FastAPI):  # pragma: no cover - startup path
     so that uvicorn reload / multiprocess workers inherit the settings.
     All keys are guaranteed to exist; values are already normalized by __main__.py.
     """
-    config = json.loads(os.environ["PASKIA_CONFIG"])
+    runtime = msgspec.json.decode(os.environ["PASKIA_CONFIG"], type=RuntimeConfig)
 
     try:
         await globals.init(
-            rp_id=config["rp_id"],
+            rp_id=runtime.config.rp_id,
-            rp_name=config["rp_name"],
+            rp_name=runtime.config.rp_name,
-            origins=config["origins"],
+            origins=runtime.config.origins,
             bootstrap=False,
         )
     except ValueError as e:
@@ -55,13 +55,10 @@ async def lifespan(app: FastAPI):  # pragma: no cover - startup path
         raise
 
     # Bootstrap and persist config now that the full DB is loaded
-    cli_config_data = config.get("cli_config")
+    await bootstrap_if_needed(config=runtime.config)
-    if cli_config_data:
+    if runtime.save:
-        cli_config = Config(**cli_config_data)
+        await db.update_config(runtime.config)
-        await bootstrap_if_needed(config=cli_config)
+    await flush()
-        if config.get("save"):
-            await db.update_config(cli_config)
-        await flush()
 
     # Restore uvicorn info logging (suppressed during startup in dev mode)
     # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages

paskia/fastapi/oid.py

@@ -259,8 +259,10 @@ async def _handle_refresh_token(
     The refresh_token is the session secret. On refresh:
     - Validates session exists and belongs to client
     - Extends session expiry (24h sliding window)
-    - Records current IP and user_agent
     - Issues new access_token and id_token
+
+    Note: ip and user_agent are NOT updated because the refresh request
+    comes from the OIDC client's backend, not the end user's browser.
     """
     if not refresh_token_value:
         return JSONResponse(
@@ -287,17 +289,13 @@ async def _handle_refresh_token(
             status_code=400,
         )
 
-    # Refresh the session - extend expiry and record IP/user_agent
+    # Refresh the session - extend expiry only
+    # Don't update ip/user_agent: the refresh request comes from the OIDC
+    # client's backend, not the end user's browser.
     now = datetime.now(UTC)
-    ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
-    if not ip:
-        ip = request.client.host if request.client else ""
-    user_agent = request.headers.get("user-agent", "")
 
     db.update_session(
         session.key,
-        ip=ip,
-        user_agent=user_agent,
         validated=now,
     )
 

paskia/fastapi/user.py

@@ -13,6 +13,7 @@ from paskia import db
 from paskia.authsession import (
     delete_credential,
     expires,
+    session_ctx,
 )
 from paskia.fastapi import authz, session
 from paskia.fastapi.response import MsgspecResponse
@@ -45,7 +46,7 @@ async def user_update_display_name(
             status_code=401, detail="Authentication Required", mode="login"
         )
     host = request.headers.get("host")
-    ctx = db.data().session_ctx(auth, host)
+    ctx = session_ctx(auth, host)
     if not ctx:
         raise authz.AuthException(
             status_code=401, detail="Session expired", mode="login"
@@ -74,7 +75,7 @@ async def user_update_info(
         raise authz.AuthException(
             status_code=401, detail="Authentication Required", mode="login"
         )
-    ctx = db.data().session_ctx(auth, request.headers.get("host"))
+    ctx = session_ctx(auth, request.headers.get("host"))
     if not ctx:
         raise authz.AuthException(
             status_code=401, detail="Session expired", mode="login"
@@ -112,7 +113,7 @@ async def user_update_theme(
         raise authz.AuthException(
             status_code=401, detail="Authentication Required", mode="login"
         )
-    ctx = db.data().session_ctx(auth, request.headers.get("host"))
+    ctx = session_ctx(auth, request.headers.get("host"))
     if not ctx:
         raise authz.AuthException(
             status_code=401, detail="Session expired", mode="login"
@@ -129,7 +130,7 @@ async def api_logout_all(request: Request, response: Response, auth=AUTH_COOKIE)
     if not auth:
         return {"message": "Already logged out"}
     host = request.headers.get("host")
-    ctx = db.data().session_ctx(auth, host)
+    ctx = session_ctx(auth, host)
     if not ctx:
         raise authz.AuthException(
             status_code=401, detail="Session expired", mode="login"
@@ -151,7 +152,7 @@ async def api_delete_session(
             status_code=401, detail="Authentication Required", mode="login"
         )
     host = request.headers.get("host")
-    ctx = db.data().session_ctx(auth, host)
+    ctx = session_ctx(auth, host)
     if not ctx:
         raise authz.AuthException(
             status_code=401, detail="Session expired", mode="login"

paskia/fastapi/ws.py

@@ -7,7 +7,7 @@ from fastapi import FastAPI, WebSocket
 
 from paskia import authcode, db
 from paskia.authcode import CookieCode, OIDCCode
-from paskia.authsession import get_reset
+from paskia.authsession import get_reset, session_ctx
 from paskia.db.structs import Session
 from paskia.fastapi import authz, remote
 from paskia.fastapi.session import AUTH_COOKIE, infodict
@@ -195,7 +195,7 @@ async def websocket_authenticate(
     # If there's an existing session, restrict to that user's credentials (reauth)
     session_user_uuid = None
     if auth:
-        existing_ctx = db.data().session_ctx(auth, host)
+        existing_ctx = session_ctx(auth, host)
         if existing_ctx:
             session_user_uuid = existing_ctx.user.uuid
 

paskia/fastapi/wschat.py

@@ -7,6 +7,7 @@ from uuid import UUID
 from fastapi import WebSocket
 
 from paskia import db
+from paskia.authsession import session_ctx
 from paskia.db import Credential, SessionContext
 from paskia.fastapi.session import infodict
 from paskia.fastapi.wsutil import validate_origin
@@ -90,7 +91,7 @@ async def authenticate_and_login(
     # Get credential IDs if restricting to a user's credentials
     credential_ids = None
     if auth:
-        existing_ctx = db.data().session_ctx(auth, host)
+        existing_ctx = session_ctx(auth, host)
         if existing_ctx:
             credential_ids = existing_ctx.user.credential_ids or None
 
@@ -107,7 +108,7 @@ async def authenticate_and_login(
     )
 
     # Fetch and return the full session context
-    ctx = db.data().session_ctx(secret, normalized_host)
+    ctx = session_ctx(secret, host)
     if not ctx:
         raise ValueError("Failed to create session context")
     return ctx, secret

paskia/oidc_notify.py

@@ -13,7 +13,7 @@ import httpx
 
 from paskia import db
 from paskia.util import oidjwt
-from paskia.util.hostutil import _load_config
+from paskia.util.runtime import _load_config
 
 _logger = logging.getLogger(__name__)
 

paskia/util/hostutil.py

@@ -1,27 +1,23 @@
 """Utilities for determining the auth UI host and base URLs."""
 
-import json
-import os
-from functools import lru_cache
 from urllib.parse import urlparse, urlsplit
 
+from paskia.util.runtime import _load_config
 
-@lru_cache(maxsize=1)
+
-def _load_config() -> dict:
+def _cfg():
-    """Load PASKIA_CONFIG JSON."""
+    return _load_config()
-    config_json = os.getenv("PASKIA_CONFIG")
-    if not config_json:
-        return {}
-    return json.loads(config_json)
 
 
 def is_root_mode() -> bool:
-    return _load_config().get("auth_host") is not None
+    cfg = _cfg()
+    return cfg is not None and cfg.config.auth_host is not None
 
 
 def dedicated_auth_host() -> str | None:
     """Return configured auth_host netloc, or None."""
-    auth_host = _load_config().get("auth_host")
+    cfg = _cfg()
+    auth_host = cfg.config.auth_host if cfg else None
     if not auth_host:
         return None
 
@@ -35,8 +31,10 @@ def ui_base_path() -> str:
 
 def auth_site_url() -> str:
     """Return the base URL for the auth site UI (computed at startup)."""
-    cfg = _load_config()
+    cfg = _cfg()
-    return cfg.get("site_url", "https://localhost") + cfg.get("site_path", "/auth/")
+    if cfg:
+        return cfg.site_url + cfg.site_path
+    return "https://localhost/auth/"
 
 
 def reset_link_url(token: str) -> str:
@@ -45,10 +43,10 @@ def reset_link_url(token: str) -> str:
 
 
 def normalize_origin(origin: str) -> str:
-    """Normalize an origin URL by adding https:// if no scheme is present."""
+    """Normalize an origin URL by adding https:// if no scheme is present, removing trailing slashes."""
     if "://" not in origin:
         return f"https://{origin}"
-    return origin
+    return origin.rstrip("/")
 
 
 def reload_config() -> None:
@@ -74,3 +72,15 @@ def normalize_host(raw_host: str | None) -> str | None:
         # Strip port from host:port
         netloc = netloc.rsplit(":", 1)[0]
     return netloc.lower() or None
+
+
+def format_endpoint(ep: dict) -> str:
+    """Format an endpoint dict to a listen string (e.g. 'unix:/path' or 'host:port')."""
+    if uds := ep.get("uds"):
+        return f"unix:{uds}"
+    host = ep["host"]
+    port = ep["port"]
+    # Bracket IPv6 addresses
+    if ":" in host:
+        host = f"[{host}]"
+    return f"{host}:{port}"

paskia/util/permutil.py

@@ -3,7 +3,7 @@
 from collections.abc import Sequence
 from fnmatch import fnmatchcase
 
-from paskia import db
+from paskia.authsession import session_ctx
 from paskia.util.hostutil import normalize_host
 
 __all__ = ["has_any", "has_all", "session_context"]
@@ -40,4 +40,4 @@ async def session_context(auth: str | None, host: str | None = None):
     if not auth:
         return None
     normalized_host = normalize_host(host) if host else None
-    return db.data().session_ctx(auth, normalized_host)
+    return session_ctx(auth, normalized_host)

paskia/util/runtime.py

@@ -0,0 +1,31 @@
+"""Runtime configuration utilities."""
+
+import os
+from functools import lru_cache
+
+import msgspec
+
+from paskia.db.structs import Config
+
+
+class RuntimeConfig(msgspec.Struct):
+    """Runtime configuration for the Paskia authentication server.
+
+    Wraps the db Config (CLI/stored settings) with computed runtime fields.
+    Serialized to PASKIA_CONFIG env var as JSON via msgspec.
+    """
+
+    config: Config  # CLI/stored configuration to persist
+    site_url: str  # Base URL without trailing path (e.g. https://example.com)
+    site_path: str  # Path to auth UI: "/" if auth_host, else "/auth/"
+    save: bool = False  # Whether to persist config to database
+
+
+@lru_cache(maxsize=1)
+def _load_config() -> "RuntimeConfig | None":
+    """Load RuntimeConfig from PASKIA_CONFIG env var."""
+    config_json = os.getenv("PASKIA_CONFIG")
+    if not config_json:
+        return None
+
+    return msgspec.json.decode(config_json.encode(), type=RuntimeConfig)

paskia/util/startupbox.py

@@ -1,14 +1,19 @@
 """Startup configuration box formatting utilities."""
 
+from __future__ import annotations
+
 import os
 import re
 from sys import stderr
 from typing import TYPE_CHECKING
 
+from fastapi_vue.hostutil import parse_endpoints
+
 from paskia._version import __version__
+from paskia.util.hostutil import format_endpoint
 
 if TYPE_CHECKING:
-    from paskia.config import PaskiaConfig
+    from paskia.util.runtime import RuntimeConfig
 
 BOX_WIDTH = 60  # Inner width (excluding box chars)
 
@@ -42,7 +47,7 @@ def bottom() -> str:
     return "┗" + "━" * (BOX_WIDTH + 2) + "┛\n"
 
 
-def print_startup_config(config: "PaskiaConfig") -> None:
+def print_startup_config(runtime: RuntimeConfig) -> None:
     """Print server configuration on startup."""
     # Key graphic with yellow shading (bright for highlights, dark for body)
     y = YELLOW  # Dark yellow for main body
@@ -57,41 +62,40 @@ def print_startup_config(config: "PaskiaConfig") -> None:
     lines.append(
         line(
             f"{b}█{y}     {b}█{y}▀▀▀▀{b}█{y}▀▀{b}█{y}▀▀{b}█{r}    {w}"
-            + config.site_url
+            + runtime.site_url
-            + config.site_path
+            + runtime.site_path
             + r
         )
     )
     lines.append(line(f" {y}▀▀▀▀▀{r}"))
 
     # Format auth host section
-    if config.auth_host:
+    if runtime.config.auth_host:
-        lines.append(line(f"Auth Host:      {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
-    devmode = os.environ.get("PASKIA_VITE_URL")
+    if DEVMODE:
-    if devmode:
+        lines.append(line(f"Dev Frontend:   {os.environ.get('PASKIA_VITE_URL')}"))
-        lines.append(line(f"Dev Frontend:   {devmode}"))
 
-    # Format listen address with scheme
+    # Format listen endpoints (dev mode only uses the first endpoint)
-    if config.uds:
+
-        listen = f"unix:{config.uds}"
+    endpoints = list(parse_endpoints(runtime.config.listen, P))
-    elif config.host:
+    if DEVMODE:
-        listen = f"http://{config.host}:{config.port}"
+        endpoints = endpoints[:1]  # server.run reload=True uses only one
-    else:
+    parts = [format_endpoint(ep) for ep in endpoints]
-        listen = f"http://0.0.0.0:{config.port} + [::]:{config.port}"
+    lines.append(line(f"Backend:        {' '.join(parts)}"))
-    lines.append(line(f"Backend:        {listen}"))
 
     # Relying Party line (omit name if same as id)
-    rp_id = config.rp_id
+    rp_id = runtime.config.rp_id
-    rp_name = config.rp_name
+    rp_name = runtime.config.rp_name
-    if rp_name and rp_name != rp_id:
+    suffix = f" ({rp_name})" if rp_name and rp_name != rp_id else ""
-        lines.append(line(f"Relying Party:  {rp_id}  ({rp_name})"))
+    lines.append(line(f"Relying Party:  {rp_id}{suffix}"))
-    else:
-        lines.append(line(f"Relying Party:  {rp_id}"))
 
     # Format origins section
-    allowed = config.origins
+    allowed = runtime.config.origins
     if allowed:
         lines.append(line("Permitted Origins:"))
         for origin in sorted(allowed):