Replace ua-parser wrapper with uarite uaparse

The **.{rp-id} wildcard was only added when the resulting origins dict
was empty, so a legacy database with a dedicated auth host but no
configured origins ended up allowing only the auth host.

paskia/__main__.py

@@ -24,6 +24,7 @@ EPILOG = """\
 Examples:
   paskia init example.com "Example Corporation"
   paskia migrate example.com
+  paskia --listen 4402 --save
   paskia
 """
 
@@ -178,9 +179,23 @@ def cmd_init(args: argparse.Namespace) -> None:
 
 
 def cmd_migrate(args: argparse.Namespace) -> None:
-    """Convert a legacy <rp-id>.paskiadb database to paskia.kantadb."""
+    """Convert or merge a legacy/current database into paskia.kantadb."""
-    rp_id = legacy.migrate_legacy_database(args.rp_id)
+    merging = db_file_path().exists()
-    print(f"✅ Converted legacy database to {db_file_path()} (domain: {rp_id})")
+    rp_ids = legacy.migrate_database(args.source)
+    action = "Merged into existing" if merging else "Converted to"
+    print(f"✅ {action} {db_file_path()} (domains: {', '.join(rp_ids)})")
+
+
+def _save_listen(db_path: Path, listen: list[str] | None) -> None:
+    """Persist the listen endpoints to the stored configuration."""
+    kanta = Kanta(str(db_path), DB())
+
+    async def _write() -> None:
+        async with kanta:
+            with kanta.transaction("serve:save_listen"):
+                kanta.data.config.listen = listen
+
+    asyncio.run(_write())
 
 
 def cmd_serve(args: argparse.Namespace) -> None:
@@ -195,6 +210,10 @@ def cmd_serve(args: argparse.Namespace) -> None:
             )
         raise SystemExit(f"Database {db_path} not found — run 'paskia init' first.")
 
+    if args.save and args.listen is not None:
+        # '--listen ""' clears the stored endpoints (back to the default)
+        _save_listen(db_path, _split_multi(args.listen) or None)
+
     config = _load_stored_config(db_path)
 
     listen = _split_multi(args.listen) or config.listen
@@ -237,6 +256,12 @@ def main():
         epilog=EPILOG,
     )
     _add_listen_option(parser)
+    parser.add_argument(
+        "--save",
+        action="store_true",
+        help="Save --listen to the database for future runs. "
+        "Use --listen \"\" to clear the stored endpoints.",
+    )
 
     init_parser = argparse.ArgumentParser(
         prog="paskia init",
@@ -263,14 +288,18 @@ def main():
 
     migrate_parser = argparse.ArgumentParser(
         prog="paskia migrate",
-        description="Convert a legacy <rp-id>.paskiadb database to paskia.kantadb",
+        description="Convert a legacy <rp-id>.paskiadb database to paskia.kantadb, "
+        "or merge a legacy database / another paskia.kantadb into an existing one",
         formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     migrate_parser.add_argument(
-        "rp_id",
+        "source",
         nargs="?",
-        help="rp-id of the legacy database to convert, selecting "
+        help="rp-id of the legacy database to convert, or path to a legacy "
-        "<rp-id>.paskiadb when several legacy candidates exist.",
+        "<rp-id>.paskiadb directory/file or a current-format paskia.kantadb "
+        "file. When paskia.kantadb already exists, the source data is merged "
+        "into it. Without an argument, a single legacy *.paskiadb candidate "
+        "in the current directory is selected automatically.",
     )
 
     argv = sys.argv[1:]

paskia/db/legacy.py

@@ -1,14 +1,15 @@
-"""Legacy database format reader and converter.
+"""Legacy database format reader, converter and database merging.
 
 Retains the msgspec structs used by the old ``<rp-id>.paskiadb/main.db``
 format so existing databases can be opened and converted to the combined
-``paskia.kantadb`` format. Only the structs whose shape differs from the
+``paskia.kantadb`` format, and implements the merge of incoming data
-current schema are redefined here; unchanged structs are imported from
+(legacy or current format) into an existing ``paskia.kantadb``. Only the
-``paskia.db.structs``.
+structs whose shape differs from the current schema are redefined here;
+unchanged structs are imported from ``paskia.db.structs``.
 
 Assumes the on-disk records are in the latest legacy format (schema
-migrations were discarded together with the old format). This module will
+migrations were discarded together with the old format). The legacy
-be deleted once legacy conversion is no longer supported.
+structs will be deleted once legacy conversion is no longer supported.
 """
 
 from __future__ import annotations
@@ -101,16 +102,23 @@ def _read_legacy(path: Path) -> LegacyDB:
     return asyncio.run(_read())
 
 
-def convert_legacy_database(src: Path, dst: Path) -> Config:
+def _read_kantadb(path: Path) -> DB:
-    """Convert a legacy main.db file into the combined kantadb format.
+    """Open a current-format database read-only and return its contents."""
+    kanta = Kanta(str(path), DB())
 
-    Reads the legacy database at ``src`` and writes a fresh database at
+    async def _read() -> DB:
-    ``dst``. All credentials and sessions are stamped with the legacy
+        await kanta.open(readonly=True)
-    database's rp-id; the OIDC provider carries over as-is (it is
+        return kanta.data
-    instance-global).
+
-    Returns the converted (new-format) configuration.
+    return asyncio.run(_read())
+
+
+def _legacy_to_db(old: LegacyDB) -> DB:
+    """Convert legacy database contents to the combined kantadb format.
+
+    All credentials and sessions are stamped with the legacy database's
+    rp-id; the OIDC provider carries over as-is (it is instance-global).
     """
