chore(claude): add Explore subagent definition

Why:
- Code under test (auth resolution, the two-phase upload) opens more than
  one session per operation; the existing fixture only exposed one
  rolled-back session.

Changes:
- Add a db_sessionmaker fixture sharing one outer transaction.
- Pin loop_scope="session" -- without it, a second async test against the
  session-scoped Postgres container fails with "Event loop is closed."

.claude/agents/Explore.md

@@ -0,0 +1,8 @@
+---
+name: Explore
+description: Fast, read-only codebase search
+model: sonnet
+effort: low
+tools: Read, Grep, Glob, Bash, WebFetch, WebSearch
+maxTurns: 20
+---

.env.example

@@ -9,7 +9,6 @@
 
 # Application
 APP_ENV=local
-APP_MAX_UPLOAD_SIZE_MB=25
 APP_READINESS_CHECK_TIMEOUT_SECONDS=2.0
 
 # Logging
@@ -37,6 +36,7 @@ MINIO_BUCKET=chatbot-source-files
 INGESTION_MAX_CONCURRENCY=4
 INGESTION_THREAD_POOL_SIZE=8
 INGESTION_TIMEOUT_SECONDS=120.0
+INGESTION_MAX_UPLOAD_SIZE_MB=25
 INGESTION_MAX_CHUNKS_PER_FILE=5000
 INGESTION_EMBED_BATCH_SIZE=128
 INGESTION_EMBED_CONCURRENCY=4

CLAUDE.md

@@ -113,6 +113,27 @@ MinIO/Qdrant/SQLAlchemy client-construction code. Use ports only for
 external side effects/persistence — not around pure local functions.
 (ADR-0015)
 
+### Prefer deep modules over shallow ones
+
+When a package exposes several small pure functions that a caller must
+compose correctly every time (right dispatch, right order, right
+thread/async offload), give it one entry point that owns that composition,
+and keep the small functions internal — exported only where their own unit
+tests need them. A shallow interface (one whose surface is nearly as complex
+as its implementation) pushes a correctness obligation onto every call site;
+a deep one absorbs it once. Apply the deletion test when unsure: if deleting
+the wrapper would concentrate the composition logic back into every caller
+rather than just relocate it, the wrapper is worth having.
+
+Worked example: `src/application/ingestion/` exposes `parse_and_chunk_document`
+as its only caller-facing entry point. It dispatches on source type and owns
+the `anyio.to_thread.run_sync` + `CapacityLimiter` offload ADR-0017 requires;
+`parse_docx`/`parse_csv`/`parse_xlsx`/`chunk_document` stay in the package,
+exported mainly for their own tests, not for outside callers to reach for
+directly. Follow this pattern in `application/` as new packages are added
+there — `points/`, `retrieval/`, `threads/` — rather than exposing their
+internals as the primary surface.
+
 ### Resource lifetime rules (ADR-0012)
 
 - Application-lifetime objects (SQLAlchemy engine/sessionmaker, Qdrant client,

src/api/dependencies/auth.py

@@ -0,0 +1,38 @@
+"""Auth dependencies (ADR-0008): resolve `AuthContext` from a bearer token,
+then gate routes on scope.
+"""
+
+from collections.abc import Awaitable, Callable
+from typing import Annotated
+
+from fastapi import Depends
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.context import AuthContext
+from src.application.auth.errors import InvalidApiKeyError, MissingScopeError
+from src.application.auth.service import resolve_auth_context
+from src.bootstrap.dependencies import get_sessionmaker
+
+_bearer_scheme = HTTPBearer(auto_error=False)
+
+
+async def get_auth_context(
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)],
+    sessionmaker: Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)],
+) -> AuthContext:
+    if credentials is None:
+        raise InvalidApiKeyError("missing Authorization header")
+    return await resolve_auth_context(sessionmaker, credentials.credentials)
+
+
+AuthContextDep = Annotated[AuthContext, Depends(get_auth_context)]
+
+
+def require_scope(scope: str) -> Callable[[AuthContext], Awaitable[AuthContext]]:
+    async def _dependency(auth: AuthContextDep) -> AuthContext:
+        if not auth.has_scope(scope):
+            raise MissingScopeError(f"missing required scope '{scope}'")
+        return auth
+
+    return _dependency

src/api/errors.py

@@ -0,0 +1,97 @@
+"""Maps application exceptions to the ADR-0008 error envelope.
+
+This is the single place that knows the exception-type -> status-code
+mapping; application/infrastructure code never imports FastAPI or raises
+`HTTPException` (ADR-0015).
+"""
+
+import structlog
+from fastapi import FastAPI, Request, status
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from starlette.exceptions import HTTPException as StarletteHTTPException
+
+from src.application.auth.errors import (
+    InvalidApiKeyError,
+    MissingScopeError,
+    TenantInactiveError,
+)
+from src.application.files.errors import FileTooLargeError, InvalidUploadError
+from src.application.ingestion.errors import DocumentParseError, UnsupportedSourceTypeError
+
+logger = structlog.get_logger(__name__)
+
+# (exception type, status code, stable error code)
+_MAPPING: tuple[tuple[type[Exception], int, str], ...] = (
+    (InvalidApiKeyError, status.HTTP_401_UNAUTHORIZED, "invalid_api_key"),
+    (TenantInactiveError, status.HTTP_401_UNAUTHORIZED, "tenant_not_found"),
+    (MissingScopeError, status.HTTP_403_FORBIDDEN, "missing_scope"),
+    (InvalidUploadError, status.HTTP_400_BAD_REQUEST, "validation_error"),
+    (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"),
+)
+
+
+def _request_id(request: Request) -> str | None:
+    return getattr(request.state, "request_id", None)
+
+
+def _envelope(
+    code: str, message: str, request_id: str | None, details: dict[str, object] | None = None
+) -> dict[str, object]:
+    return {
+        "error": {
+            "code": code,
+            "message": message,
+            "details": details or {},
+            "request_id": request_id,
+        }
+    }
+
+
+def register_exception_handlers(app: FastAPI) -> None:
+    for exc_type, status_code, error_code in _MAPPING:
+
+        def _handler(
+            request: Request,
+            exc: Exception,
+            status_code: int = status_code,
+            error_code: str = error_code,
+        ) -> JSONResponse:
+            return JSONResponse(
+                status_code=status_code,
+                content=_envelope(error_code, str(exc), _request_id(request)),
+            )
+
+        app.add_exception_handler(exc_type, _handler)
+
+    @app.exception_handler(RequestValidationError)
+    def _validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
+        return JSONResponse(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            content=_envelope(
+                "validation_error",
+                "request validation failed",
+                _request_id(request),
+                details={"errors": exc.errors()},
+            ),
+        )
+
+    @app.exception_handler(StarletteHTTPException)
+    def _http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse:
+        code = "not_found" if exc.status_code == status.HTTP_404_NOT_FOUND else "http_error"
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=_envelope(code, str(exc.detail), _request_id(request)),
+        )
+
+    @app.exception_handler(Exception)
+    def _unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
+        logger.exception("api.unhandled_exception", path=request.url.path)
+        return JSONResponse(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            content=_envelope(
+                "internal_error", "an unexpected error occurred", _request_id(request)
+            ),
+        )

src/api/middleware.py

@@ -0,0 +1,38 @@
+"""Per-request correlation id (ADR-0008, ADR-0011).
+
+Every request gets a `request_id`: reused from an incoming `X-Request-Id` if
+the caller supplied one, otherwise generated. It is bound into structlog's
+contextvars so every log line emitted while handling the request carries it,
+stored on `request.state` for exception handlers, and echoed back in the
+response header.
+"""
+
+import uuid
+from collections.abc import Awaitable, Callable
+from typing import override
+
+import structlog
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import Response
+
+_HEADER = "X-Request-Id"
+
+
+class RequestIdMiddleware(BaseHTTPMiddleware):
+    @override
+    async def dispatch(
+        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
+    ) -> Response:
+        request_id = request.headers.get(_HEADER) or str(uuid.uuid4())
+        request.state.request_id = request_id
+
+        structlog.contextvars.clear_contextvars()
+        structlog.contextvars.bind_contextvars(request_id=request_id)
+        try:
+            response = await call_next(request)
+        finally:
+            structlog.contextvars.clear_contextvars()
+
+        response.headers[_HEADER] = request_id
+        return response

src/api/router.py

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

src/api/routers/files.py

