feat(tenant): add operator provisioning for tenants, API keys, and domains

Why:
- nothing over HTTP could create the first tenant: every /v1 route needs an API
  key, and a key cannot exist before its tenant. The service was unusable by a
  human without hand-written SQL.

Changes:
- add `provision_tenant`, owning tenant reuse-or-create, key generation and
  hashing, and domain registration in one transaction
- expose it as `python -m src.cli.provision_tenant`, alongside
  `alembic upgrade head` and `qdrant_bootstrap`
- add `tenants.get_by_slug`/`create` and `api_keys.create`
- log `tenant.provisioned` / `api_key.provisioned` with the key prefix only

Impact:
- a third deployment step; the plaintext key is printed once and never logged
  or stored (ADR-0011, ADR-0009)

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/application/tenants/__init__.py

@@ -0,0 +1,9 @@
+"""Operator-run tenant provisioning (ADR-0008, ADR-0009)."""
+
+from src.application.tenants.provisioning import (
+    DEFAULT_SCOPES,
+    ProvisionResult,
+    provision_tenant,
+)
+
+__all__ = ["DEFAULT_SCOPES", "ProvisionResult", "provision_tenant"]

src/application/tenants/provisioning.py

@@ -0,0 +1,123 @@
+"""Provision a tenant, its first API key, and its domains (ADR-0008, ADR-0009).
+
+Nothing in the HTTP surface can bootstrap a tenant: every `/v1` route needs a
+key, and a key can only exist once a tenant does. That chicken-and-egg is why
+this is an operator-run deployment step (`src/cli/provision_tenant.py`) rather
+than an endpoint — the same reasoning that keeps `alembic upgrade head` and
+`qdrant_bootstrap` off the request path.
+
+This is the package's only caller-facing entry point. It owns the whole
+composition — tenant reuse-or-create, key generation and hashing, domain
+registration, and the single transaction the three share — so a caller cannot
+get the order wrong or commit a key whose tenant never landed (CLAUDE.md,
+"prefer deep modules"). The plaintext key is returned exactly once and is never
+logged (ADR-0011 forbids plaintext keys in logs); only its non-secret
+`key_prefix` appears in the event.
+"""
+
+import uuid
+from dataclasses import dataclass
+
+import structlog
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.keys import generate_api_key, hash_secret
+from src.infrastructure.postgres.repositories import api_keys as api_keys_repo
+from src.infrastructure.postgres.repositories import tenant_domains as domains_repo
+from src.infrastructure.postgres.repositories import tenants as tenants_repo
+
+logger = structlog.get_logger(__name__)
+
+DEFAULT_SCOPES = ("files:write", "domains:read", "domains:write")
+
+
+@dataclass(frozen=True)
+class ProvisionResult:
+    tenant_id: uuid.UUID
+    tenant_slug: str
+    tenant_created: bool
+    api_key_id: uuid.UUID
+    api_key_prefix: str
+    api_key: str
+    """The plaintext bearer token. Only ever returned here — never stored, never logged."""
+    domains_created: tuple[str, ...]
+    domains_existing: tuple[str, ...]
+
+
+async def provision_tenant(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    slug: str,
+    name: str | None = None,
+    key_name: str = "bootstrap",
+    scopes: tuple[str, ...] = DEFAULT_SCOPES,
+    domains: tuple[str, ...] = (),
+    actor_type: str = "backend",
+) -> ProvisionResult:
+    """Create (or reuse) the tenant, issue a key, and register `domains`.
+
+    Re-running with the same `slug` reuses the tenant and its existing domains
+    rather than failing, so an operator can add a key to a live tenant with the
+    same command they used to create it. A *new* key is issued on every run —
+    keys are write-once by construction (only the hash is stored), so there is
+    nothing to return for an existing one.
+    """
+    key_prefix, secret, full_key = generate_api_key()
+
+    async with sessionmaker() as session:
+        tenant = await tenants_repo.get_by_slug(session, slug)
+        tenant_created = tenant is None
+        if tenant is None:
+            tenant = tenants_repo.create(session, slug=slug, name=name or slug)
+            # `api_keys.tenant_id` and `tenant_domains.tenant_id` FK to this row
+            # and the mapped classes carry no ORM relationship for the unit of
+            # work to order by itself, so the insert has to land first.
+            await session.flush()
+
+        api_key = api_keys_repo.create(
+            session,
+            tenant_id=tenant.id,
+            name=key_name,
+            key_prefix=key_prefix,
+            key_hash=hash_secret(secret),
+            scopes=list(scopes),
+            actor_type=actor_type,
+            created_by="cli:provision_tenant",
+        )
+
+        created: list[str] = []
+        existing: list[str] = []
+        for domain in domains:
+            if await domains_repo.get(session, tenant_id=tenant.id, domain=domain) is not None:
+                existing.append(domain)
+                continue
+            domains_repo.create(session, tenant_id=tenant.id, domain=domain, display_name=domain)
+            created.append(domain)
+
+        await session.flush()
+        tenant_id, api_key_id, tenant_slug = tenant.id, api_key.id, tenant.slug
+        await session.commit()
+
+    if tenant_created:
+        logger.info("tenant.provisioned", tenant_id=str(tenant_id), tenant_slug=tenant_slug)
+    for domain in created:
+        logger.info("domain.created", tenant_id=str(tenant_id), domain=domain)
+    logger.info(
+        "api_key.provisioned",
+        tenant_id=str(tenant_id),
+        api_key_id=str(api_key_id),
+        key_prefix=key_prefix,
+        scopes=list(scopes),
+        actor_type=actor_type,
+    )
+
+    return ProvisionResult(
+        tenant_id=tenant_id,
+        tenant_slug=tenant_slug,
+        tenant_created=tenant_created,
+        api_key_id=api_key_id,
+        api_key_prefix=key_prefix,
+        api_key=full_key,
+        domains_created=tuple(created),
+        domains_existing=tuple(existing),
+    )