-    old = _read_legacy(src)
     rp_id = old.config.rp_id
 
     from paskia.domains import origin_key  # noqa: PLC0415  (import cycle)
@@ -120,9 +128,10 @@ def convert_legacy_database(src: Path, dst: Path) -> Config:
         origins[origin_key(origin)] = True
     if old.config.auth_host:
         origins[origin_key(old.config.auth_host)] = OriginEntry(auth_host=True)
-    if not origins:
+    if not old.config.origins:
         # Legacy semantics: no origins configured = the whole rp-id domain
-        # allowed. The new format requires explicit entries.
+        # allowed, regardless of a dedicated auth host. The new format
+        # requires explicit entries.
         origins[f"**.{rp_id}"] = True
 
     new_config = Config(
@@ -169,28 +178,94 @@ def convert_legacy_database(src: Path, dst: Path) -> Config:
         reset_tokens=old.reset_tokens,
         oidc=old.oidc,
     )
+    return converted
 
+
+def _migration_label(incoming: DB) -> str:
+    """Transaction label for a migration; multiple rp-ids join with slashes."""
+    return f"migrate:cli:{'/'.join(incoming.config.domains)}"
+
+
+def _write_fresh(data: DB, dst: Path, label: str) -> None:
+    """Write a fresh database at ``dst`` with the given contents."""
     new_db = DB()
     kanta = Kanta(str(dst), new_db)
 
-    @kanta.bootstrap
+    @kanta.bootstrap(action=label)
-    def _seed(data: DB) -> None:
+    def _seed(target: DB) -> None:
-        data.config = converted.config
+        target.config = data.config
-        data.permissions = converted.permissions
+        target.permissions = data.permissions
-        data.orgs = converted.orgs
+        target.orgs = data.orgs
-        data.roles = converted.roles
+        target.roles = data.roles
-        data.users = converted.users
+        target.users = data.users
-        data.credentials = converted.credentials
+        target.credentials = data.credentials
-        data.sessions = converted.sessions
+        target.sessions = data.sessions
-        data.reset_tokens = converted.reset_tokens
+        target.reset_tokens = data.reset_tokens
-        data.oidc = converted.oidc
+        target.oidc = data.oidc
 
     async def _write() -> None:
         async with kanta:
             pass
 
     asyncio.run(_write())