@@ -0,0 +1,74 @@
+"""`POST /v1/files`, `GET /v1/files/{file_id}` (ADR-0008).
+
+Routes adapt HTTP to `application/files` calls; they do not parse, hash,
+touch MinIO/Qdrant, or otherwise carry ingestion business logic (ADR-0015).
+"""
+
+import uuid
+from typing import Annotated
+
+from anyio import CapacityLimiter
+from fastapi import APIRouter, Depends, Form, HTTPException, Response, UploadFile, status
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.api.dependencies.auth import require_scope
+from src.api.schemas.files import FileStatusResponse, FileUploadResponse
+from src.application.auth.context import AuthContext
+from src.application.files.status import get_file_status
+from src.application.files.upload import upload_source_file
+from src.application.ports.object_storage import ObjectStorage
+from src.bootstrap.dependencies import (
+    get_ingestion_limiter,
+    get_object_storage,
+    get_sessionmaker,
+    get_settings,
+)
+from src.config import Settings
+
+router = APIRouter(prefix="/files", tags=["files"])
+
+_RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))]
+_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)]
+_ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)]
+_SettingsDep = Annotated[Settings, Depends(get_settings)]
+_IngestionLimiterDep = Annotated[CapacityLimiter, Depends(get_ingestion_limiter)]
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+async def upload_file(
+    response: Response,
+    file: UploadFile,
+    domain: Annotated[str, Form()],
+    auth: _RequireFilesWrite,
+    sessionmaker: _SessionmakerDep,
+    storage: _ObjectStorageDep,
+    settings: _SettingsDep,
+    limiter: _IngestionLimiterDep,
+) -> FileUploadResponse:
+    data = await file.read()
+    result = await upload_source_file(
+        sessionmaker=sessionmaker,
+        storage=storage,
+        auth=auth,
+        domain=domain,
+        filename=file.filename or "",
+        data=data,
+        max_upload_size_bytes=settings.ingestion.max_upload_size_bytes,
+        chunking_strategy=settings.chunking.strategy,
+        validation_limiter=limiter,
+    )
+    if not result.is_new_attempt:
+        response.status_code = status.HTTP_200_OK
+    return FileUploadResponse.from_result(result)
+
+
+@router.get("/{file_id}")
+async def get_file(
+    file_id: uuid.UUID,
+    auth: _RequireFilesWrite,
+    sessionmaker: _SessionmakerDep,
+) -> FileStatusResponse:
+    result = await get_file_status(sessionmaker, tenant_id=auth.tenant_id, source_file_id=file_id)
+    if result is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="file not found")
+    return FileStatusResponse.from_result(result)

src/api/schemas/errors.py

@@ -0,0 +1,16 @@
+"""The ADR-0008 error envelope."""
+
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class ErrorDetail(BaseModel):
+    code: str
+    message: str
+    details: dict[str, Any] = {}
+    request_id: str | None = None
+
+
+class ErrorResponse(BaseModel):
+    error: ErrorDetail

src/api/schemas/files.py

@@ -0,0 +1,50 @@
+"""Public request/response models for `/v1/files` (ADR-0008).
+
+Separate from the SQLAlchemy ORM models and the `application/files` domain
+dataclasses (ADR-0015): this is the shape callers see.
+"""
+
+import uuid
+
+from pydantic import BaseModel
+
+from src.application.files.models import UploadResult
+from src.application.files.status import FileStatusResult
+
+
+class FileUploadResponse(BaseModel):
+    file_id: uuid.UUID
+    ingestion_job_id: uuid.UUID
+    status: str
+    chunks_indexed: int
+
+    @classmethod
+    def from_result(cls, result: UploadResult) -> "FileUploadResponse":
+        return cls(
+            file_id=result.file_id,
+            ingestion_job_id=result.ingestion_job_id,
+            status=result.status,
+            chunks_indexed=result.chunks_indexed,
+        )
+
+
+class FileStatusResponse(BaseModel):
+    file_id: uuid.UUID
+    source_filename: str
+    domain: str
+    status: str
+    ingestion_job_id: uuid.UUID | None
+    ingestion_status: str | None
+    chunks_indexed: int
+
+    @classmethod
+    def from_result(cls, result: FileStatusResult) -> "FileStatusResponse":
+        return cls(
+            file_id=result.file_id,
+            source_filename=result.source_filename,
+            domain=result.domain,
+            status=result.status,
+            ingestion_job_id=result.ingestion_job_id,
+            ingestion_status=result.ingestion_status,
+            chunks_indexed=result.chunks_indexed,
+        )

src/application/auth/__init__.py

@@ -0,0 +1,29 @@
+"""API-key authentication and tenant resolution (ADR-0008).
+
+`resolve_auth_context` is the entry point: it takes a bearer token and
+returns a trusted `AuthContext`. Everything downstream of the FastAPI
+boundary receives `tenant_id` only through that context — never from a
+request body, query string, or object metadata.
+"""
+
+from src.application.auth.context import AuthContext
+from src.application.auth.errors import (
+    AuthError,
+    InvalidApiKeyError,
+    MissingScopeError,
+    TenantInactiveError,
+)
+from src.application.auth.keys import generate_api_key, hash_secret, verify_secret
+from src.application.auth.service import resolve_auth_context
+
+__all__ = [
+    "AuthContext",
+    "AuthError",
+    "InvalidApiKeyError",
+    "MissingScopeError",
+    "TenantInactiveError",
+    "generate_api_key",
+    "hash_secret",
+    "resolve_auth_context",
+    "verify_secret",
+]

src/application/auth/context.py

@@ -0,0 +1,16 @@
+"""The trusted request-scoped auth/tenant context (ADR-0008)."""
+
+import uuid
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class AuthContext:
+    tenant_id: uuid.UUID
+    tenant_slug: str
+    api_key_id: uuid.UUID
+    scopes: frozenset[str]
+    actor_type: str
+
+    def has_scope(self, scope: str) -> bool:
+        return scope in self.scopes or "admin" in self.scopes

src/application/auth/errors.py

@@ -0,0 +1,22 @@
+"""Auth failures (ADR-0008). No HTTP knowledge here — `src/api/errors.py` maps
+these to status codes.
+"""
+
+
+class AuthError(Exception):
+    """Base class for auth failures."""
+
+
+class InvalidApiKeyError(AuthError):
+    """The bearer token is missing, malformed, unknown, revoked, or expired.
+
+    Maps to `401`.
+    """
+
+
+class TenantInactiveError(AuthError):
+    """The key's tenant is suspended or deleted. Maps to `401`."""
+
+
+class MissingScopeError(AuthError):
+    """The key is valid but lacks a scope the route requires. Maps to `403`."""

src/application/auth/keys.py

