feat(tenant): add tenant_domains allowlist and /v1/domains management API

Why:
- Domain values are denormalized into every Qdrant point payload. Without
  validation, an unregistered or typo'd domain (e.g. "fier" for "fire")
  silently creates a new partition that retrieval never queries — the file
  ends up invisible rather than rejected. Tenants also need independently
  sized domain sets (one may run 14 insurance lines, another 6), which rules
  out an enum.

Changes:
- tenant_domains table (migration 41335d162de8) + repository, unique on
  (tenant_id, domain).
- src/application/domains/: ensure_domain_allowed() is the strict-allowlist
  check now run inside upload_source_file()'s first transaction, before any
  MinIO object, job row, or Qdrant point is written.
- /v1/domains (list/create/patch/disable/enable) gated on its own
  domains:read/domains:write scopes, deliberately separate from files:write
  so an upload key cannot create partitions. domain itself is immutable
  (denormalized into every point payload); only display_name is editable.
  Disable blocks new uploads without touching already-indexed points.

Impact:
- BREAKING: POST /v1/files now rejects any domain without an active
  tenant_domains row (400, unknown_domain). A domain must be created via
  POST /v1/domains before the first upload to it.

src/api/errors.py

@@ -16,6 +16,7 @@ from src.application.auth.errors import (
     MissingScopeError,
     TenantInactiveError,
 )