-    return new_config
+
+
+def convert_legacy_database(src: Path, dst: Path) -> Config:
+    """Convert a legacy main.db file into the combined kantadb format.
+
+    Reads the legacy database at ``src`` and writes a fresh database at
+    ``dst``. Returns the converted (new-format) configuration.
+    """
+    converted = _legacy_to_db(_read_legacy(src))
+    _write_fresh(converted, dst, _migration_label(converted))
+    return converted.config
+
+
+def _merge_data(data: DB, incoming: DB) -> None:
+    """Merge ``incoming`` contents into the live ``data`` object.
+
+    Records are uuid-keyed (or hash-keyed for sessions/reset tokens), so
+    identical keys denote the same item: existing entries win, new entries
+    are added. Domains merge per rp-id with a union of allowed origins;
+    the existing instance's listen endpoints and OIDC signing key win.
+    """
+    for rp_id, domain in incoming.config.domains.items():
+        existing = data.config.domains.get(rp_id)
+        if existing is None:
+            data.config.domains[rp_id] = domain
+            continue
+        for origin, entry in domain.origins.items():
+            existing.origins.setdefault(origin, entry)
+        if existing.rp_name is None:
+            existing.rp_name = domain.rp_name
+    for bucket in (
+        "permissions",
+        "orgs",
+        "roles",
+        "users",
+        "credentials",
+        "sessions",
+        "reset_tokens",
+    ):
+        target_map = getattr(data, bucket)
+        for key, value in getattr(incoming, bucket).items():
+            target_map.setdefault(key, value)
+    for uuid, client in incoming.oidc.clients.items():
+        data.oidc.clients.setdefault(uuid, client)
+    if data.oidc.key is None:
+        data.oidc.key = incoming.oidc.key
+
+
+def merge_database(dst: Path, incoming: DB) -> None:
+    """Merge ``incoming`` contents into the existing database at ``dst``."""
+    kanta = Kanta(str(dst), DB())
+
+    async def _merge() -> None:
+        async with kanta:
+            with kanta.transaction(_migration_label(incoming)):
+                _merge_data(kanta.data, incoming)
+
+    asyncio.run(_merge())
 
 
 def find_legacy_databases(cwd: Path | None = None) -> list[Path]:
@@ -211,50 +286,101 @@ def find_legacy_databases(cwd: Path | None = None) -> list[Path]:
     return candidates
 
 
-def migrate_legacy_database(rp_id: str | None = None) -> str:
+def _resolve_source(source: str | None) -> tuple[Path, bool, Path, Path | None]:
-    """Convert a legacy database to ``paskia.kantadb``.
+    """Resolve the migrate source.
 
-    With ``rp_id``, selects the ``<rp-id>.paskiadb`` candidate by name;
+    ``source`` may be an rp-id (selecting ``<rp-id>.paskiadb`` in the
-    without it, exactly one candidate must exist. Returns the migrated
+    current directory), a path to a legacy ``*.paskiadb`` directory or
-    domain's rp-id. The converted legacy directory/file is renamed aside
+    file, or a path to a current-format ``*.kantadb`` file. Without
-    to ``<name>.converted-bak`` rather than deleted.
+    ``source``, exactly one legacy candidate must exist in the current
+    directory.
 