src/cli/provision_tenant.py

@@ -0,0 +1,94 @@
+"""Create a tenant, issue its first API key, and register its domains.
+
+    uv run python -m src.cli.provision_tenant --slug acme --domain fire
+
+A deployment step, like `alembic upgrade head` and `src.cli.qdrant_bootstrap`.
+It exists because nothing over HTTP can bootstrap a tenant: every `/v1` route
+requires an API key, and a key cannot exist before its tenant does.
+
+The plaintext key is printed to **stdout once** and never stored or logged —
+Postgres holds only its SHA-256 hash (ADR-0009), so a lost key is reissued by
+re-running this command, not recovered. Structured logs go to stderr/the log
+sink and carry only the non-secret `key_prefix` (ADR-0011).
+
+Re-running with the same `--slug` reuses the tenant and any domains it already
+has, and issues an additional key.
+"""
+
+import argparse
+import asyncio
+import sys
+
+import structlog
+
+from src.application.tenants import DEFAULT_SCOPES, provision_tenant
+from src.config import Settings
+from src.infrastructure.observability.logging import configure_logging
+from src.infrastructure.postgres.database import create_engine, create_sessionmaker
+
+logger = structlog.get_logger(__name__)
+
+
+def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        prog="python -m src.cli.provision_tenant",
+        description="Create a tenant, issue an API key, and register domains.",
+    )
+    parser.add_argument("--slug", required=True, help="URL-safe tenant identifier, e.g. 'acme'")
+    parser.add_argument("--name", default=None, help="Display name (defaults to --slug)")
+    parser.add_argument("--key-name", default="bootstrap", help="Label for the issued API key")
+    parser.add_argument(
+        "--scopes",
+        default=",".join(DEFAULT_SCOPES),
+        help=f"Comma-separated scopes for the key (default: {','.join(DEFAULT_SCOPES)})",
+    )
+    parser.add_argument(
+        "--domain",
+        action="append",
+        default=[],
+        dest="domains",
+        help="Domain to register; repeatable. Uploads reject an unregistered domain.",
+    )
+    return parser.parse_args(argv)
+
+
+async def run(argv: list[str] | None = None, settings: Settings | None = None) -> int:
+    args = _parse_args(argv)
+    resolved = settings or Settings()
+    configure_logging(resolved.logging, resolved.app)
+
+    scopes = tuple(scope.strip() for scope in args.scopes.split(",") if scope.strip())
+    if not scopes:
+        logger.error("tenant.provision.failed", reason="no_scopes")
+        return 2
+
+    engine = create_engine(resolved.postgres)
+    try:
+        result = await provision_tenant(
+            create_sessionmaker(engine),
+            slug=args.slug,
+            name=args.name,
+            key_name=args.key_name,
+            scopes=scopes,
+            domains=tuple(args.domains),
+        )
+    finally:
+        await engine.dispose()
+
+    # stdout, not the logger: this is the one value the operator must copy, and
+    # it must never reach a log sink (ADR-0011).
+    print(f"tenant_id={result.tenant_id}")
+    print(f"tenant_slug={result.tenant_slug}")
+    print(f"api_key_id={result.api_key_id}")
+    print(f"domains={','.join(result.domains_created + result.domains_existing)}")
+    print(f"api_key={result.api_key}")
+    print("Store the api_key now -- only its hash is persisted and it cannot be shown again.")
+    return 0
+
+
+def main() -> None:
+    sys.exit(asyncio.run(run()))
+
+
+if __name__ == "__main__":
+    main()