@@ -0,0 +1,39 @@
+"""API-key generation and hashing (ADR-0008, ADR-0009).
+
+Keys are `sk_{prefix}_{secret}`. `prefix` is non-secret and indexed
+(`api_keys.key_prefix`); `secret` is 256 bits of `secrets.token_urlsafe`
+entropy, stored only as a SHA-256 hash. A random 256-bit secret does not
+benefit from a slow password-hashing KDF the way a human-chosen password
+does — the cost that defends against dictionary/brute-force guessing over a
+low-entropy input has nothing to defend here, and would only tax every
+request. Comparison is constant-time to avoid a hash-timing oracle.
+"""
+
+import hashlib
+import hmac
+import secrets
+
+_PREFIX_LENGTH = 16
+
+
+def generate_api_key() -> tuple[str, str, str]:
+    """Return `(key_prefix, secret, full_key)` for a newly issued key."""
+    key_prefix = secrets.token_hex(_PREFIX_LENGTH // 2)
+    secret = secrets.token_urlsafe(32)
+    return key_prefix, secret, f"sk_{key_prefix}_{secret}"
+
+
+def parse_api_key(full_key: str) -> tuple[str, str] | None:
+    """Return `(key_prefix, secret)`, or `None` if the token is malformed."""
+    parts = full_key.split("_", 2)
+    if len(parts) != 3 or parts[0] != "sk" or not parts[1] or not parts[2]:
+        return None
+    return parts[1], parts[2]
+
+
+def hash_secret(secret: str) -> str:
+    return hashlib.sha256(secret.encode("utf-8")).hexdigest()
+
+
+def verify_secret(secret: str, key_hash: str) -> bool:
+    return hmac.compare_digest(hash_secret(secret), key_hash)

src/application/auth/service.py

@@ -0,0 +1,47 @@
+"""Resolve a bearer token to a trusted `AuthContext` (ADR-0008).
+
+This opens and releases its own session rather than borrowing a
+request-scoped one, so auth resolution never pins a pool connection across
+the rest of the request — including the ADR-0017 ingestion work phase, which
+must run with no Postgres session held open at all.
+"""
+
+from datetime import UTC, datetime
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.context import AuthContext
+from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError
+from src.application.auth.keys import parse_api_key, verify_secret
+from src.infrastructure.postgres.repositories import api_keys as api_keys_repo
+from src.infrastructure.postgres.repositories import tenants as tenants_repo
+
+
+async def resolve_auth_context(
+    sessionmaker: async_sessionmaker[AsyncSession], bearer_token: str
+) -> AuthContext:
+    parsed = parse_api_key(bearer_token)
+    if parsed is None:
+        raise InvalidApiKeyError("malformed API key")
+    key_prefix, secret = parsed
+
+    async with sessionmaker() as session:
+        api_key = await api_keys_repo.get_by_prefix(session, key_prefix)
+        if api_key is None or not verify_secret(secret, api_key.key_hash):
+            raise InvalidApiKeyError("unknown API key")
+        if api_key.status != "active":
+            raise InvalidApiKeyError(f"API key is {api_key.status}")
+        if api_key.expires_at is not None and api_key.expires_at <= datetime.now(UTC):
+            raise InvalidApiKeyError("API key has expired")
+
+        tenant = await tenants_repo.get_by_id(session, api_key.tenant_id)
+        if tenant is None or tenant.status != "active":
+            raise TenantInactiveError("tenant is not active")
+
+        return AuthContext(
+            tenant_id=tenant.id,
+            tenant_slug=tenant.slug,
+            api_key_id=api_key.id,
+            scopes=frozenset(api_key.scopes),
+            actor_type=api_key.actor_type,
+        )

src/application/files/__init__.py

@@ -0,0 +1,19 @@
+"""Source-file upload and status use cases (ADR-0008, ADR-0009, ADR-0017)."""
+
+from src.application.files.errors import FilesError, FileTooLargeError, InvalidUploadError
+from src.application.files.models import UploadResult, ValidatedUpload
+from src.application.files.status import FileStatusResult, get_file_status
+from src.application.files.upload import upload_source_file
+from src.application.files.validation import validate_and_hash_upload
+
+__all__ = [
+    "FileStatusResult",
+    "FileTooLargeError",
+    "FilesError",
+    "InvalidUploadError",
+    "UploadResult",
+    "ValidatedUpload",
+    "get_file_status",
+    "upload_source_file",
+    "validate_and_hash_upload",
+]

src/application/files/errors.py

@@ -0,0 +1,17 @@
+"""Upload-validation failures (ADR-0008). No HTTP knowledge here —
+`src/api/errors.py` maps these to status codes.
+"""
+
+
+class FilesError(Exception):
+    """Base class for file-upload failures."""
+
+
+class InvalidUploadError(FilesError):
+    """Missing domain, empty file, or content that doesn't match its
+    declared extension. Maps to `400`.
+    """
+
+
+class FileTooLargeError(FilesError):
+    """The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`."""

src/application/files/models.py

@@ -0,0 +1,26 @@
+"""Domain models for the upload use case (ADR-0008, ADR-0009)."""
+
+import uuid
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ValidatedUpload:
+    """The result of extension/content validation, before any I/O."""
+
+    source_type: str
+    content_type: str
+    content_sha256: str
+
+
+@dataclass(frozen=True)
+class UploadResult:
+    """What `upload_source_file` returns; the route maps this to `FileUploadResponse`."""
+
+    file_id: uuid.UUID
+    ingestion_job_id: uuid.UUID
+    status: str
+    chunks_indexed: int
+    is_new_attempt: bool
+    """`False` when an identical active upload already succeeded and no new
+    ingestion attempt was made (route returns `200`, not `201`)."""

src/application/files/status.py

@@ -0,0 +1,52 @@
+"""`GET /v1/files/{file_id}` read model (ADR-0008, ADR-0009)."""
+
+import uuid
+from dataclasses import dataclass
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+
+
+@dataclass(frozen=True)
+class FileStatusResult:
+    file_id: uuid.UUID
+    source_filename: str
+    domain: str
+    status: str
+    ingestion_job_id: uuid.UUID | None
+    ingestion_status: str | None
+    chunks_indexed: int
+
+
+async def get_file_status(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    source_file_id: uuid.UUID,
+) -> FileStatusResult | None:
+    """Returns `None` when the file doesn't exist under this tenant — the
+    route maps that to `404`, never `403` (ADR-0016: cross-tenant access
+    returns 404).
+    """
+    async with sessionmaker() as session:
+        source_file = await source_files_repo.get_by_id(
+            session, tenant_id=tenant_id, source_file_id=source_file_id
+        )
+        if source_file is None:
+            return None
+
+        latest_job = await jobs_repo.get_latest_for_source_file(
+            session, tenant_id=tenant_id, source_file_id=source_file.id
+        )
+
+        return FileStatusResult(
+            file_id=source_file.id,
+            source_filename=source_file.source_filename,
+            domain=source_file.domain,
+            status=source_file.status,
+            ingestion_job_id=latest_job.id if latest_job else None,
+            ingestion_status=latest_job.status if latest_job else None,
+            chunks_indexed=latest_job.points_created if latest_job else 0,
+        )

src/application/files/storage_keys.py

@@ -0,0 +1,11 @@
+"""Object-storage key derivation (ADR-0013).
+
+Object keys are internal identifiers, never the caller-supplied filename.
+Pure and synchronous.
+"""
+
+import uuid
+
+
+def source_file_object_key(tenant_id: uuid.UUID, source_file_id: uuid.UUID) -> str:
+    return f"tenants/{tenant_id}/source-files/{source_file_id}/original"

src/application/files/upload.py

@@ -0,0 +1,205 @@
+"""`POST /v1/files` orchestration: the ADR-0017 three-phase upload.
+
+This service owns two separate short-lived sessions/transactions rather than
+one request-scoped session, because the request is two units of work
+(ADR-0012, ADR-0017):
+
+    txn A (short):  source_files [+ ingestion_jobs(status='running')], commit
+    no txn:          store bytes in MinIO
+    txn B (short):  ingestion_jobs -> succeeded/failed, append event, commit
+
+No Postgres session is open during the MinIO write. A storage failure between
+txn A and txn B still leaves a durable, inspectable `failed` job — never a
+job stuck in `running`.
+
+Parsing/chunking/Qdrant indexing are Phase 4/5 work, not implemented here:
+this phase stores bytes only, so a successful job reports `chunks_indexed=0`.
+"""
+
+import uuid
+
+import structlog
+from anyio import CapacityLimiter, to_thread
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.context import AuthContext
+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
+from src.application.files.validation import validate_and_hash_upload
+from src.application.ports.object_storage import ObjectStorage
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+
+logger = structlog.get_logger(__name__)
+
+
+async def _mark_job_failed(
+    sessionmaker: async_sessionmaker[AsyncSession],
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    error_code: str,
+    error_message: str,
+) -> None:
+    async with sessionmaker() as session:
+        job = await jobs_repo.mark_terminal(
+            session,
+            tenant_id=tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            status="failed",
+            error_code=error_code,
+            error_message=error_message,
+        )
+        if job is not None:
+            jobs_repo.append_event(
+                session,
+                tenant_id=tenant_id,
+                ingestion_job_id=ingestion_job_id,
+                level="error",
+                stage="received",
+                message=error_message,
+            )
+        await session.commit()
+
+
+async def upload_source_file(
+    *,
+    sessionmaker: async_sessionmaker[AsyncSession],
+    storage: ObjectStorage,
+    auth: AuthContext,
+    domain: str,
+    filename: str,
+    data: bytes,
+    max_upload_size_bytes: int,
+    chunking_strategy: str,
+    validation_limiter: CapacityLimiter,
+) -> UploadResult:
+    domain = domain.strip()
+    if not domain:
+        raise InvalidUploadError("domain is required")
+
+    validated = await to_thread.run_sync(
+        lambda: validate_and_hash_upload(
+            filename=filename, data=data, max_size_bytes=max_upload_size_bytes
+        ),
+        limiter=validation_limiter,
+    )
+
+    async with sessionmaker() as session:
+        existing = await source_files_repo.find_active_by_content_hash(
+            session,
+            tenant_id=auth.tenant_id,
+            domain=domain,
+            content_sha256=validated.content_sha256,
+        )
+
+        if existing is not None:
+            latest_job = await jobs_repo.get_latest_for_source_file(
+                session, tenant_id=auth.tenant_id, source_file_id=existing.id
+            )
+            if latest_job is not None and latest_job.status == "succeeded":
+                logger.info(
+                    "files.upload.duplicate",
+                    tenant_id=str(auth.tenant_id),
+                    file_id=str(existing.id),
+                )
+                return UploadResult(
+                    file_id=existing.id,
+                    ingestion_job_id=latest_job.id,
+                    status=latest_job.status,
+                    chunks_indexed=latest_job.points_created,
+                    is_new_attempt=False,
+                )
+            source_file_id = existing.id
+            object_key = existing.storage_uri or source_file_object_key(
+                auth.tenant_id, source_file_id
+            )
+        else:
+            source_file_id = uuid.uuid4()
+            object_key = source_file_object_key(auth.tenant_id, source_file_id)
+            source_files_repo.create(
+                session,
+                source_file_id=source_file_id,
+                tenant_id=auth.tenant_id,
+                domain=domain,
+                source_filename=filename,
+                source_type=validated.source_type,
+                content_sha256=validated.content_sha256,
+                byte_size=len(data),
+                storage_uri=object_key,
+                created_by_api_key_id=auth.api_key_id,
+            )
+            # `ingestion_jobs.source_file_id` FKs to this row; flush so the
+            # insert below sees it, since the two mapped classes carry no
+            # ORM relationship for the unit of work to order by itself.
+            await session.flush()
+
+        job = jobs_repo.create_running(
+            session,
+            tenant_id=auth.tenant_id,
+            source_file_id=source_file_id,
+            requested_by_api_key_id=auth.api_key_id,
+            chunking_strategy=chunking_strategy,
+        )
+        jobs_repo.append_event(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=job.id,
+            level="info",
+            stage="received",
+            message="upload accepted, storing object",
+        )
+        await session.commit()
+        ingestion_job_id = job.id
+
+    # Phase 2: no Postgres session open across this work (ADR-0017).
+    try:
+        await storage.put_object(key=object_key, data=data, content_type=validated.content_type)
+    except Exception as exc:
+        logger.warning(
+            "files.upload.storage_failed",
+            tenant_id=str(auth.tenant_id),
+            file_id=str(source_file_id),
+            ingestion_job_id=str(ingestion_job_id),
+        )
+        await _mark_job_failed(
+            sessionmaker,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            error_code="storage_upload_failed",
+            error_message=f"failed to store object: {exc}",
+        )
+        raise
+
+    async with sessionmaker() as session:
+        await jobs_repo.mark_terminal(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            status="succeeded",
+            points_created=0,
+        )
+        jobs_repo.append_event(
+            session,
+            tenant_id=auth.tenant_id,
+            ingestion_job_id=ingestion_job_id,
+            level="info",
+            stage="completed",
+            message="object stored; parsing/embedding/indexing not yet implemented",
+        )
+        await session.commit()
+
+    logger.info(
+        "files.upload.succeeded",
+        tenant_id=str(auth.tenant_id),
+        file_id=str(source_file_id),
+        ingestion_job_id=str(ingestion_job_id),
+    )
+    return UploadResult(
+        file_id=source_file_id,
+        ingestion_job_id=ingestion_job_id,
+        status="succeeded",
+        chunks_indexed=0,
+        is_new_attempt=True,
+    )

src/application/files/validation.py

@@ -0,0 +1,55 @@
+"""Upload extension/content-type/size validation (ADR-0008).
+
+Pure and synchronous: no I/O. `content_sha256` computation lives here too —
+hashing is blocking CPU work (ADR-0017), so the caller runs this whole
+function through `anyio.to_thread.run_sync` with the ingestion
+`CapacityLimiter`, the same rule applied to parsing/chunking.
+"""
+
+import hashlib
+
+from src.application.files.errors import FileTooLargeError, InvalidUploadError
+from src.application.files.models import ValidatedUpload
+from src.application.ingestion.errors import UnsupportedSourceTypeError
+
+_CONTENT_TYPES = {
+    "csv": "text/csv",
+    "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+}
+_OOXML_MAGIC = b"PK\x03\x04"
+
+
+def _source_type_from_filename(filename: str) -> str:
+    suffix = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
+    if suffix == "doc":
+        raise UnsupportedSourceTypeError(
+            "legacy .doc is not ingestible until an out-of-process conversion "
+            "service exists (ADR-0018)"
+        )
+    if suffix not in _CONTENT_TYPES:
+        raise UnsupportedSourceTypeError(f"'.{suffix}' is not an ingestible file type")
+    return suffix
+
+
+def validate_and_hash_upload(*, filename: str, data: bytes, max_size_bytes: int) -> ValidatedUpload:
+    source_type = _source_type_from_filename(filename)
+
+    if not data:
+        raise InvalidUploadError("uploaded file is empty")
+    if len(data) > max_size_bytes:
+        raise FileTooLargeError(
+            f"upload is {len(data)} bytes, over the {max_size_bytes}-byte limit"
+        )
+
+    is_ooxml = data[:4] == _OOXML_MAGIC
+    if source_type in ("docx", "xlsx") and not is_ooxml:
+        raise InvalidUploadError(f"content does not match the declared .{source_type} extension")
+    if source_type == "csv" and is_ooxml:
+        raise InvalidUploadError("content does not match the declared .csv extension")
+
+    return ValidatedUpload(
+        source_type=source_type,
+        content_type=_CONTENT_TYPES[source_type],
+        content_sha256=hashlib.sha256(data).hexdigest(),
+    )

src/application/ingestion/__init__.py

@@ -1,9 +1,11 @@
 """Document parsing and fixed-size chunking (ADR-0004, ADR-0018).
 
-Everything here is pure and synchronous: no I/O, no ports, no SDK clients
-(ADR-0015 reserves ports for external side effects). Parsing and chunking are
-blocking CPU work, so callers run them through `anyio.to_thread.run_sync` with
-the ingestion `CapacityLimiter` rather than on the event loop (ADR-0017).
+`parse_and_chunk_document` is the entry point callers outside this package
+should use: it dispatches on source type and owns the
+`anyio.to_thread.run_sync` + `CapacityLimiter` offload required by ADR-0017.
+The individual parsers and `chunk_document` are pure, synchronous, and
+exported mainly for their own unit tests — calling them directly from an
+`async def` route or service is the defect ADR-0017 warns about.
 """
 
 from src.application.ingestion.chunking import chunk_document, chunk_id_for, split_by_tokens
@@ -22,6 +24,7 @@ from src.application.ingestion.models import (
     StructuralUnit,
 )
 from src.application.ingestion.normalization import normalize_persian_text
+from src.application.ingestion.pipeline import parse_and_chunk_document
 from src.application.ingestion.spreadsheet_parser import parse_csv, parse_xlsx
 from src.application.ingestion.tokenizer import count_tokens, get_encoder
 
@@ -40,6 +43,7 @@ __all__ = [
     "count_tokens",
     "get_encoder",
     "normalize_persian_text",
+    "parse_and_chunk_document",
     "parse_csv",
     "parse_docx",
     "parse_xlsx",

src/application/ingestion/pipeline.py

@@ -0,0 +1,65 @@
+"""The one caller-facing entry point for parsing and chunking (ADR-0017).
+
+`parse_docx`/`parse_csv`/`parse_xlsx`/`chunk_document` are blocking, pure
+functions; calling any of them directly from an `async def` route or service
+is the defect ADR-0017 names explicitly ("one large `python-docx` parse would
+stall every concurrent request"). `parse_and_chunk_document` is the only
+version of this pipeline callers should reach for: it owns source-type
+dispatch and the `anyio.to_thread.run_sync` + `CapacityLimiter` offload, so
+that obligation cannot be forgotten at a call site.
+"""
+
+import uuid
+from functools import partial
+
+from anyio import CapacityLimiter, to_thread
+
+from src.application.ingestion.chunking import chunk_document
+from src.application.ingestion.docx_parser import parse_docx
+from src.application.ingestion.errors import UnsupportedSourceTypeError
+from src.application.ingestion.models import Chunk, ParsedDocument
+from src.application.ingestion.spreadsheet_parser import parse_csv, parse_xlsx
+from src.config import ChunkingSettings
+
+_PARSERS = {"csv", "xlsx", "docx"}
+
+
+def _parse(data: bytes, source_type: str, settings: ChunkingSettings) -> ParsedDocument:
+    if source_type == "docx":
+        return parse_docx(data, settings)
+    if source_type == "xlsx":
+        return parse_xlsx(data)
+    if source_type == "csv":
+        return parse_csv(data)
+    raise UnsupportedSourceTypeError(f"'{source_type}' is not an ingestible source type")
+
+
+def _parse_and_chunk(
+    data: bytes, source_type: str, file_id: uuid.UUID, settings: ChunkingSettings
+) -> list[Chunk]:
+    parsed = _parse(data, source_type, settings)
+    return chunk_document(parsed, file_id=file_id, settings=settings)
+
+
+async def parse_and_chunk_document(
+    data: bytes,
+    *,
+    source_type: str,
+    file_id: uuid.UUID,
+    settings: ChunkingSettings,
+    limiter: CapacityLimiter,
+) -> list[Chunk]:
+    """Parse and chunk a document off the event loop, bounded by `limiter`.
+
+    Raises `UnsupportedSourceTypeError` (415), `DocumentParseError` (400), or
+    `ChunkTooLargeError` — see `src/application/ingestion/errors.py`. Callers
+    map these to status codes; this module carries no HTTP knowledge
+    (ADR-0015). The `max_chunks_per_file` ceiling (413) is enforced by the
+    caller, not here — see Phase 4 of plan 001.
+    """
+    if source_type not in _PARSERS:
+        raise UnsupportedSourceTypeError(f"'{source_type}' is not an ingestible source type")
+    return await to_thread.run_sync(
+        partial(_parse_and_chunk, data, source_type, file_id, settings),
+        limiter=limiter,
+    )

src/application/ports/__init__.py

@@ -0,0 +1,7 @@
+"""Narrow contracts for external side effects (ADR-0015).
+
+Ports exist for external side effects/persistence that need a swappable or
+fake-able boundary — not as a blanket wrapper around every database access.
+`object_storage.py` is one: MinIO is a real external system with its own
+failure modes, and ADR-0016 requires a hand-written fake for it in tests.
+"""

src/application/ports/object_storage.py

@@ -0,0 +1,14 @@
+"""The object-storage port (ADR-0013).
+
+`src/infrastructure/minio/storage.py` is the production adapter; tests use a
+hand-written fake (ADR-0016). Application code depends on this Protocol, not
+on the `minio` SDK.
+"""
+
+from typing import Protocol
+
+
+class ObjectStorage(Protocol):
+    async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
+        """Store `data` privately under `key`. Overwrites an existing object."""
+        ...

src/bootstrap/dependencies.py

@@ -1,11 +1,13 @@
 from collections.abc import AsyncIterator
 from dataclasses import dataclass
 
+from anyio import CapacityLimiter
 from fastapi import Request
 from minio import Minio
 from qdrant_client import AsyncQdrantClient
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
 
+from src.application.ports.object_storage import ObjectStorage
 from src.config import Settings
 
 
@@ -16,6 +18,8 @@ class AppResources:
     db_sessionmaker: async_sessionmaker[AsyncSession]
     minio_client: Minio
     qdrant_client: AsyncQdrantClient
+    object_storage: ObjectStorage
+    ingestion_limiter: CapacityLimiter
 
 
 def _resources(request: Request) -> AppResources:
@@ -34,6 +38,25 @@ def get_qdrant_client(request: Request) -> AsyncQdrantClient:
     return _resources(request).qdrant_client
 
 
+def get_object_storage(request: Request) -> ObjectStorage:
+    return _resources(request).object_storage
+
+
+def get_ingestion_limiter(request: Request) -> CapacityLimiter:
+    return _resources(request).ingestion_limiter
+
+
+def get_sessionmaker(request: Request) -> async_sessionmaker[AsyncSession]:
+    """The session *factory*, not a request-scoped session.
+
+    Application services that own more than one transaction in a single
+    request (ADR-0017's two-phase upload) need to open and close sessions
+    themselves rather than borrow one request-scoped session that would
+    otherwise stay open across the whole request.
+    """
+    return _resources(request).db_sessionmaker
+
+
 async def get_db_session(request: Request) -> AsyncIterator[AsyncSession]:
     sessionmaker = _resources(request).db_sessionmaker
     async with sessionmaker() as session:

src/bootstrap/lifespan.py

@@ -2,13 +2,14 @@ from collections.abc import AsyncIterator, Callable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 
 import structlog
-from anyio import to_thread
+from anyio import CapacityLimiter, to_thread
 from fastapi import FastAPI
 
 from src.application.ingestion import get_encoder
 from src.bootstrap.dependencies import AppResources
 from src.config import Settings
 from src.infrastructure.minio.client import create_client as create_minio_client
+from src.infrastructure.minio.storage import MinioObjectStorage
 from src.infrastructure.observability.logging import configure_logging
 from src.infrastructure.postgres.database import create_engine, create_sessionmaker
 from src.infrastructure.qdrant.client import create_client as create_qdrant_client
@@ -43,12 +44,22 @@ def create_lifespan(
         qdrant_client = create_qdrant_client(resolved_settings.qdrant)
         logger.info("lifespan.qdrant.client.created")
 
+        # Bounds threads spent on blocking ingestion work (parsing, chunking,
+        # hashing, the sync minio SDK) so it cannot exhaust Starlette's own
+        # thread pool (ADR-0017).
+        ingestion_limiter = CapacityLimiter(resolved_settings.ingestion.thread_pool_size)
+        object_storage = MinioObjectStorage(
+            minio_client, bucket=resolved_settings.minio.bucket, limiter=ingestion_limiter
+        )
+
         app.state.resources = AppResources(
             settings=resolved_settings,
             db_engine=db_engine,
             db_sessionmaker=db_sessionmaker,
             minio_client=minio_client,
             qdrant_client=qdrant_client,
+            object_storage=object_storage,
+            ingestion_limiter=ingestion_limiter,
         )
 
         try:

src/config.py

@@ -38,10 +38,15 @@ class IngestionSettings(BaseSettings):
     max_concurrency: int = 4
     thread_pool_size: int = 8
     timeout_seconds: float = 120.0
+    max_upload_size_mb: int = 25
     max_chunks_per_file: int = 5000
     embed_batch_size: int = 128
     embed_concurrency: int = 4
 
+    @property
+    def max_upload_size_bytes(self) -> int:
+        return self.max_upload_size_mb * 1024 * 1024
+
 
 class ChunkingSettings(BaseSettings):
     """Parsing and chunking parameters (ADR-0018).
@@ -86,7 +91,6 @@ class AppLimitSettings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="APP_", extra="ignore")
 
     env: str = "local"
-    max_upload_size_mb: int = 25
     readiness_check_timeout_seconds: float = 2.0
 
 

src/infrastructure/minio/storage.py

@@ -0,0 +1,33 @@
+"""MinIO adapter for the `ObjectStorage` port (ADR-0013, ADR-0017).
+
+The `minio` SDK is synchronous, so every call runs through
+`anyio.to_thread.run_sync` bounded by the ingestion `CapacityLimiter` — the
+same rule ADR-0017 applies to parsing/chunking. Calling the SDK directly from
+`async def` would block every concurrent request in the process.
+"""
+
+import io
+from functools import partial
+
+from anyio import CapacityLimiter, to_thread
+from minio import Minio
+
+
+class MinioObjectStorage:
+    def __init__(self, client: Minio, *, bucket: str, limiter: CapacityLimiter) -> None:
+        self._client = client
+        self._bucket = bucket
+        self._limiter = limiter
+
+    async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
+        await to_thread.run_sync(
+            partial(
+                self._client.put_object,
+                self._bucket,
+                key,
+                io.BytesIO(data),
+                length=len(data),
+                content_type=content_type,
+            ),
+            limiter=self._limiter,
+        )

src/infrastructure/postgres/repositories/api_keys.py

@@ -0,0 +1,17 @@
+"""API-key lookups (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
+happens in `src/application/auth`, not here — this module only fetches rows
+by their non-secret `key_prefix`.
+"""
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+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()

src/infrastructure/postgres/repositories/ingestion_jobs.py

@@ -0,0 +1,119 @@
+"""`ingestion_jobs`/`ingestion_job_events` persistence (ADR-0009, ADR-0017).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012) — the two-transaction
+shape in `src/application/files/upload.py` depends on that. Every read is
+tenant-scoped by a required `tenant_id` argument.
+"""
+
+import uuid
+from datetime import UTC, datetime
+
+from sqlalchemy import desc, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.ingestion_job import IngestionJob
+from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent
+
+
+async def get_by_id(
+    session: AsyncSession, *, tenant_id: uuid.UUID, ingestion_job_id: uuid.UUID
+) -> IngestionJob | None:
+    result = await session.execute(
+        select(IngestionJob).where(
+            IngestionJob.id == ingestion_job_id, IngestionJob.tenant_id == tenant_id
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+async def get_latest_for_source_file(
+    session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID
+) -> IngestionJob | None:
+    result = await session.execute(
+        select(IngestionJob)
+        .where(
+            IngestionJob.tenant_id == tenant_id,
+            IngestionJob.source_file_id == source_file_id,
+        )
+        .order_by(desc(IngestionJob.created_at))
+        .limit(1)
+    )
+    return result.scalar_one_or_none()
+
+
+def create_running(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    source_file_id: uuid.UUID,
+    requested_by_api_key_id: uuid.UUID | None,
+    chunking_strategy: str,
+) -> IngestionJob:
+    job = IngestionJob(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        source_file_id=source_file_id,
+        requested_by_api_key_id=requested_by_api_key_id,
+        status="running",
+        chunking_strategy=chunking_strategy,
+        started_at=datetime.now(UTC),
+    )
+    session.add(job)
+    return job
+
+
+async def mark_terminal(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    status: str,
+    points_created: int = 0,
+    points_updated: int = 0,
+    points_soft_deleted: int = 0,
+    points_skipped: int = 0,
+    error_code: str | None = None,
+    error_message: str | None = None,
+) -> IngestionJob | None:
+    """Move a job from `running` to a terminal status.
+
+    Never reads/writes a job whose current status is already terminal — a
+    terminal job must not transition back to `running` or to a different
+    terminal status (ADR-0017).
+    """
+    job = await get_by_id(session, tenant_id=tenant_id, ingestion_job_id=ingestion_job_id)
+    if job is None or job.status != "running":
+        return None
+    job.status = status
+    job.completed_at = datetime.now(UTC)
+    job.points_created = points_created
+    job.points_updated = points_updated
+    job.points_soft_deleted = points_soft_deleted
+    job.points_skipped = points_skipped
+    job.error_code = error_code
+    job.error_message = error_message
+    return job
+
+
+def append_event(
+    session: AsyncSession,
+    *,
+    tenant_id: uuid.UUID,
+    ingestion_job_id: uuid.UUID,
+    level: str,
+    stage: str,
+    message: str,
+    details: dict[str, object] | None = None,
+) -> IngestionJobEvent:
+    event = IngestionJobEvent(
+        id=uuid.uuid4(),
+        tenant_id=tenant_id,
+        ingestion_job_id=ingestion_job_id,
+        level=level,
+        stage=stage,
+        message=message,
+        details=details or {},
+    )
+    session.add(event)
+    return event

src/infrastructure/postgres/repositories/source_files.py

@@ -0,0 +1,69 @@
+"""`source_files` 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 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 sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.models.source_file import SourceFile
+
+
+async def get_by_id(
+    session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID
+) -> SourceFile | None:
+    result = await session.execute(
+        select(SourceFile).where(SourceFile.id == source_file_id, SourceFile.tenant_id == tenant_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def find_active_by_content_hash(
+    session: AsyncSession, *, tenant_id: uuid.UUID, domain: str, content_sha256: str
+) -> SourceFile | None:
+    result = await session.execute(
+        select(SourceFile).where(
+            SourceFile.tenant_id == tenant_id,
+            SourceFile.domain == domain,
+            SourceFile.content_sha256 == content_sha256,
+            SourceFile.status == "active",
+        )
+    )
+    return result.scalar_one_or_none()
+
+
+def create(
+    session: AsyncSession,
+    *,
+    source_file_id: uuid.UUID,
+    tenant_id: uuid.UUID,
+    domain: str,
+    source_filename: str,
+    source_type: str,
+    content_sha256: str,
+    byte_size: int,
+    storage_uri: str,
+    created_by_api_key_id: uuid.UUID | None,
+) -> SourceFile:
+    """`source_file_id` is caller-generated: the upload service derives the
+    MinIO object key from it before this row exists, so the id has to be
+    chosen up front rather than assigned by the database.
+    """
+    source_file = SourceFile(
+        id=source_file_id,
+        tenant_id=tenant_id,
+        domain=domain,
+        source_filename=source_filename,
+        source_type=source_type,
+        content_sha256=content_sha256,
+        byte_size=byte_size,
+        storage_uri=storage_uri,
+        created_by_api_key_id=created_by_api_key_id,
+    )
+    session.add(source_file)
+    return source_file

src/infrastructure/postgres/repositories/tenants.py

@@ -0,0 +1,15 @@
+"""Tenant lookups (ADR-0009).
+
+Plain functions over an `AsyncSession` the caller owns. No function here
+commits, rolls back, or closes the session (ADR-0012).
+"""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+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)

src/main.py

@@ -1,5 +1,7 @@
 from fastapi import FastAPI
 
+from src.api.errors import register_exception_handlers
+from src.api.middleware import RequestIdMiddleware
 from src.api.router import router as v1_router
 from src.api.routers.health import router as health_router
 from src.bootstrap.lifespan import create_lifespan
@@ -8,6 +10,8 @@ from src.config import Settings
 
 def create_app(settings: Settings | None = None) -> FastAPI:
     app = FastAPI(lifespan=create_lifespan(settings))
+    app.add_middleware(RequestIdMiddleware)
+    register_exception_handlers(app)
     app.include_router(health_router)
     app.include_router(v1_router, prefix="/v1")
     return app

tests/fakes.py

@@ -1,5 +1,17 @@
-"""Hand-written fakes for narrow application-owned ports.
+"""Hand-written fakes for narrow application-owned ports (ADR-0016)."""
 
-No application ports exist yet (Phase 1 only wires infrastructure client
-lifecycle). Fakes are added here as ports are introduced in later phases.
-"""
+from dataclasses import dataclass, field
+
+
+@dataclass
+class FakeObjectStorage:
+    """In-memory `ObjectStorage`. `fail_next` simulates one upload failure."""
+
+    objects: dict[str, bytes] = field(default_factory=dict)
+    fail_next: bool = False
+
+    async def put_object(self, *, key: str, data: bytes, content_type: str) -> None:
+        if self.fail_next:
+            self.fail_next = False
+            raise OSError("simulated object storage failure")
+        self.objects[key] = data

tests/integration/minio/conftest.py

@@ -0,0 +1,39 @@
+from collections.abc import Iterator
+
+import pytest
+from testcontainers.community.minio import MinioContainer
+
+from src.config import MinioSettings
+from src.infrastructure.minio.client import create_client
+
+_BUCKET = "test-source-files"
+
+
+@pytest.fixture(scope="session")
+def minio_container() -> Iterator[MinioContainer]:
+    with MinioContainer() as container:
+        yield container
+
+
+@pytest.fixture(scope="session")
+def minio_settings(minio_container: MinioContainer) -> MinioSettings:
+    config = minio_container.get_config()
+    # Pinned to IPv4 for the same reason as postgres_url (see
+    # tests/integration/postgres/conftest.py): `localhost` resolves to `::1`
+    # first, but Docker only publishes the mapped port on IPv4, so the
+    # connection hangs instead of failing.
+    endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:")
+    return MinioSettings(
+        endpoint=endpoint,
+        access_key=config["access_key"],
+        secret_key=config["secret_key"],
+        secure=False,
+        bucket=_BUCKET,
+    )
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _ensure_bucket(minio_settings: MinioSettings) -> None:
+    client = create_client(minio_settings)
+    if not client.bucket_exists(minio_settings.bucket):
+        client.make_bucket(minio_settings.bucket)

tests/integration/minio/test_storage.py

@@ -0,0 +1,39 @@
+import uuid
+
+import pytest
+from anyio import CapacityLimiter
+
+from src.application.files.storage_keys import source_file_object_key
+from src.config import MinioSettings
+from src.infrastructure.minio.client import create_client
+from src.infrastructure.minio.storage import MinioObjectStorage
+
+pytestmark = [pytest.mark.integration, pytest.mark.minio, pytest.mark.asyncio]
+
+
+async def test_put_object_stores_bytes_under_server_derived_key(
+    minio_settings: MinioSettings,
+) -> None:
+    client = create_client(minio_settings)
+    storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2))
+    key = source_file_object_key(uuid.uuid4(), uuid.uuid4())
+    data = b"tenant-scoped, id-addressed bytes"
+
+    await storage.put_object(key=key, data=data, content_type="text/csv")
+
+    stored = client.get_object(minio_settings.bucket, key).read()
+    assert stored == data
+
+
+async def test_put_object_overwrites_existing_object_at_same_key(
+    minio_settings: MinioSettings,
+) -> None:
+    client = create_client(minio_settings)
+    storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2))
+    key = source_file_object_key(uuid.uuid4(), uuid.uuid4())
+
+    await storage.put_object(key=key, data=b"first", content_type="text/csv")
+    await storage.put_object(key=key, data=b"second", content_type="text/csv")
+
+    stored = client.get_object(minio_settings.bucket, key).read()
+    assert stored == b"second"

tests/integration/postgres/conftest.py

@@ -54,7 +54,7 @@ def migrated_postgres_url(postgres_url: str) -> str:
     return postgres_url
 
 
-@pytest_asyncio.fixture(scope="session")
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
 async def postgres_engine(migrated_postgres_url: str) -> AsyncIterator[AsyncEngine]:
     engine = create_engine(_settings_from_url(migrated_postgres_url))
     try:
@@ -63,18 +63,36 @@ async def postgres_engine(migrated_postgres_url: str) -> AsyncIterator[AsyncEngi
         await engine.dispose()
 
 
-@pytest_asyncio.fixture
-async def db_session(postgres_engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
-    """One session per test, bound to a rolled-back outer transaction.
+@pytest_asyncio.fixture(loop_scope="session")
+async def db_sessionmaker(
+    postgres_engine: AsyncEngine,
+) -> AsyncIterator[async_sessionmaker[AsyncSession]]:
+    """A session *factory* per test, bound to a rolled-back outer transaction.
 
-    Isolates each test's writes (ADR-0016: isolate data per test) without
-    needing a fresh container or unique keys per test.
+    Every session it produces shares one connection/outer transaction, so
+    writes `commit()`ed by one session are visible to the next -- needed for
+    code under test that opens more than one session per operation (auth
+    resolution, the ADR-0017 two-phase upload) -- while the whole test's
+    writes still roll back together at teardown (ADR-0016: isolate data per
+    test).
     """
     async with postgres_engine.connect() as connection:
         outer_transaction = await connection.begin()
         sessionmaker = async_sessionmaker(
             bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
         )
-        async with sessionmaker() as session:
-            yield session
+        yield sessionmaker
         await outer_transaction.rollback()
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def db_session(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> AsyncIterator[AsyncSession]:
+    """One session per test, bound to a rolled-back outer transaction.
+
+    Isolates each test's writes (ADR-0016: isolate data per test) without
+    needing a fresh container or unique keys per test.
+    """
+    async with db_sessionmaker() as session:
+        yield session

tests/integration/postgres/test_auth_service.py

@@ -0,0 +1,65 @@
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError
+from src.application.auth.service import resolve_auth_context
+from tests.support.factories import create_api_key, create_tenant
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def test_resolve_auth_context_accepts_valid_key(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, full_key = await create_api_key(db_session, tenant=tenant, scopes=["files:write"])
+    await db_session.commit()
+
+    auth = await resolve_auth_context(db_sessionmaker, full_key)
+
+    assert auth.tenant_id == tenant.id
+    assert auth.has_scope("files:write")
+
+
+async def test_resolve_auth_context_rejects_wrong_secret(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    api_key, _ = await create_api_key(db_session, tenant=tenant)
+    await db_session.commit()
+
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_wrong-secret")
+
+
+async def test_resolve_auth_context_rejects_unknown_prefix(
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+) -> None:
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, "sk_doesnotexist_secret")
+
+
+async def test_resolve_auth_context_rejects_revoked_key(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session)
+    _, full_key = await create_api_key(db_session, tenant=tenant, status="revoked")
+    await db_session.commit()
+
+    with pytest.raises(InvalidApiKeyError):
+        await resolve_auth_context(db_sessionmaker, full_key)
+
+
+async def test_resolve_auth_context_rejects_suspended_tenant(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    tenant = await create_tenant(db_session, status="suspended")
+    _, full_key = await create_api_key(db_session, tenant=tenant)
+    await db_session.commit()
+
+    with pytest.raises(TenantInactiveError):
+        await resolve_auth_context(db_sessionmaker, full_key)

tests/integration/postgres/test_migrations.py

@@ -2,7 +2,11 @@ import pytest
 from sqlalchemy import inspect
 from sqlalchemy.ext.asyncio import AsyncEngine
 
-pytestmark = [pytest.mark.integration, pytest.mark.postgres, pytest.mark.asyncio]
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
 
 EXPECTED_TABLES = {
     "tenants",

tests/integration/postgres/test_repositories.py

@@ -0,0 +1,113 @@
+import uuid
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo
+from src.infrastructure.postgres.repositories import source_files as source_files_repo
+from tests.support.factories import create_tenant
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def test_source_files_get_by_id_is_tenant_scoped(db_session: AsyncSession) -> None:
+    owner = await create_tenant(db_session)
+    other = await create_tenant(db_session)
+    source_file_id = uuid.uuid4()
+    source_files_repo.create(
+        db_session,
+        source_file_id=source_file_id,
+        tenant_id=owner.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="a" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.commit()
+
+    found_for_owner = await source_files_repo.get_by_id(
+        db_session, tenant_id=owner.id, source_file_id=source_file_id
+    )
+    found_for_other = await source_files_repo.get_by_id(
+        db_session, tenant_id=other.id, source_file_id=source_file_id
+    )
+
+    assert found_for_owner is not None
+    assert found_for_other is None
+
+
+async def test_source_files_find_active_by_content_hash_matches_tenant_domain_hash(
+    db_session: AsyncSession,
+) -> None:
+    tenant = await create_tenant(db_session)
+    source_files_repo.create(
+        db_session,
+        source_file_id=uuid.uuid4(),
+        tenant_id=tenant.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="b" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.commit()
+
+    found = await source_files_repo.find_active_by_content_hash(
+        db_session, tenant_id=tenant.id, domain="general", content_sha256="b" * 64
+    )
+    not_found_other_domain = await source_files_repo.find_active_by_content_hash(
+        db_session, tenant_id=tenant.id, domain="other", content_sha256="b" * 64
+    )
+
+    assert found is not None
+    assert not_found_other_domain is None
+
+
+async def test_ingestion_jobs_mark_terminal_rejects_non_running_job(
+    db_session: AsyncSession,
+) -> None:
+    """A job already in a terminal state must not be re-marked (ADR-0017)."""
+    tenant = await create_tenant(db_session)
+    source_file_id = uuid.uuid4()
+    source_files_repo.create(
+        db_session,
+        source_file_id=source_file_id,
+        tenant_id=tenant.id,
+        domain="general",
+        source_filename="report.csv",
+        source_type="csv",
+        content_sha256="c" * 64,
+        byte_size=10,
+        storage_uri="tenants/x/source-files/y/original",
+        created_by_api_key_id=None,
+    )
+    await db_session.flush()
+    job = jobs_repo.create_running(
+        db_session,
+        tenant_id=tenant.id,
+        source_file_id=source_file_id,
+        requested_by_api_key_id=None,
+        chunking_strategy="fixed_size",
+    )
+    await db_session.commit()
+
+    first = await jobs_repo.mark_terminal(
+        db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="succeeded"
+    )
+    await db_session.commit()
+    second = await jobs_repo.mark_terminal(
+        db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="failed"
+    )
+
+    assert first is not None
+    assert first.status == "succeeded"
+    assert second is None

tests/integration/postgres/test_upload_service.py

@@ -0,0 +1,194 @@
+"""ADR-0017's two-transaction upload shape, exercised against real Postgres.
+
+Object storage is faked (`FakeObjectStorage`) -- it's a port, not the thing
+under test here. `MinioObjectStorage` itself is covered in
+`tests/integration/minio/`.
+"""
+
+import pytest
+from anyio import CapacityLimiter
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.application.auth.context import AuthContext
+from src.application.files.upload import upload_source_file
+from src.infrastructure.postgres.models.ingestion_job import IngestionJob
+from tests.fakes import FakeObjectStorage
+from tests.support.factories import create_api_key, create_tenant
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.postgres,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+_CSV_BYTES = b"name,value\nfirst,1\n"
+
+
+async def _auth_for(db_session: AsyncSession) -> AuthContext:
+    tenant = await create_tenant(db_session)
+    api_key, _ = await create_api_key(db_session, tenant=tenant)
+    await db_session.commit()
+    return AuthContext(
+        tenant_id=tenant.id,
+        tenant_slug=tenant.slug,
+        api_key_id=api_key.id,
+        scopes=frozenset({"files:write"}),
+        actor_type="backend",
+    )
+
+
+async def test_upload_source_file_commits_running_job_before_storage_write(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """After phase 1 commits, a *separate* session must already see the
+    running job -- proving txn A committed and released before phase 2 work,
+    per ADR-0017.
+    """
+    auth = await _auth_for(db_session)
+    storage = FakeObjectStorage()
+
+    result = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=storage,
+        auth=auth,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+
+    assert result.status == "succeeded"
+    assert result.chunks_indexed == 0
+    assert result.is_new_attempt
+
+    async with db_sessionmaker() as verify_session:
+        job = await verify_session.get(IngestionJob, result.ingestion_job_id)
+    assert job is not None
+    assert job.status == "succeeded"
+    assert storage.objects  # bytes were actually written
+
+
+async def test_upload_source_file_duplicate_hash_returns_existing_without_reingesting(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    auth = await _auth_for(db_session)
+    storage = FakeObjectStorage()
+
+    first = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=storage,
+        auth=auth,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+    second = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=storage,
+        auth=auth,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+
+    assert second.file_id == first.file_id
+    assert second.ingestion_job_id == first.ingestion_job_id
+    assert not second.is_new_attempt
+    assert len(storage.objects) == 1  # no second write
+
+
+async def test_upload_source_file_retries_after_failed_job(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """A duplicate upload whose last attempt failed must get a fresh job,
+    not be treated as already-succeeded (ADR-0017: safe to run more than
+    once).
+    """
+    auth = await _auth_for(db_session)
+    failing_storage = FakeObjectStorage(fail_next=True)
+
+    with pytest.raises(OSError):
+        await upload_source_file(
+            sessionmaker=db_sessionmaker,
+            storage=failing_storage,
+            auth=auth,
+            domain="general",
+            filename="report.csv",
+            data=_CSV_BYTES,
+            max_upload_size_bytes=1_000_000,
+            chunking_strategy="fixed_size",
+            validation_limiter=CapacityLimiter(2),
+        )
+
+    async with db_sessionmaker() as verify_session:
+        jobs = (
+            (
+                await verify_session.execute(
+                    select(IngestionJob).where(IngestionJob.tenant_id == auth.tenant_id)
+                )
+            )
+            .scalars()
+            .all()
+        )
+    assert len(jobs) == 1
+    assert jobs[0].status == "failed"
+
+    retry = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+
+    assert retry.status == "succeeded"
+    assert retry.is_new_attempt
+    assert retry.file_id == jobs[0].source_file_id
+
+
+async def test_upload_source_file_is_tenant_isolated_for_identical_content(
+    db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """Two different tenants uploading byte-identical files must not collide
+    on the (tenant_id, domain, content_sha256) idempotency key.
+    """
+    auth_a = await _auth_for(db_session)
+    auth_b = await _auth_for(db_session)
+
+    result_a = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth_a,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+    result_b = await upload_source_file(
+        sessionmaker=db_sessionmaker,
+        storage=FakeObjectStorage(),
+        auth=auth_b,
+        domain="general",
+        filename="report.csv",
+        data=_CSV_BYTES,
+        max_upload_size_bytes=1_000_000,
+        chunking_strategy="fixed_size",
+        validation_limiter=CapacityLimiter(2),
+    )
+
+    assert result_a.file_id != result_b.file_id

tests/support/factories.py

@@ -1 +1,49 @@
-"""Object builders for test fixtures, populated as later phases need them."""
+"""Object builders for test fixtures (ADR-0016).
+
+Insert rows via a caller-supplied `AsyncSession` without committing — callers
+decide their own transaction boundary (the `db_session` fixture rolls back
+after each test).
+"""
+
+import uuid
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.application.auth.keys import generate_api_key, hash_secret
+from src.infrastructure.postgres.models.api_key import ApiKey
+from src.infrastructure.postgres.models.tenant import Tenant
+
+
+async def create_tenant(
+    session: AsyncSession, *, slug: str | None = None, status: str = "active"
+) -> Tenant:
+    slug = slug or f"tenant-{uuid.uuid4().hex[:8]}"
+    tenant = Tenant(id=uuid.uuid4(), slug=slug, name=slug, status=status)
+    session.add(tenant)
+    await session.flush()
+    return tenant
+
+
+async def create_api_key(
+    session: AsyncSession,
+    *,
+    tenant: Tenant,
+    scopes: list[str] | None = None,
+    status: str = "active",
+) -> tuple[ApiKey, str]:
+    """Returns `(api_key, full_key)`. `full_key` is the bearer token to send;
+    only its hash is persisted.
+    """
+    key_prefix, secret, full_key = generate_api_key()
+    api_key = ApiKey(
+        id=uuid.uuid4(),
+        tenant_id=tenant.id,
+        name="test-key",
+        key_prefix=key_prefix,
+        key_hash=hash_secret(secret),
+        scopes=scopes or ["files:write"],
+        status=status,
+    )
+    session.add(api_key)
+    await session.flush()
+    return api_key, full_key

tests/unit/application/auth/test_keys.py

@@ -0,0 +1,38 @@
+import pytest
+
+from src.application.auth.keys import generate_api_key, hash_secret, parse_api_key, verify_secret
+
+pytestmark = pytest.mark.unit
+
+
+def test_generate_api_key_round_trips_through_parse() -> None:
+    key_prefix, secret, full_key = generate_api_key()
+
+    parsed = parse_api_key(full_key)
+
+    assert parsed == (key_prefix, secret)
+
+
+def test_generate_api_key_produces_unique_secrets() -> None:
+    _, secret_a, _ = generate_api_key()
+    _, secret_b, _ = generate_api_key()
+
+    assert secret_a != secret_b
+
+
+@pytest.mark.parametrize(
+    "malformed",
+    ["", "sk_onlyprefix", "wrongscheme_prefix_secret", "sk__secret", "sk_prefix_"],
+)
+def test_parse_api_key_rejects_malformed_input(malformed: str) -> None:
+    assert parse_api_key(malformed) is None
+
+
+def test_verify_secret_accepts_matching_secret() -> None:
+    _, secret, _ = generate_api_key()
+
+    assert verify_secret(secret, hash_secret(secret))
+
+
+def test_verify_secret_rejects_wrong_secret() -> None:
+    assert not verify_secret("wrong-secret", hash_secret("correct-secret"))

tests/unit/application/files/test_storage_keys.py

@@ -0,0 +1,25 @@
+import uuid
+
+import pytest
+
+from src.application.files.storage_keys import source_file_object_key
+
+pytestmark = pytest.mark.unit
+
+
+def test_source_file_object_key_is_server_derived_not_filename() -> None:
+    tenant_id = uuid.uuid4()
+    source_file_id = uuid.uuid4()
+
+    key = source_file_object_key(tenant_id, source_file_id)
+
+    assert key == f"tenants/{tenant_id}/source-files/{source_file_id}/original"
+
+
+def test_source_file_object_key_differs_per_tenant() -> None:
+    source_file_id = uuid.uuid4()
+
+    key_a = source_file_object_key(uuid.uuid4(), source_file_id)
+    key_b = source_file_object_key(uuid.uuid4(), source_file_id)
+
+    assert key_a != key_b

tests/unit/application/files/test_validation.py

@@ -0,0 +1,58 @@
+import pytest
+
+from src.application.files.errors import FileTooLargeError, InvalidUploadError
+from src.application.files.validation import validate_and_hash_upload
+from src.application.ingestion.errors import UnsupportedSourceTypeError
+
+pytestmark = pytest.mark.unit
+
+_DOCX_HEADER = b"PK\x03\x04" + b"\x00" * 20
+_CSV_BYTES = b"name,value\nfirst,1\n"
+
+
+def test_validate_and_hash_upload_accepts_matching_csv() -> None:
+    result = validate_and_hash_upload(filename="report.csv", data=_CSV_BYTES, max_size_bytes=1_000)
+
+    assert result.source_type == "csv"
+    assert result.content_type == "text/csv"
+    assert len(result.content_sha256) == 64
+
+
+def test_validate_and_hash_upload_accepts_matching_docx() -> None:
+    result = validate_and_hash_upload(
+        filename="report.docx", data=_DOCX_HEADER, max_size_bytes=1_000
+    )
+
+    assert result.source_type == "docx"
+
+
+def test_validate_and_hash_upload_rejects_doc_as_unsupported() -> None:
+    with pytest.raises(UnsupportedSourceTypeError):
+        validate_and_hash_upload(filename="legacy.doc", data=_CSV_BYTES, max_size_bytes=1_000)
+
+
+def test_validate_and_hash_upload_rejects_unknown_extension() -> None:
+    with pytest.raises(UnsupportedSourceTypeError):
+        validate_and_hash_upload(filename="report.pdf", data=_CSV_BYTES, max_size_bytes=1_000)
+
+
+def test_validate_and_hash_upload_rejects_empty_file() -> None:
+    with pytest.raises(InvalidUploadError):
+        validate_and_hash_upload(filename="report.csv", data=b"", max_size_bytes=1_000)
+
+
+def test_validate_and_hash_upload_rejects_oversized_file() -> None:
+    with pytest.raises(FileTooLargeError):
+        validate_and_hash_upload(filename="report.csv", data=_CSV_BYTES, max_size_bytes=4)
+
+
+def test_validate_and_hash_upload_rejects_spoofed_docx_extension() -> None:
+    """Content is really CSV text, but the filename claims `.docx`."""
+    with pytest.raises(InvalidUploadError):
+        validate_and_hash_upload(filename="report.docx", data=_CSV_BYTES, max_size_bytes=1_000)
+
+
+def test_validate_and_hash_upload_rejects_spoofed_csv_extension() -> None:
+    """Content is really an OOXML zip, but the filename claims `.csv`."""
+    with pytest.raises(InvalidUploadError):
+        validate_and_hash_upload(filename="report.csv", data=_DOCX_HEADER, max_size_bytes=1_000)

tests/unit/application/test_pipeline.py

@@ -0,0 +1,62 @@
+"""Source-type dispatch and thread-offload for `parse_and_chunk_document`."""
+
+import uuid
+
+import pytest
+from anyio import CapacityLimiter
+
+from src.application.ingestion import UnsupportedSourceTypeError, parse_and_chunk_document
+from src.config import ChunkingSettings
+from tests.support.documents import PROSE_DOCX, load_document
+
+pytestmark = [pytest.mark.unit, pytest.mark.asyncio]
+
+
+@pytest.fixture
+def settings() -> ChunkingSettings:
+    return ChunkingSettings()
+
+
+@pytest.fixture
+def limiter() -> CapacityLimiter:
+    return CapacityLimiter(2)
+
+
+async def test_parse_and_chunk_document_docx_returns_chunks(
+    settings: ChunkingSettings, limiter: CapacityLimiter
+) -> None:
+    chunks = await parse_and_chunk_document(
+        load_document(PROSE_DOCX),
+        source_type="docx",
+        file_id=uuid.uuid4(),
+        settings=settings,
+        limiter=limiter,
+    )
+
+    assert chunks
+    assert all(chunk.chunk_index == index for index, chunk in enumerate(chunks))
+
+
+async def test_parse_and_chunk_document_csv_returns_chunks(
+    settings: ChunkingSettings, limiter: CapacityLimiter
+) -> None:
+    data = b"name,value\nfirst,1\nsecond,2\n"
+
+    chunks = await parse_and_chunk_document(
+        data, source_type="csv", file_id=uuid.uuid4(), settings=settings, limiter=limiter
+    )
+
+    assert chunks
+
+
+async def test_parse_and_chunk_document_unsupported_type_raises(
+    settings: ChunkingSettings, limiter: CapacityLimiter
+) -> None:
+    with pytest.raises(UnsupportedSourceTypeError):
+        await parse_and_chunk_document(
+            b"whatever",
+            source_type="doc",
+            file_id=uuid.uuid4(),
+            settings=settings,
+            limiter=limiter,
+        )