-    Raises SystemExit when ``paskia.kantadb`` already exists, when no
+    Returns ``(db_file, is_legacy, users_dir, rename_target)`` where
-    candidate matches, or when several candidates exist and no ``rp_id``
+    ``users_dir`` holds auxiliary user files (avatars) and
-    was given to select one.
+    ``rename_target`` is the legacy directory/file to rename aside after
+    a successful migration (None for current-format sources).
     """
-    target = db_file_path()
+
-    if target.exists():
+    def legacy(src: Path) -> tuple[Path, bool, Path, Path]:
-        raise SystemExit(f"Database {target} already exists — nothing to migrate.")
+        return (
-    candidates = find_legacy_databases()
+            src / "main.db" if src.is_dir() else src,
-    if rp_id is not None:
+            True,
-        name = f"{rp_id}.paskiadb"
+            src / "users" if src.is_dir() else src.parent / "users",
-        matches = [c for c in candidates if c.name == name]
+            src,
+        )
+
+    if source is not None:
+        path = Path(source)
+        if path.is_dir():
+            if (path / "main.db").is_file():
+                return legacy(path)
+            raise SystemExit(f"No legacy main.db found in directory {path}.")
+        if path.is_file():
+            if path.suffix == ".paskiadb":
+                return legacy(path)
+            return path, False, path.parent / "paskia.data" / "users", None
+        # Not a path: treat as rp-id selecting a legacy candidate by name
+        name = f"{source}.paskiadb"
+        matches = [c for c in find_legacy_databases() if c.name == name]
         if not matches:
-            found = ", ".join(str(c) for c in candidates) or "none"
+            found = ", ".join(str(c) for c in find_legacy_databases()) or "none"
             raise SystemExit(
                 f"No legacy database {name} in this directory (candidates: {found})."
             )
-        src = matches[0]
+        return legacy(matches[0])
-    elif not candidates:
+    candidates = find_legacy_databases()
+    if not candidates:
         raise SystemExit("No legacy *.paskiadb database found — nothing to migrate.")
-    elif len(candidates) > 1:
+    if len(candidates) > 1:
         names = ", ".join(str(c) for c in candidates)
         raise SystemExit(
             f"Multiple legacy databases found ({names}) — select one with "
             "'paskia migrate <rp-id>'."
         )
-    else:
+    return legacy(candidates[0])
-        src = candidates[0]
-    legacy_file = src / "main.db" if src.is_dir() else src
-    config = convert_legacy_database(legacy_file, target)
 
-    # Move persisted user files (avatars) to the new data root
+
-    legacy_users = src / "users" if src.is_dir() else None
+def _move_user_files(src_users: Path) -> None:
-    if legacy_users is not None and legacy_users.is_dir():
+    """Move persisted user files (avatars) to the new data root."""
+    if not src_users.is_dir():
+        return
     target_users = users_root_path(create_root=True)
-        for child in legacy_users.iterdir():
+    for child in src_users.iterdir():
+        if (target_users / child.name).exists():
+            continue
         shutil.move(str(child), str(target_users / child.name))
 
-    shutil.move(str(src), str(src.with_name(src.name + ".converted-bak")))
+
-    return next(iter(config.domains))
+def migrate_database(source: str | None = None) -> list[str]:
+    """Convert or merge a database into ``paskia.kantadb``.
+
+    The source may be a legacy ``<rp-id>.paskiadb`` database (selected by
+    rp-id or path) or a current-format ``*.kantadb`` file given by path.
+    When ``paskia.kantadb`` already exists, the incoming data is merged
+    into it (uuid-keyed records make conflicts a non-issue); otherwise a
+    fresh database is written. Returns the migrated domains' rp-ids. A
+    migrated legacy source is renamed aside to ``<name>.converted-bak``
+    rather than deleted; a merged kantadb source is left in place.
+    """
+    target = db_file_path()
+    db_file, is_legacy, users_dir, rename_target = _resolve_source(source)
+    if db_file.resolve() == target.resolve():
+        raise SystemExit(f"{db_file} is the active database — nothing to migrate.")
+
+    incoming = (
+        _legacy_to_db(_read_legacy(db_file)) if is_legacy else _read_kantadb(db_file)
+    )
+    rp_ids = list(incoming.config.domains)
+
+    if target.exists():
+        merge_database(target, incoming)
+    else:
+        _write_fresh(incoming, target, _migration_label(incoming))
+
+    _move_user_files(users_dir)
+    if rename_target is not None and rename_target.exists():
+        shutil.move(
+            str(rename_target),
+            str(rename_target.with_name(rename_target.name + ".converted-bak")),
+        )
+    return rp_ids

paskia/fastapi/remote.py

@@ -15,6 +15,7 @@ from uuid import UUID
 
 import base64url
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from uarite import uaparse
 
 from paskia import authcode, db, remoteauth
 from paskia.authcode import CookieCode
@@ -23,7 +24,7 @@ from paskia.domains import current_domain, registry
 from paskia.fastapi.session import AUTH_COOKIE, infodict
 from paskia.fastapi.wschat import authenticate_and_login
 from paskia.fastapi.wsutil import validate_origin, websocket_error_handler
-from paskia.util import pow, useragent
+from paskia.util import pow
 
 # Create a FastAPI subapp for remote auth WebSocket endpoints
 app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -458,9 +459,7 @@ async def websocket_remote_auth_permit(ws: WebSocket, auth=AUTH_COOKIE):
                         if requesting_domain
                         else request.rp_id
                     ),
-                    "user_agent_pretty": useragent.compact_user_agent(
+                    "user_agent_pretty": uaparse(request.user_agent).pretty,
-                        request.user_agent
-                    ),
                     "client_ip": request.ip,
                     "action": request.action,
                     "pow": {

paskia/util/apistructs.py

@@ -11,10 +11,10 @@ from datetime import datetime
 from uuid import UUID
 
 import msgspec
+from uarite import uaparse
 
 from paskia import db
 from paskia.db.structs import Credential, Org, OriginEntry, Permission, Role, User
-from paskia.util import useragent
 
 # -------------------------------------------------------------------------
 # API structs - inherit from db structs, add uuid for serialization
@@ -124,7 +124,7 @@ class ApiUserSession(msgspec.Struct, omit_defaults=True):
             credential_uuid=s.credential_uuid,
             host=s.host,
             ip=s.ip,
-            user_agent=useragent.compact_user_agent(s.user_agent),
+            user_agent=uaparse(s.user_agent).pretty,
             validated=s.validated,
             last_renewed=s.validated,
             is_current=s.key == current_key,

paskia/util/startupbox.py

@@ -6,11 +6,13 @@ import os
 import re
 from sys import stderr
 from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 from fastapi_vue.hostutil import parse_endpoints
 
 from paskia._version import __version__
 from paskia.domains import auth_host_url, origin_url, partition_origins
+from paskia.util import hostutil
 from paskia.util.constants import DEFAULT_PORT, DEVMODE
 from paskia.util.hostutil import format_endpoint, wildcard_base
 
@@ -85,9 +87,47 @@ def _origin_phrase(key: str, rp_id: str) -> str:
     return _compact_url(origin_url(key))
 
 
+def _covered_by_wildcard(key: str, pattern: str) -> bool:
+    """Whether an origins-table key is redundant given a wildcard key.
+
+    Mirrors DomainConfig matching (sansio._allowlisted): a wildcard covers
+    hostnames under its base over https (any port), except under localhost
+    where any scheme and any port match. Plain http entries outside
+    localhost are therefore never covered and stay listed.
+    """
+    base = wildcard_base(pattern)
+    if base is None:
+        return False
+    # Keys are bare hosts (https:// and '/' stripped by origin_key, port
+    # kept) or full origins; urlparse needs a scheme or '//' prefix.
+    hostname = urlparse(key if "://" in key else f"//{key}").hostname
+    if not hostname:
+        return False
+    if pattern.startswith("**."):
+        matched = hostutil.is_subdomain(hostname, base)
+    else:
+        # '*.base' covers exactly one subdomain level
+        matched = hostname.endswith(f".{base}") and "." not in hostname[
+            : -len(base) - 1
+        ]
+    if not matched:
+        return False
+    if hostutil.is_subdomain(base, "localhost"):
+        return True  # localhost: any scheme, any port
+    return "://" not in key or key.startswith("https://")
+
+
 def _signin_summary(in_domain: list[str], rp_id: str) -> str:
     """Compact summary of a domain's in-domain sign-in sites."""