+from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError
 from src.application.files.errors import FileTooLargeError, InvalidUploadError
 from src.application.ingestion.errors import (
     ChunkLimitExceededError,
@@ -40,6 +41,8 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = (
     (TenantInactiveError, status.HTTP_401_UNAUTHORIZED, "tenant_not_found"),
     (MissingScopeError, status.HTTP_403_FORBIDDEN, "missing_scope"),
     (InvalidUploadError, status.HTTP_400_BAD_REQUEST, "validation_error"),
+    (UnknownDomainError, status.HTTP_400_BAD_REQUEST, "unknown_domain"),
+    (DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"),
     (DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"),
     (UnsupportedSourceTypeError, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "unsupported_media_type"),
     (FileTooLargeError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"),

src/api/router.py

@@ -1,6 +1,8 @@
 from fastapi import APIRouter
 
+from src.api.routers.domains import router as domains_router
 from src.api.routers.files import router as files_router
 
 router = APIRouter()
+router.include_router(domains_router)
 router.include_router(files_router)

src/api/routers/domains.py

@@ -0,0 +1,110 @@
+"""`/v1/domains` (ADR-0008, ADR-0009).
+
+The management surface for a tenant's domain allowlist, used by the calling
+backend rather than by an operator with a psql prompt.
+
+Gated on `domains:read`/`domains:write`, deliberately **not** on `files:write`:
+if an upload key could create domains, the allowlist would no longer prevent a
+typo'd `domain` from creating a Qdrant partition, which is its only purpose.
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.api.dependencies.auth import require_scope
+from src.api.schemas.domains import (
+    CreateDomainRequest,
+    DomainListResponse,
+    DomainResponse,
+    UpdateDomainRequest,
+)
+from src.application.auth.context import AuthContext
+from src.application.domains import (
+    create_domain,
+    list_domains,
+    set_domain_status,
+    update_domain,
+)
+from src.bootstrap.dependencies import get_sessionmaker
+
+router = APIRouter(prefix="/domains", tags=["domains"])
+
+_RequireDomainsRead = Annotated[AuthContext, Depends(require_scope("domains:read"))]
+_RequireDomainsWrite = Annotated[AuthContext, Depends(require_scope("domains:write"))]
+_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
+
+
+@router.get("")
+async def list_tenant_domains(
+    auth: _RequireDomainsRead,
+    sessionmaker: _SessionmakerDep,
+    include_disabled: bool = False,
+) -> DomainListResponse:
+    results = await list_domains(
+        sessionmaker, tenant_id=auth.tenant_id, include_disabled=include_disabled
+    )
+    return DomainListResponse(domains=[DomainResponse.from_result(item) for item in results])
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+async def create_tenant_domain(
+    request: CreateDomainRequest,
+    auth: _RequireDomainsWrite,
+    sessionmaker: _SessionmakerDep,
+) -> DomainResponse:
+    result = await create_domain(
+        sessionmaker,
+        tenant_id=auth.tenant_id,
+        domain=request.domain,
+        display_name=request.display_name,
+        metadata=request.metadata,
+    )
+    return DomainResponse.from_result(result)
+
+
+@router.patch("/{domain}")
+async def update_tenant_domain(
+    domain: str,
+    request: UpdateDomainRequest,
+    auth: _RequireDomainsWrite,
+    sessionmaker: _SessionmakerDep,
+) -> DomainResponse:
+    result = await update_domain(
+        sessionmaker,
+        tenant_id=auth.tenant_id,
+        domain=domain,
+        display_name=request.display_name,
+    )
+    return DomainResponse.from_result(result)
+
+
+@router.delete("/{domain}")
+async def disable_tenant_domain(
+    domain: str,
+    auth: _RequireDomainsWrite,
+    sessionmaker: _SessionmakerDep,
+) -> DomainResponse:
+    """Disable, not delete.
+
+    Blocks new uploads and drops the domain from pickers while leaving the
+    points already indexed under it intact and retrievable. Actually removing
+    them needs the tenant-erasure workflow plan 001 defers.
+    """
+    result = await set_domain_status(
+        sessionmaker, tenant_id=auth.tenant_id, domain=domain, status="disabled"
+    )
+    return DomainResponse.from_result(result)
+
+
+@router.post("/{domain}/enable")
+async def enable_tenant_domain(
+    domain: str,
+    auth: _RequireDomainsWrite,
+    sessionmaker: _SessionmakerDep,
+) -> DomainResponse:
+    result = await set_domain_status(
+        sessionmaker, tenant_id=auth.tenant_id, domain=domain, status="active"
+    )
+    return DomainResponse.from_result(result)

src/api/schemas/domains.py

@@ -0,0 +1,63 @@
+"""Public request/response models for `/v1/domains` (ADR-0008, ADR-0009).
+
+`tenant_id` appears in none of these: it comes from the authenticated key, and
+accepting it from a body would break the isolation boundary (ADR-0002).
+"""
+
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel, Field, field_validator
+
+from src.application.domains.models import DomainResult
+
+# Lowercase alphanumerics plus - and _; the key is embedded in every Qdrant
+# payload and filtered on as a keyword, so it stays boring on purpose.
+_DOMAIN_PATTERN = r"^[a-z0-9][a-z0-9_-]*$"
+
+
+class DomainResponse(BaseModel):
+    id: uuid.UUID
+    domain: str
+    display_name: str
+    status: str
+    metadata: dict[str, object]
+    created_at: datetime
+    updated_at: datetime
+
+    @classmethod
+    def from_result(cls, result: DomainResult) -> "DomainResponse":
+        return cls(
+            id=result.id,
+            domain=result.domain,
+            display_name=result.display_name,
+            status=result.status,
+            metadata=result.metadata,
+            created_at=result.created_at,
+            updated_at=result.updated_at,
+        )
+
+
+class DomainListResponse(BaseModel):
+    domains: list[DomainResponse]
+
+
+class CreateDomainRequest(BaseModel):
+    domain: str = Field(min_length=1, max_length=80, pattern=_DOMAIN_PATTERN)
+    display_name: str = Field(min_length=1, max_length=200)
+    metadata: dict[str, object] = Field(default_factory=dict)
+
+    @field_validator("domain")
+    @classmethod
+    def _normalize(cls, value: str) -> str:
+        return value.strip()
+
+
+class UpdateDomainRequest(BaseModel):
+    """`domain` is absent by design — the key is immutable.
+
+    It is denormalized into every point payload and into `source_files`, so
+    renaming it is a migration rather than an edit (ADR-0009).
+    """
+
+    display_name: str = Field(min_length=1, max_length=200)

src/application/domains/__init__.py

@@ -0,0 +1,27 @@
+"""Tenant-domain management and the upload-time allowlist check (ADR-0009)."""
+
+from src.application.domains.errors import (
+    DomainAlreadyExistsError,
+    DomainsError,
+    UnknownDomainError,
+)
+from src.application.domains.models import DomainResult
+from src.application.domains.service import (
+    create_domain,
+    ensure_domain_allowed,
+    list_domains,
+    set_domain_status,
+    update_domain,
+)
+
+__all__ = [
+    "DomainAlreadyExistsError",
+    "DomainResult",
+    "DomainsError",
+    "UnknownDomainError",
+    "create_domain",
+    "ensure_domain_allowed",
+    "list_domains",
+    "set_domain_status",
+    "update_domain",
+]

src/application/domains/errors.py

@@ -0,0 +1,21 @@
+"""Domain-management failures (ADR-0009). No HTTP knowledge here —
+`src/api/errors.py` maps these to status codes.
+"""
+
+
+class DomainsError(Exception):
+    """Base class for tenant-domain failures."""
+
+
+class UnknownDomainError(DomainsError):
+    """The upload named a domain the tenant has not registered, or one that is
+    disabled. Maps to `400`.
+
+    Rejecting is the whole point: an unrecognized `domain` would otherwise
+    create a new Qdrant partition silently, and a file in a partition nothing
+    queries is invisible rather than failed (ADR-0009).
+    """
+
+
+class DomainAlreadyExistsError(DomainsError):
+    """The tenant already has a domain with this key. Maps to `409`."""

src/application/domains/models.py

@@ -0,0 +1,16 @@
+"""Transport-agnostic results for the domain-management service."""
+
+import uuid
+from dataclasses import dataclass
+from datetime import datetime
+
+
+@dataclass(frozen=True)
+class DomainResult:
+    id: uuid.UUID
+    domain: str
+    display_name: str
+    status: str
+    metadata: dict[str, object]
+    created_at: datetime
+    updated_at: datetime

src/application/domains/service.py

@@ -0,0 +1,142 @@
+"""Tenant-domain management (ADR-0009).
+
+A tenant's domain set is per-tenant and varies in size — one may run 14
+insurance lines, another 6 — so it is data, not an enum.
+
+`ensure_domain_allowed` is the reason this package exists: it is the strict
+allowlist check the upload path runs before anything is written. Everything
+else here is the management surface the calling backend uses to populate that
+allowlist, under its own `domains:write` scope so an upload key cannot create
+partitions.
+
+`tenant_id` is always a required parameter taken from `AuthContext`, never from
+a request body (ADR-0002).
+"""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError
+from src.application.domains.models import DomainResult
+from src.infrastructure.postgres.models.tenant_domain import TenantDomain
+from src.infrastructure.postgres.repositories import tenant_domains as repo
+
+
+async def _flush_and_refresh(session: AsyncSession, tenant_domain: TenantDomain) -> None:
+    """Materialize server-generated columns before the row leaves the session.
+
+    `updated_at` is `onupdate=func.now()`, so after an UPDATE its value lives in
+    the database, not in the instance. Reading it later would trigger a lazy
+    load outside any greenlet context (`MissingGreenlet`), so it is fetched here
+    while the session is still open.
+    """
+    await session.flush()
+    await session.refresh(tenant_domain)
+
+
+def _to_result(tenant_domain: TenantDomain) -> DomainResult:
+    return DomainResult(
+        id=tenant_domain.id,
+        domain=tenant_domain.domain,
+        display_name=tenant_domain.display_name,
+        status=tenant_domain.status,
+        metadata=tenant_domain.metadata_,
+        created_at=tenant_domain.created_at,
+        updated_at=tenant_domain.updated_at,
+    )
+
+
+async def ensure_domain_allowed(
+    session: AsyncSession, *, tenant_id: uuid.UUID, domain: str
+) -> None:
+    """Raise `UnknownDomainError` unless the tenant has this domain active.
+
+    Takes a session rather than a sessionmaker: the upload path calls this
+    inside its existing txn A, so the check costs no extra connection and
+    cannot pass and then go stale before the row is written.
+    """
+    tenant_domain = await repo.get(session, tenant_id=tenant_id, domain=domain)
+    if tenant_domain is None:
+        raise UnknownDomainError(
+            f"domain '{domain}' is not registered for this tenant; "
+            f"create it via POST /v1/domains before uploading to it"
+        )
+    if tenant_domain.status != "active":
+        raise UnknownDomainError(f"domain '{domain}' is disabled for this tenant")
+
+
+async def list_domains(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    include_disabled: bool = False,
+) -> list[DomainResult]:
+    async with sessionmaker() as session:
+        found = await repo.list_for_tenant(
+            session, tenant_id=tenant_id, include_disabled=include_disabled
+        )
+    return [_to_result(item) for item in found]
+
+
+async def create_domain(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    display_name: str,
+    metadata: dict[str, object] | None = None,
+) -> DomainResult:
+    async with sessionmaker() as session:
+        if await repo.get(session, tenant_id=tenant_id, domain=domain) is not None:
+            raise DomainAlreadyExistsError(f"domain '{domain}' already exists for this tenant")
+        created = repo.create(
+            session,
+            tenant_id=tenant_id,
+            domain=domain,
+            display_name=display_name,
+            metadata=metadata,
+        )
+        await session.commit()
+        return _to_result(created)
+
+
+async def update_domain(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    display_name: str,
+) -> DomainResult:
+    """Only the label is mutable — see `repo.update_display_name`."""
+    async with sessionmaker() as session:
+        found = await repo.get(session, tenant_id=tenant_id, domain=domain)
+        if found is None:
+            raise UnknownDomainError(f"domain '{domain}' is not registered for this tenant")
+        repo.update_display_name(found, display_name=display_name)
+        await _flush_and_refresh(session, found)
+        await session.commit()
+        return _to_result(found)
+
+
+async def set_domain_status(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    status: str,
+) -> DomainResult:
+    """Disable or re-enable a domain.
+
+    Disabling blocks new uploads and hides the domain from pickers. It does not
+    touch the points already indexed under it — removing those needs the
+    tenant-erasure workflow plan 001 defers.
+    """
+    async with sessionmaker() as session:
+        found = await repo.get(session, tenant_id=tenant_id, domain=domain)
+        if found is None:
+            raise UnknownDomainError(f"domain '{domain}' is not registered for this tenant")
+        repo.set_status(found, status=status)
+        await _flush_and_refresh(session, found)
+        await session.commit()
+        return _to_result(found)

src/application/files/upload.py

@@ -30,6 +30,7 @@ from anyio import CapacityLimiter, Semaphore, fail_after, to_thread
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from src.application.auth.context import AuthContext
+from src.application.domains import ensure_domain_allowed
 from src.application.files.errors import InvalidUploadError
 from src.application.files.models import UploadResult
 from src.application.files.storage_keys import source_file_object_key
@@ -118,6 +119,12 @@ async def upload_source_file(
 
     async with acquire_ingestion_slot(concurrency_limiter):
         async with sessionmaker() as session:
+            # Strict allowlist, checked inside txn A before anything is written
+            # (ADR-0009). An unregistered domain would otherwise create a new
+            # Qdrant partition silently, leaving the file invisible to
+            # retrieval rather than failing.
+            await ensure_domain_allowed(session, tenant_id=auth.tenant_id, domain=domain)
+
             existing = await source_files_repo.find_active_by_content_hash(
                 session,
                 tenant_id=auth.tenant_id,

src/infrastructure/postgres/models/__init__.py

@@ -4,6 +4,7 @@ from src.infrastructure.postgres.models.ingestion_job import IngestionJob
 from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent
 from src.infrastructure.postgres.models.source_file import SourceFile
 from src.infrastructure.postgres.models.tenant import Tenant
+from src.infrastructure.postgres.models.tenant_domain import TenantDomain
 
 __all__ = [
     "ApiKey",
@@ -12,4 +13,5 @@ __all__ = [
     "IngestionJobEvent",
     "SourceFile",
     "Tenant",
+    "TenantDomain",
 ]

src/infrastructure/postgres/models/tenant_domain.py

@@ -0,0 +1,52 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import CheckConstraint, DateTime, ForeignKey, String, UniqueConstraint, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from src.infrastructure.postgres.models.base import Base
+
+TENANT_DOMAIN_STATUSES = ("active", "disabled")
+
+
+class TenantDomain(Base):
+    """A domain a tenant is allowed to ingest into (ADR-0009).
+
+    Tenants do not share a domain list — one may run 14 insurance lines and
+    another 6 — so this is a per-tenant table rather than an enum or a global
+    lookup.
+
+    Its purpose is to stop an arbitrary caller-supplied `domain` from silently
+    creating a new Qdrant partition. `domain` is denormalized into every point's
+    payload and into `source_files`, and a typo like `fier` for `fire` produces
+    no error anywhere: the file indexes into a partition retrieval never queries,
+    so it is invisible rather than failed.
+
+    `domain` is the immutable key. Renaming it would mean rewriting every point
+    payload that carries it, which is a migration, not an edit — `display_name`
+    is the mutable human-facing label instead.
+    """
+
+    __tablename__ = "tenant_domains"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "domain", name="uq_tenant_domains_tenant_id_domain"),
+        CheckConstraint(f"status IN {TENANT_DOMAIN_STATUSES}", name="ck_tenant_domains_status"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
+    tenant_id: Mapped[uuid.UUID] = mapped_column(
+        ForeignKey("tenants.id", ondelete="CASCADE"), index=True
+    )
+    domain: Mapped[str] = mapped_column(String(80))
+    display_name: Mapped[str] = mapped_column(String(200))
+    status: Mapped[str] = mapped_column(String(20), default="active", server_default="active")
+    metadata_: Mapped[dict[str, object]] = mapped_column(
+        "metadata", JSONB, default=dict, server_default="{}"
+    )
+
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
+    )
+    disabled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)

src/infrastructure/postgres/repositories/tenant_domains.py

@@ -0,0 +1,73 @@
+"""`tenant_domains` persistence (ADR-0009).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012). Every read and write is
+tenant-scoped by a required `tenant_id` argument, so a missing filter is a
+signature error rather than a cross-tenant leak.
+"""
+
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.tenant_domain import TenantDomain
+
+
+async def get(session: AsyncSession, *, tenant_id: uuid.UUID, domain: str) -> TenantDomain | None:
+    result = await session.execute(
+        select(TenantDomain).where(
+            TenantDomain.tenant_id == tenant_id, TenantDomain.domain == domain
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+async def list_for_tenant(
+    session: AsyncSession, *, tenant_id: uuid.UUID, include_disabled: bool = False
+) -> list[TenantDomain]:
+    statement = select(TenantDomain).where(TenantDomain.tenant_id == tenant_id)
+    if not include_disabled:
+        statement = statement.where(TenantDomain.status == "active")
+    result = await session.execute(statement.order_by(TenantDomain.domain))
+    return list(result.scalars().all())
+
+
+def create(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    domain: str,
+    display_name: str,
+    metadata: dict[str, object] | None = None,
+) -> TenantDomain:
+    tenant_domain = TenantDomain(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        domain=domain,
+        display_name=display_name,
+        metadata_=metadata or {},
+    )
+    session.add(tenant_domain)
+    return tenant_domain
+
+
+def update_display_name(tenant_domain: TenantDomain, *, display_name: str) -> TenantDomain:
+    """`domain` itself is deliberately not updatable.
+
+    It is denormalized into every Qdrant point payload and into `source_files`,
+    so changing the key would mean rewriting all of them — a migration, not an
+    edit. The label is what callers actually want to change.
+    """
+    tenant_domain.display_name = display_name
+    return tenant_domain
+
+
+def set_status(tenant_domain: TenantDomain, *, status: str) -> TenantDomain:
+    """Disable/re-enable a domain. Existing points are untouched either way —
+    disabling blocks new uploads, it is not a delete (ADR-0002).
+    """
+    tenant_domain.status = status
+    tenant_domain.disabled_at = datetime.now(UTC) if status == "disabled" else None
+    return tenant_domain