src/infrastructure/postgres/repositories/api_keys.py

@@ -1,4 +1,4 @@
-"""API-key lookups (ADR-0008, ADR-0009).
+"""API-key lookups and issuance (ADR-0008, ADR-0009).
 
 Plain functions over an `AsyncSession` the caller owns. No function here
 commits, rolls back, or closes the session (ADR-0012). Secret comparison
@@ -6,6 +6,8 @@ happens in `src/application/auth`, not here — this module only fetches rows
 by their non-secret `key_prefix`.
 """
 
+import uuid
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -15,3 +17,29 @@ from src.infrastructure.postgres.models.api_key import ApiKey
 async def get_by_prefix(session: AsyncSession, key_prefix: str) -> ApiKey | None:
     result = await session.execute(select(ApiKey).where(ApiKey.key_prefix == key_prefix))
     return result.scalar_one_or_none()
+
+
+def create(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    name: str,
+    key_prefix: str,
+    key_hash: str,
+    scopes: list[str],
+    actor_type: str = "backend",
+    created_by: str | None = None,
+) -> ApiKey:
+    """Persist an issued key. The caller hashes the secret; this never sees it."""
+    api_key = ApiKey(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        name=name,
+        key_prefix=key_prefix,
+        key_hash=key_hash,
+        scopes=scopes,
+        actor_type=actor_type,
+        created_by=created_by,
+    )
+    session.add(api_key)
+    return api_key

src/infrastructure/postgres/repositories/tenants.py

@@ -1,4 +1,4 @@
-"""Tenant lookups (ADR-0009).
+"""Tenant lookups and creation (ADR-0009).
 
 Plain functions over an `AsyncSession` the caller owns. No function here
 commits, rolls back, or closes the session (ADR-0012).
@@ -6,6 +6,7 @@ commits, rolls back, or closes the session (ADR-0012).
 
 import uuid
 
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.infrastructure.postgres.models.tenant import Tenant
@@ -13,3 +14,14 @@ from src.infrastructure.postgres.models.tenant import Tenant
 
 async def get_by_id(session: AsyncSession, tenant_id: uuid.UUID) -> Tenant | None:
     return await session.get(Tenant, tenant_id)
+
+
+async def get_by_slug(session: AsyncSession, slug: str) -> Tenant | None:
+    result = await session.execute(select(Tenant).where(Tenant.slug == slug))
+    return result.scalar_one_or_none()
+
+
+def create(session: AsyncSession, *, slug: str, name: str) -> Tenant:
+    tenant = Tenant(id=uuid.uuid4(), slug=slug, name=name)
+    session.add(tenant)
+    return tenant

tests/integration/postgres/test_provisioning.py

@@ -0,0 +1,95 @@
+"""Tenant provisioning against real Postgres (ADR-0009).
+
+The property worth a real database here is the one a fake cannot show: the
+issued key authenticates through the *production* auth path, and the row it
+authenticates against holds no plaintext.
+"""
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.service import resolve_auth_context
+from src.application.domains import ensure_domain_allowed
+from src.application.domains.errors import UnknownDomainError
+from src.application.tenants import provision_tenant
+from src.infrastructure.postgres.models.api_key import ApiKey
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def test_provision_tenant_issues_a_key_that_authenticates(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> None:
+    result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
+
+    auth = await resolve_auth_context(db_sessionmaker, result.api_key)
+
+    assert auth.tenant_id == result.tenant_id
+    assert auth.tenant_slug == "acme"
+    assert auth.api_key_id == result.api_key_id
+    assert "files:write" in auth.scopes
+
+
+async def test_provision_tenant_persists_only_the_key_hash(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """A plaintext key in Postgres would make every later hashing decision moot."""
+    result = await provision_tenant(db_sessionmaker, slug="acme")
+
+    stored = (
+        await db_session.execute(select(ApiKey).where(ApiKey.id == result.api_key_id))
+    ).scalar_one()
+
+    assert result.api_key not in stored.key_hash
+    assert stored.key_hash != result.api_key
+    assert stored.key_prefix == result.api_key_prefix
+    assert result.api_key.startswith(f"sk_{result.api_key_prefix}_")
+
+
+async def test_provision_tenant_registers_domains_so_uploads_are_allowed(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    result = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
+
+    await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="fire")
+    with pytest.raises(UnknownDomainError):
+        await ensure_domain_allowed(db_session, tenant_id=result.tenant_id, domain="life")
+
+    assert result.domains_created == ("fire",)
+
+
+async def test_provision_tenant_rerun_reuses_the_tenant_and_issues_a_new_key(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> None:
+    """Adding a key to a live tenant must not need a different command."""
+    first = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire",))
+
+    second = await provision_tenant(db_sessionmaker, slug="acme", domains=("fire", "life"))
+
+    assert second.tenant_id == first.tenant_id
+    assert second.tenant_created is False
+    assert second.api_key_id != first.api_key_id
+    assert second.domains_created == ("life",)
+    assert second.domains_existing == ("fire",)
+    # Both keys stay valid -- reprovisioning adds a key, it does not rotate one.
+    assert (await resolve_auth_context(db_sessionmaker, first.api_key)).tenant_id == first.tenant_id
+    assert (
+        await resolve_auth_context(db_sessionmaker, second.api_key)
+    ).tenant_id == first.tenant_id
+
+
+async def test_provision_tenant_honours_requested_scopes(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> None:
+    """A key scoped to uploads must not be able to manage the allowlist."""
+    result = await provision_tenant(db_sessionmaker, slug="acme", scopes=("files:write",))
+
+    auth = await resolve_auth_context(db_sessionmaker, result.api_key)
+
+    assert auth.scopes == frozenset({"files:write"})
+    assert not auth.has_scope("domains:write")

tests/unit/cli/test_provision_tenant.py

@@ -0,0 +1,34 @@
+"""Argument parsing for `python -m src.cli.provision_tenant`.
+
+Only the argv -> arguments mapping is covered here; the provisioning behaviour
+itself needs a real database and lives in
+`tests/integration/postgres/test_provisioning.py`.
+"""
+
+import pytest
+
+from src.application.tenants import DEFAULT_SCOPES
+from src.cli.provision_tenant import _parse_args
+
+pytestmark = pytest.mark.unit
+
+
+def test_parse_args_defaults_scopes_and_domains() -> None:
+    args = _parse_args(["--slug", "acme"])
+
+    assert args.slug == "acme"
+    assert args.name is None
+    assert args.key_name == "bootstrap"
+    assert tuple(args.scopes.split(",")) == DEFAULT_SCOPES
+    assert args.domains == []
+
+
+def test_parse_args_collects_repeated_domain_flags() -> None:
+    args = _parse_args(["--slug", "acme", "--domain", "fire", "--domain", "life"])
+
+    assert args.domains == ["fire", "life"]
+
+
+def test_parse_args_requires_a_slug() -> None:
+    with pytest.raises(SystemExit):
+        _parse_args(["--domain", "fire"])