-    phrases = [_origin_phrase(key, rp_id) for key in sorted(in_domain)]
+    # Prune entries already covered by a reported wildcard (e.g. the auth
+    # host under '**.{rp-id}'); http origins outside localhost survive.
+    wildcards = [k for k in in_domain if wildcard_base(k)]
+    keys = [
+        k
+        for k in in_domain
+        if wildcard_base(k) or not any(_covered_by_wildcard(k, w) for w in wildcards)
+    ]
+    phrases = [_origin_phrase(key, rp_id) for key in sorted(keys)]
     if len(phrases) > 2:
         n = len(phrases) - 1
         return f"{phrases[0]}, +{n} site{'s' if n > 1 else ''}"

paskia/util/useragent.py

@@ -1,29 +0,0 @@
-from ua_parser import parse
-
-
-def compact_user_agent(ua: str | None) -> str:
-    """Format user agent string into a compact display format.
-
-    Returns empty string for empty/missing user agents.
-    Returns original UA for unrecognized ones.
-    """
-    if not ua or not ua.strip() or ua == "-":
-        return ""
-    r = parse(ua)
-    browser = r.user_agent.family if r.user_agent else None
-    ver = r.user_agent.major if r.user_agent else ""
-    os_name = r.os.family if r.os else None
-    dev = r.device.family if r.device else None
-    # If browser is unrecognized, return original UA
-    if browser in (None, "Other") and os_name in (None, "Other"):
-        return ua
-    # Filter out "Other" values
-    browser = browser if browser and browser != "Other" else ""
-    os_name = os_name if os_name and os_name != "Other" else ""
-    # Exclude device if it's "Other" or matches browser family (parser bug)
-    if dev in (None, "Other") or dev == browser:
-        dev = ""
-    # Build compact string, filtering empty parts
-    parts = [f"{browser}/{ver}" if browser else "", os_name, dev]
-    result = " ".join(p for p in parts if p).strip()
-    return result

pyproject.toml

@@ -22,8 +22,8 @@ dependencies = [
     "jsondiff>=2.2.1",
     "msgspec>=0.20.0",
     "fastapi-vue~=1.4.2",
-    "ua-parser[regex]>=1.0.1",
     "kanta>=0.7.0",
+    "uarite>=0.2.1",
 ]
 [dependency-groups]
 dev = [

tests/test_cli.py

@@ -2,7 +2,8 @@
 
 The CLI is split into ``paskia init`` (create the combined paskia.kantadb
 with the initial domain(s)), ``paskia migrate`` (convert a legacy
-``<rp-id>.paskiadb`` database), and bare ``paskia`` (serve the stored
+``<rp-id>.paskiadb`` database, or merge a legacy/current database into an
+existing paskia.kantadb), and bare ``paskia`` (serve the stored
 domains; never migrates).
 """
 
@@ -20,7 +21,7 @@ from kanta import Kanta
 
 from paskia.__main__ import _load_stored_config, main
 from paskia.db import legacy
-from paskia.db.structs import DB, Config
+from paskia.db.structs import DB, Config, DomainConfig
 from paskia.util.runtime import ServeConfig, clear_cache
 
 
@@ -184,6 +185,22 @@ def test_serve_listen_override_not_persisted(run_cli, tmp_path):
     assert stored_config(tmp_path).listen == ["4402"]
 
 
+def test_serve_listen_save_persists(run_cli, tmp_path):
+    run_cli("init", "--listen", "4402")
+    calls = run_cli("--listen", "4403", "--save")
+
+    assert calls["listen"] == ["4403"]
+    assert stored_config(tmp_path).listen == ["4403"]
+
+
+def test_serve_listen_save_clear(run_cli, tmp_path):
+    """--listen "" --save clears the stored endpoints (back to default)."""
+    run_cli("init", "--listen", "4402")
+    run_cli("--listen", "", "--save")
+
+    assert stored_config(tmp_path).listen is None
+
+
 def test_serve_suggests_migrate_when_legacy_present(run_cli, tmp_path):
     write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="example.com"))
     with pytest.raises(SystemExit, match="paskia migrate"):
@@ -204,6 +221,8 @@ def test_migrate_converts_legacy_database(run_cli, tmp_path):
     config = stored_config(tmp_path)
     assert list(config.domains) == ["example.com"]
     assert config.domains["example.com"].rp_name == "Legacy Name"
+    # Migration transaction is labeled with the migrated rp-id
+    assert b"migrate:cli:example.com" in (tmp_path / "paskia.kantadb").read_bytes()
     # Legacy directory renamed aside, user files moved over
     assert not src_dir.exists()
     assert (tmp_path / "example.com.paskiadb.converted-bak").is_dir()
@@ -242,11 +261,93 @@ def test_migrate_unknown_rp_id(run_cli, tmp_path):
         run_cli("migrate", "nope.com")
 
 
-def test_migrate_refuses_existing_database(run_cli):
+def test_migrate_merges_legacy_into_existing_database(run_cli, tmp_path):
-    run_cli("init")
+    """An existing paskia.kantadb is not refused — data is merged in."""
-    with pytest.raises(SystemExit, match="already exists"):
+    run_cli("init", "company.com", "Company")
+    write_legacy_db(tmp_path, legacy.LegacyConfig(rp_id="example.com", rp_name="Ex"))
+
     run_cli("migrate")
 
+    config = stored_config(tmp_path)
+    assert list(config.domains) == ["company.com", "example.com"]
+    assert config.domains["example.com"].rp_name == "Ex"
+    assert (tmp_path / "example.com.paskiadb.converted-bak").is_dir()
+
+
+def write_kantadb(root: Path, domains: dict, name: str = "paskia.kantadb") -> Path:
+    """Create a current-format database file with the given config domains."""
+    db_file = root / name
+
+    config = Config(
+        domains={rp_id: DomainConfig(rp_name=name_) for rp_id, name_ in domains.items()}
+    )
+
+    async def _write() -> None:
+        new_db = DB()
+        kanta = Kanta(str(db_file), new_db)
+
+        @kanta.bootstrap
+        def _seed(data: DB) -> None:
+            data.config = config
+
+        async with kanta:
+            pass
+
+    asyncio.run(_write())
+    return db_file
+
+
+def test_migrate_merges_kantadb_into_existing_database(run_cli, tmp_path):
+    run_cli("init", "company.com", "Company")
+    src = write_kantadb(tmp_path, {"other.com": "Other"}, name="other.kantadb")
+
+    run_cli("migrate", str(src))
+
+    config = stored_config(tmp_path)
+    assert list(config.domains) == ["company.com", "other.com"]
+    assert config.domains["other.com"].rp_name == "Other"
+    # Current-format sources are left in place
+    assert src.is_file()
+    assert b"migrate:cli:other.com" in (tmp_path / "paskia.kantadb").read_bytes()
+
+
+def test_migrate_merge_label_combines_rp_ids(run_cli, tmp_path):
+    """A multi-domain source merges in one transaction, rp-ids slash-joined."""
+    run_cli("init", "company.com")
+    src = write_kantadb(tmp_path, {"one.com": "One", "two.com": "Two"}, name="x.kantadb")
+
+    run_cli("migrate", str(src))
+
+    assert b"migrate:cli:one.com/two.com" in (tmp_path / "paskia.kantadb").read_bytes()
+
+
+def test_migrate_merges_shared_domain_origins(run_cli, tmp_path):
+    """Same rp-id in both databases: origins union, existing rp-name wins."""
+    run_cli("init", "example.com", "Existing Name")
+    src = write_kantadb(tmp_path, {"example.com": "Incoming Name"}, name="x.kantadb")
+
+    run_cli("migrate", str(src))
+
+    domain = stored_config(tmp_path).domains["example.com"]
+    assert domain.rp_name == "Existing Name"
+    assert set(domain.origins) == {"**.example.com"}
+
+
+def test_migrate_refuses_active_database_as_source(run_cli):
+    run_cli("init")
+    with pytest.raises(SystemExit, match="active database"):
+        run_cli("migrate", "paskia.kantadb")
+
+
+def test_migrate_kantadb_to_fresh_target(run_cli, tmp_path):
+    src_dir = tmp_path / "elsewhere"
+    src_dir.mkdir()
+    src = write_kantadb(src_dir, {"other.com": "Other"})
+
+    run_cli("migrate", str(src))
+
+    assert list(stored_config(tmp_path).domains) == ["other.com"]
+
 
 def test_migrate_without_legacy_database(run_cli):
     with pytest.raises(SystemExit, match="No legacy"):

tests/test_domains.py

@@ -941,6 +941,22 @@ class TestLegacyConversion:
         config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
         assert config.domains["example.com"].origins == {"**.example.com": True}
 
+    def test_convert_auth_host_with_empty_origins_keeps_wildcard(self, tmp_path):
+        """A dedicated auth host with no configured origins still allowed
+        the whole rp-id domain in the legacy format — the auth host must
+        not become the only allowed origin."""
+        src_file = tmp_path / "main.db"
+        asyncio.run(
+            _write_legacy(
+                src_file,
+                LegacyConfig(rp_id="example.com", auth_host="auth.example.com"),
+            )
+        )
+        config = convert_legacy_database(src_file, tmp_path / "paskia.kantadb")
+        origins = config.domains["example.com"].origins
+        assert origins["**.example.com"] is True
+        assert origins["auth.example.com"] == OriginEntry(auth_host=True)
+
 
 # -------------------------------------------------------------------------
 # Transaction log censoring