docs(ops): add the operator runbook and record Phase 6 as complete

Why:
- the slice had no operator documentation: no tuning guidance, no failure
  procedure, no statement of the proxy timeout requirement.

Changes:
- add docs/runbook.md: startup, deployment steps, provisioning, the INGESTION_*
  tuning table with each bound's status code, the proxy read-timeout rule,
  /healthz vs /readyz, failure investigation by real event name plus job/event
  SQL, retry semantics, and alert thresholds
- link it from the README and note provisioning there
- mark plan 001 Phase 6 done and refresh CLAUDE.md's status paragraph

Impact:
- documentation only

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

CLAUDE.md

@@ -27,9 +27,16 @@ Also working: `tenant_domains` plus `/v1/domains` (`src/application/domains/`),
 a strict per-tenant allowlist — `POST /v1/files` rejects an unregistered or
 disabled `domain` with `400` before anything is written, and domain management
 sits behind its own `domains:read`/`domains:write` scopes, never `files:write`.
-Not built yet: `/v1/points` CRUD and keyword search (plan 002), and
-`src/agent/`. That maps to plan 001 Phases 1-5 done, Phase 6 (runbook, e2e
-tests, Compose smoke test) not started.
+Also working: the operator runbook (`docs/runbook.md`), tenant/API-key/domain
+provisioning (`uv run python -m src.cli.provision_tenant` — the third deployment
+step, since nothing over HTTP can create the first tenant), a Testcontainers
+e2e suite in the default pytest run (`tests/e2e/test_ingestion_slice.py`:
+duplicate upload, retry after failure, tenant isolation, capacity, timeout,
+parse and Qdrant failure), and the one Compose-based test — `scripts/smoke.sh`
+driving `tests/e2e/test_compose_smoke.py` against a real uvicorn process, which
+skips itself unless `SMOKE_BASE_URL` is set. Not built yet: `/v1/points` CRUD
+and keyword search (plan 002), and `src/agent/`. That maps to plan 001 Phases
+1-6 done.
 
 Architecture decisions live in `docs/adr/` (18 ADRs plus the 0000 template;
 0001–0004 are `Accepted` — 0004 amended by 0018; 0014 is `Superseded by 0017`;
@@ -285,9 +292,14 @@ belongs, not the row-level function underneath it.
 - Test naming: `test_<unit>_<scenario>_<outcome>`, Arrange–Act–Assert.
 - Layout mirrors architecture: `tests/unit/{application,agent}`,
   `tests/integration/{postgres,minio,qdrant}`, `tests/e2e/`.
-- Integration tests use **Testcontainers** (never a developer's local
-  services or Langfuse-owned storage/credentials) — this is the standard
-  automated mechanism, not Docker Compose. Isolate data per test via unique
+- Integration **and e2e** tests use **Testcontainers** (never a developer's
+  local services or Langfuse-owned storage/credentials) — this is the standard
+  automated mechanism, not Docker Compose. Compose is reserved for exactly one
+  thing: the serialized operational smoke test of the *running web process*
+  (`scripts/smoke.sh`), which is gated out of `uv run pytest`. Shared container
+  fixtures live in `tests/support/containers.py`, registered from the root
+  `tests/conftest.py` via `pytest_plugins` (a non-root conftest cannot declare
+  it). Isolate data per test via unique
   keys/queue/collection names; parallel integration execution is disabled
   until fixture isolation is proven safe.
 - Pytest never calls a live/paid model provider in routine runs — that's

README.md

@@ -2,6 +2,9 @@
 
 Architecture decisions live in [`docs/adr`](docs/adr). The first implementation
 milestone is documented in the [ingestion vertical-slice plan](docs/plans/001-ingestion-vertical-slice.md).
+Day-to-day operation — tuning the ingestion bounds, the proxy timeout
+requirement, and how to investigate or retry a failed upload — is the
+[operator runbook](docs/runbook.md).
 
 ## Provisioning the datastores
 
@@ -16,6 +19,14 @@ uv run python -m src.cli.qdrant_bootstrap     # the `chunks` collection
 uv run fastapi dev src/main.py
 ```
 
+Nothing over HTTP can create the first tenant — every `/v1` route needs an API
+key, and a key cannot exist before its tenant. One command issues both, plus any
+domains, printing the key once (only its hash is stored):
+
+```bash
+uv run python -m src.cli.provision_tenant --slug acme --domain fire
+```
+
 Before a tenant can upload, its domains must be registered — `POST /v1/files`
 rejects an unregistered or disabled `domain` with `400`. The calling backend
 manages them over `/v1/domains` using a key with the `domains:write` scope:
@@ -31,6 +42,10 @@ Both bootstrap commands are idempotent and safe to re-run. `qdrant_bootstrap` ve
 existing collection against the pinned schema and exits non-zero on a mismatch,
 rather than leaving a silently degraded sparse index in place.
 
+`./scripts/smoke.sh` verifies the whole path — Compose up, both deployment
+steps, provisioning, an upload through the running web process to indexed Qdrant
+points. See the [runbook](docs/runbook.md#12-verifying-a-deployment).
+
 ## Local Langfuse
 
 This repo includes a root-level development Compose file for Langfuse:

docs/plans/001-ingestion-vertical-slice.md

@@ -289,17 +289,28 @@ retrying the upload produces a correct final state without duplicate chunks.
 
 1. Add an operator runbook covering local startup, migrations, MinIO bucket
    setup, the run command, ingestion-bound tuning, the proxy/client timeout
-   requirement, and how to retry a failed ingestion.
+   requirement, and how to retry a failed ingestion. — `docs/runbook.md`.
 2. Add a serialized Compose-based operational smoke test covering upload through
    indexed points against the running web process. Testcontainers remains the
-   standard pytest mechanism for individual adapter integration tests.
+   standard pytest mechanism for individual adapter integration tests. —
+   `scripts/smoke.sh` driving `tests/e2e/test_compose_smoke.py`, which skips
+   itself unless `SMOKE_BASE_URL` is set so `uv run pytest` never invokes
+   Compose.
 3. Add end-to-end tests for duplicate upload, retrying a failed upload, tenant
-   isolation, capacity/timeout rejection, and failed parser/Qdrant behavior.
+   isolation, capacity/timeout rejection, and failed parser/Qdrant behavior. —
+   `tests/e2e/test_ingestion_slice.py`, on Testcontainers, in the default suite.
 4. Add health/readiness checks that distinguish process health from dependency
-   readiness.
+   readiness. — `/healthz` and `/readyz`; `/readyz` additionally requires the
+   `chunks` collection to exist, since a reachable but unbootstrapped Qdrant
+   would `502` on the first upload.
 5. Update the README with local-start instructions and links to ADRs, this plan,
    and the operations runbook.
 
+Provisioning a tenant and its first API key turned out to be a prerequisite for
+1 and 2 rather than a separate milestone: nothing over HTTP can create the first
+tenant, so `src/cli/provision_tenant.py` was added alongside the other two
+deployment-step commands.
+
 **Exit criteria:** a new developer can start the stack, apply migrations, upload a
 a document, observe the job through completion, and understand how to investigate or
 retry a failure.

docs/runbook.md

@@ -0,0 +1,291 @@
+# Operator runbook
+
+How to start this service, configure its ingestion bounds, and investigate or
+retry a failed upload. Architecture rationale lives in [`docs/adr/`](adr/); the
+implementation milestone is
+[plan 001](plans/001-ingestion-vertical-slice.md). This document covers
+operating what those describe.
+
+The service is a **single process with no background work**. `POST /v1/files`
+parses, chunks, embeds, and indexes inline and returns a terminal result
+(ADR-0017). There is no queue, no worker, and no automatic retry — the caller
+owns the retry decision, which makes the request's duration a deployment
+constraint. That fact drives most of this document.
+
+## 1. Prerequisites and local startup
+
+Docker, and [`uv`](https://docs.astral.sh/uv/) with Python 3.13.
+
+```bash
+cp .env.example .env      # non-secret local defaults; .env is gitignored
+uv sync
+docker compose up -d --wait
+```
+
+`docker-compose.yml` runs Postgres (`127.0.0.1:5433`), MinIO
+(`127.0.0.1:9100`, console `9101`), and Qdrant (`127.0.0.1:6343`). It is the
+local development stack and says so in its header — it is not a production
+deployment.
+
+## 2. Deployment steps
+
+Two schema steps run **before** the application, never at startup: FastAPI
+performs no DDL, for Postgres (ADR-0009) or for Qdrant (ADR-0001, "Collection
+provisioning"). Both commands and their reasoning are in the README's
+[Provisioning the datastores](../README.md#provisioning-the-datastores)
+section:
+
+```bash
+uv run alembic upgrade head                   # Postgres schema
+uv run python -m src.cli.qdrant_bootstrap     # the `chunks` collection
+```
+
+Both are idempotent. `qdrant_bootstrap` verifies an existing collection against
+the pinned schema and **exits non-zero on a mismatch** rather than leaving a
+silently degraded sparse index in place — the `sparse` vector's
+`modifier="idf"` and the pinned dense dimensions (768 / 3072) fail silently if
+wrong, which is why they are checked rather than assumed.
+
+Run both again after every deploy that ships a migration or a collection-schema
+change.
+
+## 3. MinIO bucket
+
+Under Compose the bucket already exists: the `app-minio` service's entrypoint
+runs `mkdir -p /data/${MINIO_BUCKET:-chatbot-source-files}` before starting the
+server, so first boot creates it. Nothing else needs to be done locally.
+
+Outside Compose, create the bucket named by `MINIO_BUCKET` before the first
+upload — the application never creates it. It must stay **private**; ADR-0013
+keeps source bytes non-public and this slice ships no download API or presigned
+URLs.
+
+## 4. Provisioning a tenant, an API key, and its domains
+
+Nothing over HTTP can bootstrap a tenant: every `/v1` route needs an API key,
+and a key cannot exist before its tenant. So the first key is issued by an
+operator command:
+
+```bash
+uv run python -m src.cli.provision_tenant \
+  --slug acme --domain fire --domain life --scopes files:write,domains:read
+```
+
+It prints `api_key=sk_...` **once**. Postgres stores only its SHA-256 hash
+(ADR-0009), so a lost key is reissued by re-running the command, never
+recovered. Structured logs carry only the non-secret `key_prefix` — a plaintext
+key must never reach a log sink (ADR-0011).
+
+Re-running with the same `--slug` reuses the tenant and any domains it already
+has, and issues an **additional** key. Both keys stay valid; this adds a key, it
+does not rotate one.
+
+Scopes are the security boundary between uploading and managing the allowlist.
+Give an upload client `files:write` only. `domains:write` lets its holder create
+new domains, which is exactly what the allowlist exists to prevent an upload key
+from doing.
+
+### Domains after the first one
+
+`POST /v1/files` rejects an unregistered or disabled `domain` with `400`
+(`unknown_domain`) before anything is written. Ongoing domain management is the
+`/v1/domains` API, under `domains:read` / `domains:write`:
+
+```bash
+curl -X POST http://localhost:8000/v1/domains \
+  -H "Authorization: Bearer $API_KEY" -H 'Content-Type: application/json' \
+  -d '{"domain": "fire", "display_name": "Fire insurance"}'
+```
+
+The `domain` key itself is immutable — it is denormalized into every Qdrant
+point payload and into `source_files`, so renaming it is a migration, not an
+edit (ADR-0009). Disabling a domain blocks new uploads; it does not delete
+existing points.
+
+## 5. Running the service
+
+```bash
+uv run fastapi dev src/main.py                                   # local, reload
+uv run uvicorn src.main:app --host 0.0.0.0 --port 8000           # deployed shape
+```
+
+Run more than one worker/replica only after reading §6: ingestion bounds are
+**per process**, so `INGESTION_MAX_CONCURRENCY` multiplies by the number of
+processes.
+
+## 6. Ingestion bounds and tuning
+
+Every bound is enforced server-side and maps to a status code. All are in
+`.env.example`. Ingestion is CPU- and network-bound in the request, so these are
+the numbers that decide whether the service degrades gracefully or falls over.
+
+| Setting | Bounds | On breach | Size it against |
+|---|---|---|---|
+| `INGESTION_MAX_CONCURRENCY` | Ingestions in flight **per process** | `503` + `Retry-After` | Memory per in-flight upload (whole file plus its chunks and vectors are resident) and the embedder's capacity. Rejecting is deliberate: ADR-0017 refuses rather than queues. |
+| `INGESTION_THREAD_POOL_SIZE` | Threads for blocking work (parse, chunk, hash, BM25, the sync `minio` SDK) | — (waits) | CPU cores. It exists to stop ingestion exhausting Starlette's own thread pool, so keep it below the total thread budget. |
+| `INGESTION_TIMEOUT_SECONDS` | The whole work phase | `504`, job marked `failed` | The slowest legitimate document, plus headroom. See §7 — this must stay under every read timeout in front of it. |
+| `INGESTION_MAX_UPLOAD_SIZE_MB` | Bytes accepted | `413` | Memory: the upload is read fully into the process before any work starts. |
+| `INGESTION_MAX_CHUNKS_PER_FILE` | Chunks per file, checked before embedding | `413` | Embedder cost/time per chunk × `INGESTION_TIMEOUT_SECONDS`. This is the real defence against one pathological file eating a slot. |
+| `INGESTION_EMBED_BATCH_SIZE` | Texts per embedder request | `502` on embedder failure | The provider's per-request limits. Batch before parallelizing. |
+| `INGESTION_EMBED_CONCURRENCY` | Concurrent embed batches | `502` | Provider rate limits and the self-hosted embedder's throughput. Never unbounded. |
+| `QDRANT_UPSERT_BATCH_SIZE` / `_CONCURRENCY` | Points per upsert and concurrent upserts | `502` (`index_error`) | Qdrant's ingest capacity; the batch size stays in ADR-0001's 64–256 band. |
+
+Two settings that look like tuning knobs but are not:
+
+- **`EMBEDDING_NOMIC_KEEP_ALIVE`** holds the self-hosted model resident. A cold
+  load of `nomic-embed-text-v2-moe` takes over 150 s — longer than any sane
+  `INGESTION_TIMEOUT_SECONDS` — so an idle period followed by an upload would
+  otherwise `504`. The lifespan also warms both dense embedders at startup for
+  the same reason.
+- **The BM25 analyzer and weights** (`EMBEDDING_SPARSE_*`) are a measured
+  artifact ported from the `emet` evaluation lab, verified token-for-token
+  against it (ADR-0005). Re-benchmark; do not tune them in place.
+
+## 7. The proxy and client read-timeout requirement
+
+**Every read timeout in front of this service must exceed
+`INGESTION_TIMEOUT_SECONDS`.** That includes the reverse proxy / ingress, any
+load balancer, and the calling backend's own HTTP client.
+
+If a proxy times out first, the client gets that proxy's error, the upload keeps
+running in the process, and the caller learns nothing about the outcome from the
+response. The job row still reaches a terminal status, so
+`GET /v1/files/{file_id}` remains the way to find out what happened — but the
+response contract is broken for that request. ADR-0017 names this the main cost
+of inline ingestion.
+
+A workable local ordering: client read timeout > proxy read timeout >
+`INGESTION_TIMEOUT_SECONDS`.
+
+## 8. Health and readiness
+
+| Endpoint | Question it answers | Use for |
+|---|---|---|
+| `GET /healthz` | Is the process alive? | Liveness probes / restart policy. Never depends on Postgres, MinIO, or Qdrant. |
+| `GET /readyz` | Can it actually serve? | Load-balancer admission and post-deploy gating. `200` with each dependency `true`, `503` if any is `false`. |
+
+`/readyz` checks Postgres, MinIO, and Qdrant reachability **and** that the
+`chunks` collection exists. A reachable-but-unbootstrapped Qdrant reports
+`{"qdrant": false}` on purpose: uploads to it would fail with `502`, so it is
+not ready, and this is how a skipped `qdrant_bootstrap` surfaces at deploy time
+instead of on a user's first upload.
+
+## 9. Investigating a failure
+
+Logs are structured (`structlog`, JSON in production) with stable event names —
+grep the event name, not prose (ADR-0011). Set `LOG_FILE_PATH` for a local
+JSON file sink alongside the console renderer; leave it unset in production,
+where stdout collection is preferred.
+
+**Correlate by `request_id`.** Every request has one, echoed in the
+`X-Request-Id` response header and included in every error envelope, and bound
+into every log line emitted while handling that request. A client reporting a
+failed upload should quote it. `tenant_id`, `file_id`, and `ingestion_job_id`
+are the other join keys.
+
+Events worth knowing:
+
+| Event | Level | Means |
+|---|---|---|
+| `ingestion.job.started` | info | Txn A committed; work phase beginning. Carries `tenant_id`, `ingestion_job_id`, `file_id`, `domain`, `source_type`. |
+| `ingestion.job.completed` | info | Terminal success, with `chunks_parsed`, `points_upserted`, `points_soft_deleted`. |
+| `ingestion.job.failed` | warning | Terminal failure. **`error_code` says which stage**: `storage_upload_failed`, `parse_failed`, `chunk_limit_exceeded`, `embedding_failed`, `index_failed`, `timeout`. |
+| `files.upload.duplicate` | info | Identical content already ingested; the existing file/job was returned and nothing was re-ingested. |
+| `domain.rejected` | warning | Upload refused before any row was written; `reason` is `unregistered` or `disabled`. |
+| `auth.failed` | warning | `reason` is `malformed_key`, `unknown_key`, `key_inactive`, `key_expired`, or `tenant_inactive`. Never contains key material. |
+| `auth.succeeded` | info | Carries `tenant_id`, `api_key_id`, `actor_type`. |
+| `lifespan.embedder.warm_failed` | warning | An embedder was unreachable at startup. Boot continues by design — `/readyz` and the first upload are where this bites. |
+| `qdrant.bootstrap.schema_mismatch` | error | The existing collection diverges from the pinned schema. The bootstrap exits non-zero; do not start the app against it. |
+| `api.unhandled_exception` | error | A bug: an exception with no mapping to the error envelope. Always worth a look. |
+
+A `503` (`ingestion_at_capacity`) is rejected before a job row exists, so it
+appears in the access log and metrics, not in `ingestion_jobs`.
+
+### The durable record
+
+Logs may roll; `ingestion_jobs` and `ingestion_job_events` do not. For one file:
+
+```sql
+SELECT id, status, error_code, error_message, points_created, points_soft_deleted,
+       created_at, updated_at
+FROM ingestion_jobs
+WHERE tenant_id = :tenant_id AND source_file_id = :file_id
+ORDER BY created_at DESC;
+
+SELECT stage, level, message, details, created_at
+FROM ingestion_job_events
+WHERE tenant_id = :tenant_id AND ingestion_job_id = :ingestion_job_id
+ORDER BY created_at;
+```
+
+Recent failures across a tenant:
+
+```sql
+SELECT error_code, count(*), max(created_at)
+FROM ingestion_jobs
+WHERE tenant_id = :tenant_id AND status = 'failed' AND created_at > now() - interval '1 day'
+GROUP BY error_code ORDER BY 2 DESC;
+```
+
+`GET /v1/files/{file_id}` reports the same terminal status over HTTP, scoped to
+the owning tenant — a file belonging to another tenant returns `404`, not `403`.
+
+## 10. Retrying a failed ingestion
+
+**Re-upload the same bytes.** There is no retry endpoint and no automatic retry;
+the client owns that decision (ADR-0017).
+
+What that guarantees:
+
+- Identical content with a **succeeded** job is recognized by
+  `(tenant_id, domain, content_sha256)` and returned as-is with `200` — no
+  re-ingestion, no duplicate points.
+- Identical content whose last job **failed** starts a fresh job against the
+  same `source_files` row. A terminal job is never moved back to `running`.
+- Point ids are deterministic from `file_id` + `chunk_index` (ADR-0001), so the
+  retry **overwrites in place** — it cannot duplicate chunks.
+- A failed attempt never empties or partially deletes a working index: the
+  soft-delete sweep that retires a shortened file's leftover points runs only
+  after every upsert has succeeded. An interrupted attempt can leave a prefix
+  updated; a retry converges (ADR-0017, "Re-running an ingestion stays safe").
+
+Fix the cause first — the `error_code` says where to look:
+
+| `error_code` | Usual cause |
+|---|---|
+| `parse_failed` | The file is corrupt or is not really the type its extension claims. Retrying identical bytes will fail identically. |
+| `chunk_limit_exceeded` | The file is genuinely too large for one inline ingestion. Split it, or raise `INGESTION_MAX_CHUNKS_PER_FILE` knowing what §6 says about the timeout. |
+| `embedding_failed` | The embedder is down, rate-limiting, or unauthenticated. Fix it, then retry — this one usually succeeds unchanged. |
+| `index_failed` | Qdrant is down, or the collection is missing (run `qdrant_bootstrap`). |
+| `timeout` | The work exceeded `INGESTION_TIMEOUT_SECONDS`. Check whether the embedder was cold (see `lifespan.embedder.warm_failed` and `KEEP_ALIVE`) before raising the bound. |
+| `storage_upload_failed` | MinIO is unreachable or the bucket is missing (§3). |
+
+## 11. What to alert on
+
+ADR-0017's own triggers for moving ingestion back off the request path. These
+are the numbers that say the inline design has stopped fitting:
+
+- **p95 ingestion duration** approaching `INGESTION_TIMEOUT_SECONDS`.
+- **`503` and `504` rates** ceasing to be negligible.
+- **Jobs stuck in `running` past the timeout** — every handled failure writes a
+  terminal status, so a non-zero count here means the process died mid-request:
+
+  ```sql
+  SELECT count(*) FROM ingestion_jobs
+  WHERE status = 'running' AND created_at < now() - interval '5 minutes';
+  ```
+
+Also worth alerting: any `qdrant.bootstrap.schema_mismatch`, a sustained
+`/readyz` `503`, and any `api.unhandled_exception`.
+
+## 12. Verifying a deployment
+
+```bash
+./scripts/smoke.sh
+```
+
+Brings up Compose, runs both deployment steps, provisions a throwaway tenant,
+starts the web process, and drives an upload through to indexed Qdrant points
+against the **running process** — including asserting the structured log output
+from §9. It is the only Compose-based test; everything else runs on
+Testcontainers under `uv run pytest` (ADR-0016). Run it before a release.

scripts/smoke.sh

@@ -0,0 +1,98 @@
+#!/usr/bin/env bash
+# Serialized operational smoke test of the running web process (ADR-0016, plan
+# 001 Phase 6).
+#
+#   ./scripts/smoke.sh
+#
+# Brings up the Compose stack, runs both deployment steps for real, provisions
+# a tenant, starts uvicorn, and drives `tests/e2e/test_compose_smoke.py`
+# against it over a socket. This is the only Compose-based test: every other
+# test uses Testcontainers and an in-process ASGI transport, which is exactly
+# what makes this one worth having -- it is the only thing that exercises the
+# deployment steps, the real logging configuration, and a real HTTP server.
+#
+# Not part of `uv run pytest`: the smoke test skips itself unless SMOKE_BASE_URL
+# is set, so this script is the only way it runs. Run it before a release.
+#
+# Leaves the Compose stack running (it is the local dev stack); only the uvicorn
+# process and the temporary log file are cleaned up.
+set -euo pipefail
+
+cd "$(dirname "${BASH_SOURCE[0]}")/.."
+
+PORT="${SMOKE_PORT:-8021}"
+SLUG="smoke-$(date +%s)"
+DOMAIN="smoke"
+LOG_FILE="$(mktemp -t smoke-app-log.XXXXXX.jsonl)"
+CONSOLE_LOG="$(mktemp -t smoke-app-console.XXXXXX.log)"
+APP_PID=""
+
+cleanup() {
+  if [[ -n "${APP_PID}" ]] && kill -0 "${APP_PID}" 2>/dev/null; then
+    kill "${APP_PID}" 2>/dev/null || true
+    wait "${APP_PID}" 2>/dev/null || true
+  fi
+  rm -f "${LOG_FILE}" "${CONSOLE_LOG}"
+}
+trap cleanup EXIT
+
+if [[ ! -f .env ]]; then
+  echo "no .env found; copy .env.example first (see docs/runbook.md)" >&2
+  exit 1
+fi
+
+echo "==> starting Postgres, MinIO, Qdrant"
+docker compose up -d --wait
+
+echo "==> applying deployment steps"
+uv run alembic upgrade head
+uv run python -m src.cli.qdrant_bootstrap
+
+echo "==> provisioning tenant '${SLUG}'"
+PROVISION_OUTPUT="$(uv run python -m src.cli.provision_tenant \
+  --slug "${SLUG}" --domain "${DOMAIN}" --scopes files:write 2>/dev/null)"
+API_KEY="$(printf '%s\n' "${PROVISION_OUTPUT}" | sed -n 's/^api_key=//p')"
+if [[ -z "${API_KEY}" ]]; then
+  echo "provisioning did not return an api_key" >&2
+  exit 1
+fi
+
+echo "==> starting the web process on port ${PORT}"
+# JSON to a file sink, because the smoke test asserts the real ADR-0011 log
+# output -- the one thing no in-process test can check.
+LOG_JSON_FORMAT=true LOG_FILE_PATH="${LOG_FILE}" \
+  uv run python -m uvicorn src.main:app --host 127.0.0.1 --port "${PORT}" \
+  >"${CONSOLE_LOG}" 2>&1 &
+APP_PID=$!
+
+echo "==> waiting for /readyz"
+for _ in $(seq 1 60); do
+  if curl -fsS "http://127.0.0.1:${PORT}/readyz" >/dev/null 2>&1; then
+    break
+  fi
+  if ! kill -0 "${APP_PID}" 2>/dev/null; then
+    echo "the web process exited before becoming ready:" >&2
+    tail -20 "${CONSOLE_LOG}" >&2
+    exit 1
+  fi
+  sleep 1
+done
+
+if ! curl -fsS "http://127.0.0.1:${PORT}/readyz" >/dev/null 2>&1; then
+  # Most often an unbootstrapped Qdrant or an unreachable embedder host; the
+  # runbook's health/readiness section covers reading this.
+  echo "the web process never became ready:" >&2
+  tail -20 "${CONSOLE_LOG}" >&2
+  exit 1
+fi
+
+echo "==> running the smoke test"
+SMOKE_BASE_URL="http://127.0.0.1:${PORT}" \
+SMOKE_API_KEY="${API_KEY}" \
+SMOKE_DOMAIN="${DOMAIN}" \
+SMOKE_LOG_PATH="${LOG_FILE}" \
+SMOKE_QDRANT_URL="${QDRANT_URL:-http://127.0.0.1:6343}" \
+SMOKE_QDRANT_COLLECTION="${QDRANT_COLLECTION:-chunks}" \
+  uv run python -m pytest tests/e2e/test_compose_smoke.py -q
+
+echo "==> smoke test passed"

src/api/errors.py

@@ -97,7 +97,7 @@ def register_exception_handlers(app: FastAPI) -> None:
     @app.exception_handler(RequestValidationError)
     def _validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
         return JSONResponse(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
             content=_envelope(
                 "validation_error",
                 "request validation failed",

src/application/tenants/__init__.py

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

src/application/tenants/provisioning.py

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

src/cli/provision_tenant.py

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

src/infrastructure/postgres/repositories/api_keys.py

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

src/infrastructure/postgres/repositories/tenants.py

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

tests/conftest.py

@@ -10,6 +10,12 @@ from httpx import ASGITransport, AsyncClient
 from src.config import Settings
 from src.main import create_app
 
+# Disposable Postgres/MinIO/Qdrant containers (ADR-0016). Registered here
+# because `pytest_plugins` is only honoured in the root conftest, and both
+# `tests/integration/*/` and `tests/e2e/` need the same containers. The
+# fixtures are session-scoped but lazy, so unit runs still need no Docker.
+pytest_plugins = ["tests.support.containers"]
+
 
 @pytest.fixture(autouse=True)
 def _reset_structlog_after_test() -> Iterator[None]:

tests/e2e/conftest.py

@@ -0,0 +1,244 @@
+"""The vertical slice against real Postgres, MinIO, and Qdrant (ADR-0016).
+
+These are the ADR-0016 end-to-end tests: "a small vertical-slice acceptance
+contract through the real application composition". They run on
+Testcontainers, in the default `uv run pytest` run, because ADR-0016 makes
+Testcontainers "the standard automated integration-test resource mechanism"
+and reserves Docker Compose for manual validation and the separate serialized
+smoke test of the *running web process* (`tests/e2e/test_compose_smoke.py`).
+
+What is real here: routing, auth, scopes, the domain allowlist, the ADR-0008
+error envelope, the two-transaction upload, MinIO object writes, BM25 sparse
+embedding, and Qdrant upserts into a per-test collection created by the
+production `ensure_chunks_collection`.
+
+What is faked, and only this: the two dense embedders. They are paid/remote
+network calls, and ADR-0016 forbids routine runs from touching a live
+provider. Their fakes return the pinned 768/3072 dimensions, because the real
+collection rejects anything else.
+"""
+
+import uuid
+from collections.abc import AsyncIterator, Callable, Coroutine
+from contextlib import AsyncExitStack
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+import pytest_asyncio
+from asgi_lifespan import LifespanManager
+from httpx import ASGITransport, AsyncClient
+from qdrant_client import AsyncQdrantClient, models
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+
+from src.bootstrap.dependencies import get_dense_embedders, get_sessionmaker
+from src.config import MinioSettings, PostgresSettings, QdrantSettings, Settings
+from src.infrastructure.postgres.database import create_sessionmaker
+from src.infrastructure.qdrant.collection import (
+    DENSE_NOMIC_DIMENSIONS,
+    DENSE_OPENAI_DIMENSIONS,
+    ensure_chunks_collection,
+)
+from src.main import create_app
+from tests.fakes import FakeDenseEmbedder
+from tests.support.factories import create_api_key, create_tenant, create_tenant_domain
+
+CSV_BYTES = b"question,answer\nhow do I file a claim,call the branch\nwhat is covered,see policy\n"
+
+# Past validation's OOXML magic-byte check, so the failure lands in the DOCX
+# parser rather than in upload validation -- the branch this exercises.
+CORRUPT_DOCX_BYTES = b"PK\x03\x04" + b"not really a docx" * 8
+
+
+@dataclass
+class E2EStack:
+    """One configured app instance plus the handles a test asserts against."""
+
+    client: AsyncClient
+    collection: str
+    dense_embedders: list[FakeDenseEmbedder]
+
+
+StackFactory = Callable[..., Coroutine[Any, Any, E2EStack]]
+
+
+@dataclass(frozen=True)
+class Principal:
+    """A provisioned tenant with a bearer token and a registered domain.
+
+    Plain values, not the ORM `Tenant`: tests re-read rows the app committed on
+    a shared connection, and any expiry/refresh of a held ORM instance would
+    lazy-load outside a greenlet context (`MissingGreenlet`).
+    """
+
+    tenant_id: uuid.UUID
+    tenant_slug: str
+    token: str
+    domain: str
+
+    @property
+    def headers(self) -> dict[str, str]:
+        return {"Authorization": f"Bearer {self.token}"}
+
+
+@pytest.fixture
+def e2e_settings(
+    postgres_settings: PostgresSettings,
+    minio_settings: MinioSettings,
+    qdrant_settings: QdrantSettings,
+) -> Settings:
+    """Settings wired to this test's containers.
+
+    Every nested settings object is passed explicitly. `src/config.py` cascades
+    `.env` into each nested class, so omitting one would let a developer's real
+    `EMBEDDING_NOMIC_BASE_URL` (a colleague's Ollama box) into a routine test
+    run. The embedder URLs below point at a closed port for the same reason:
+    the lifespan warms the *real* embedders before the fakes are substituted,
+    and connection-refused is both instant and free.
+    """
+    return Settings(
+        postgres=postgres_settings,
+        minio=minio_settings,
+        qdrant=qdrant_settings,
+        embedding={
+            "nomic": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
+            "openai": {"base_url": "http://127.0.0.1:1/v1", "timeout_seconds": 0.5},
+        },
+        app={"readiness_check_timeout_seconds": 2.0},
+    )
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def make_stack(
+    e2e_settings: Settings,
+    db_sessionmaker: async_sessionmaker[AsyncSession],
+    qdrant_client: AsyncQdrantClient,
+) -> AsyncIterator[StackFactory]:
+    """Build an app under test, optionally with tightened ingestion bounds.
+
+    A factory rather than a fixture because the capacity and timeout tests need
+    their own `INGESTION_*` values, and those are read at app construction.
+    """
+    async with AsyncExitStack() as exit_stack:
+
+        async def _make(
+            *,
+            ingestion: dict[str, object] | None = None,
+            collection: str | None = None,
+            dense_delay_seconds: float = 0.0,
+            bootstrap_collection: bool = True,
+            sessionmaker: async_sessionmaker[AsyncSession] | None = None,
+        ) -> E2EStack:
+            resolved_collection = collection or e2e_settings.qdrant.collection
+            if bootstrap_collection:
+                await ensure_chunks_collection(qdrant_client, collection=resolved_collection)
+
+            settings = e2e_settings.model_copy(
+                update={
+                    "qdrant": e2e_settings.qdrant.model_copy(
+                        update={"collection": resolved_collection}
+                    ),
+                    "ingestion": e2e_settings.ingestion.model_copy(update=ingestion or {}),
+                }
+            )
+
+            dense = [
+                FakeDenseEmbedder(
+                    name="dense_nomic",
+                    dimensions=DENSE_NOMIC_DIMENSIONS,
+                    value=0.1,
+                    delay_seconds=dense_delay_seconds,
+                ),
+                FakeDenseEmbedder(
+                    name="dense_openai",
+                    dimensions=DENSE_OPENAI_DIMENSIONS,
+                    value=0.2,
+                    delay_seconds=dense_delay_seconds,
+                ),
+            ]
+
+            app = create_app(settings)
+            # The session factory is the test's, so factory-created tenants are
+            # visible to the app and every write rolls back at teardown. Object
+            # storage, point storage, and the sparse embedder stay real.
+            resolved_sessionmaker = sessionmaker or db_sessionmaker
+            app.dependency_overrides[get_sessionmaker] = lambda: resolved_sessionmaker
+            app.dependency_overrides[get_dense_embedders] = lambda: dense
+
+            manager = await exit_stack.enter_async_context(LifespanManager(app))
+            client = await exit_stack.enter_async_context(
+                AsyncClient(
+                    transport=ASGITransport(app=manager.app),
+                    base_url="http://test",
+                    timeout=30.0,
+                )
+            )
+            return E2EStack(client=client, collection=resolved_collection, dense_embedders=dense)
+
+        yield _make
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def committed_sessionmaker(
+    postgres_engine: AsyncEngine,
+) -> async_sessionmaker[AsyncSession]:
+    """A session factory whose writes really commit, one connection each.
+
+    The default `db_sessionmaker` pins every session to a single connection
+    inside one rolled-back transaction, which is what isolates a test's writes
+    -- but savepoints on a shared connection cannot interleave, so any test
+    that issues genuinely *concurrent* requests deadlocks or fails with
+    `InvalidSavepointSpecificationError`. Those tests use this instead and rely
+    on their uuid-keyed tenant for isolation (ADR-0016 allows either).
+    """
+    return create_sessionmaker(postgres_engine)
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def principal(db_session: AsyncSession) -> Principal:
+    """A tenant with an upload-scoped key and one registered domain.
+
+    Registering the domain is not optional: `POST /v1/files` rejects an
+    unregistered one with `400 unknown_domain` before anything else runs
+    (ADR-0009).
+    """
+    return await provision_principal(db_session)
+
+
+async def provision_principal(
+    session: AsyncSession, *, domain: str = "fire", scopes: list[str] | None = None
+) -> Principal:
+    tenant = await create_tenant(session)
+    _, token = await create_api_key(session, tenant=tenant, scopes=scopes or ["files:write"])
+    await create_tenant_domain(session, tenant=tenant, domain=domain)
+    await session.commit()
+    return Principal(tenant_id=tenant.id, tenant_slug=tenant.slug, token=token, domain=domain)
+
+
+def upload_payload(
+    *, domain: str, filename: str = "faq.csv", data: bytes = CSV_BYTES
+) -> dict[str, Any]:
+    return {"files": {"file": (filename, data, "text/csv")}, "data": {"domain": domain}}
+
+
+async def count_active_points(
+    client: AsyncQdrantClient, collection: str, *, tenant_id: uuid.UUID
+) -> int:
+    """Active points for one tenant, read through the raw client.
+
+    `PointStorage` is deliberately write-only (point reads are plan 002's
+    `/v1/points`), so the verification read here uses the SDK directly.
+    """
+    result = await client.count(
+        collection_name=collection,
+        count_filter=models.Filter(
+            must=[
+                models.FieldCondition(
+                    key="tenant_id", match=models.MatchValue(value=str(tenant_id))
+                ),
+                models.FieldCondition(key="is_active", match=models.MatchValue(value=True)),
+            ]
+        ),
+        exact=True,
+    )
+    return result.count

tests/e2e/test_compose_smoke.py

@@ -0,0 +1,143 @@
+"""Operational smoke test against the *running web process* (ADR-0016).
+
+Different in kind from every other test in this repo. Everything else drives
+the app in-process through `ASGITransport`; this drives a real uvicorn process
+over a real socket, backed by the Docker Compose stack, after the two
+deployment steps (`alembic upgrade head`, `src.cli.qdrant_bootstrap`) have run
+for real. ADR-0016 reserves Compose for exactly this: "manual local validation
+and a later, serialized operational smoke test of the running web process",
+run "as a serialized pre-release or scheduled gate".
+
+It is therefore **skipped unless `SMOKE_BASE_URL` is set**, so a plain
+`uv run pytest` never touches Docker Compose. Run it through `scripts/smoke.sh`,
+which brings the stack up, provisions a tenant, starts the process, and exports
+the environment below.
+
+Because the app runs in its own process, this is also the only place in the
+suite where the real `configure_logging()` is in effect --
+`tests/conftest.py::_no_real_logging_configuration` no-ops it everywhere else
+to keep global structlog state out of other tests. So the ADR-0011 log sink is
+asserted here and nowhere else.
+"""
+
+import json
+import os
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+from httpx import AsyncClient
+from qdrant_client import AsyncQdrantClient, models
+
+BASE_URL = os.environ.get("SMOKE_BASE_URL")
+API_KEY = os.environ.get("SMOKE_API_KEY", "")
+DOMAIN = os.environ.get("SMOKE_DOMAIN", "smoke")
+LOG_PATH = os.environ.get("SMOKE_LOG_PATH", "")
+QDRANT_URL = os.environ.get("SMOKE_QDRANT_URL", "http://127.0.0.1:6343")
+QDRANT_COLLECTION = os.environ.get("SMOKE_QDRANT_COLLECTION", "chunks")
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.slow,
+    pytest.mark.asyncio,
+    pytest.mark.skipif(
+        not BASE_URL,
+        reason="SMOKE_BASE_URL is unset; run this through scripts/smoke.sh",
+    ),
+    # Well past the global 10s budget: this drives a real process over a
+    # socket, and the upload does real embedding work end to end.
+    pytest.mark.timeout(120),
+]
+
+# Unique per run so a re-run against the same persistent Compose volumes is a
+# new file rather than an idempotent duplicate-hash hit (ADR-0017).
+_CSV = f"question,answer\nsmoke run {uuid.uuid4().hex},indexed\n".encode()
+
+
+@pytest_asyncio.fixture
+async def client() -> AsyncIterator[AsyncClient]:
+    async with AsyncClient(base_url=BASE_URL or "", timeout=120.0) as async_client:
+        yield async_client
+
+
+async def test_running_process_reports_healthy_and_ready() -> None:
+    """Readiness must be green *after* the deployment steps, not before --
+    an unbootstrapped Qdrant is deliberately unready.
+    """
+    async with AsyncClient(base_url=BASE_URL or "", timeout=30.0) as client:
+        healthz = await client.get("/healthz")
+        readyz = await client.get("/readyz")
+
+    assert healthz.status_code == 200
+    assert healthz.json() == {"status": "ok"}
+    assert readyz.status_code == 200, readyz.text
+    assert readyz.json() == {"postgres": True, "minio": True, "qdrant": True}
+
+
+async def test_upload_through_the_running_process_indexes_retrievable_points(
+    client: AsyncClient,
+) -> None:
+    """The whole slice against the deployed shape: HTTP in, Qdrant points out."""
+    upload = await client.post(
+        "/v1/files",
+        files={"file": ("smoke.csv", _CSV, "text/csv")},
+        data={"domain": DOMAIN},
+        headers={"Authorization": f"Bearer {API_KEY}"},
+    )
+    assert upload.status_code == 201, upload.text
+    body = upload.json()
+    assert body["status"] == "succeeded"
+    assert body["chunks_indexed"] > 0
+
+    status = await client.get(
+        f"/v1/files/{body['file_id']}", headers={"Authorization": f"Bearer {API_KEY}"}
+    )
+    assert status.status_code == 200
+    assert status.json()["ingestion_status"] == "succeeded"
+
+    qdrant = AsyncQdrantClient(url=QDRANT_URL)
+    try:
+        count = await qdrant.count(
+            collection_name=QDRANT_COLLECTION,
+            count_filter=models.Filter(
+                must=[
+                    models.FieldCondition(
+                        key="file_id", match=models.MatchValue(value=body["file_id"])
+                    ),
+                    models.FieldCondition(key="is_active", match=models.MatchValue(value=True)),
+                ]
+            ),
+            exact=True,
+        )
+    finally:
+        await qdrant.close()
+
+    assert count.count == body["chunks_indexed"]
+
+
+async def test_running_process_writes_structured_ingestion_logs() -> None:
+    """The real ADR-0011 sink, which only a separate process exercises.
+
+    An upload that succeeds while emitting nothing parseable is an upload no
+    operator can investigate -- and the runbook's failure procedure is written
+    against these exact event names.
+    """
+    if not LOG_PATH:
+        pytest.skip("SMOKE_LOG_PATH is unset; scripts/smoke.sh normally provides it")
+
+    lines = Path(LOG_PATH).read_text("utf-8").splitlines()
+    events = [json.loads(line) for line in lines if line.startswith("{")]
+
+    by_name = {event.get("event") for event in events}
+    assert "ingestion.job.started" in by_name
+    assert "ingestion.job.completed" in by_name
+
+    completed = next(event for event in events if event.get("event") == "ingestion.job.completed")
+    assert completed["tenant_id"]
+    assert completed["ingestion_job_id"]
+    assert completed["points_upserted"] > 0
+    # Static process context bound once at startup (ADR-0011).
+    assert completed["env"]
+    assert completed["service_version"]

tests/e2e/test_ingestion_slice.py

@@ -0,0 +1,329 @@
+"""The ingestion vertical slice end to end (plan 001, Phase 6).
+
+Every test here asserts a reliability invariant ADR-0016 lists as a reusable
+contract, not a happy path: the bound maps to its status code, the failure
+still writes a terminal job row, the retry converges, and the tenant filter
+holds. Failures additionally assert the ADR-0008 error envelope, because the
+status code alone is not the contract a client integrates against.
+"""
+
+import asyncio
+import uuid
+
+import pytest
+from httpx import Response
+from qdrant_client import AsyncQdrantClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from src.infrastructure.postgres.models.ingestion_job import IngestionJob
+from tests.e2e.conftest import (
+    CORRUPT_DOCX_BYTES,
+    Principal,
+    StackFactory,
+    count_active_points,
+    provision_principal,
+    upload_payload,
+)
+
+pytestmark = [
+    pytest.mark.e2e,
+    pytest.mark.postgres,
+    pytest.mark.minio,
+    pytest.mark.qdrant,
+    pytest.mark.asyncio(loop_scope="session"),
+]
+
+
+async def _latest_job(session: AsyncSession, *, tenant_id: uuid.UUID) -> IngestionJob:
+    """The tenant's most recent job row, read fresh.
+
+    `populate_existing` matters: this session shares a connection with the
+    app's sessions, so without it the identity map can answer from a row
+    loaded before the request under test committed its update.
+    """
+    result = await session.execute(
+        select(IngestionJob)
+        .where(IngestionJob.tenant_id == tenant_id)
+        .order_by(IngestionJob.created_at.desc(), IngestionJob.id)
+        .execution_options(populate_existing=True)
+    )
+    return result.scalars().first() or pytest.fail("no ingestion job was written")
+
+
+def _envelope(response: Response) -> dict[str, object]:
+    """The ADR-0008 error body. Asserting the code here, not just the HTTP
+    status, is the difference between testing the contract and testing the
+    number."""
+    error: dict[str, object] = response.json()["error"]
+    assert error["request_id"], "every error envelope carries its correlation id"
+    return error
+
+
+async def test_upload_succeeds_and_indexes_points_under_the_tenant_filter(
+    make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 201
+    body = response.json()
+    assert body["status"] == "succeeded"
+    assert body["chunks_indexed"] > 0
+    indexed = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+    assert indexed == body["chunks_indexed"]
+
+
+async def test_get_file_reports_the_terminal_status_after_upload(
+    make_stack: StackFactory, principal: Principal
+) -> None:
+    stack = await make_stack()
+    created = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    status = await stack.client.get(
+        f"/v1/files/{created.json()['file_id']}", headers=principal.headers
+    )
+
+    assert status.status_code == 200
+    assert status.json()["ingestion_status"] == "succeeded"
+    assert status.json()["domain"] == principal.domain
+
+
+async def test_upload_duplicate_content_returns_the_existing_job_without_reingesting(
+    make_stack: StackFactory, principal: Principal, qdrant_client: AsyncQdrantClient
+) -> None:
+    stack = await make_stack()
+    first = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    points_after_first = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+
+    second = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert first.status_code == 201
+    # 200, not 201: nothing was created the second time (ADR-0017 idempotency).
+    assert second.status_code == 200
+    assert second.json()["file_id"] == first.json()["file_id"]
+    assert second.json()["ingestion_job_id"] == first.json()["ingestion_job_id"]
+    assert (
+        await count_active_points(qdrant_client, stack.collection, tenant_id=principal.tenant_id)
+        == points_after_first
+    )
+
+
+async def test_upload_retried_after_a_failed_job_succeeds_without_duplicate_points(
+    make_stack: StackFactory,
+    principal: Principal,
+    db_session: AsyncSession,
+    qdrant_client: AsyncQdrantClient,
+) -> None:
+    """ADR-0017, "Re-running an ingestion stays safe": a failed attempt is
+    retried by re-uploading the same bytes, and deterministic point ids make
+    the retry converge instead of duplicating.
+    """
+    stack = await make_stack()
+    stack.dense_embedders[0].fail_next = True
+
+    failed = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    failed_job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    retried = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert failed.status_code == 502
+    assert failed_job.status == "failed"
+    assert failed_job.error_code == "embedding_failed"
+    assert retried.status_code == 201
+    assert retried.json()["status"] == "succeeded"
+    # Same source file, new attempt -- a terminal job is never reused or
+    # transitioned back to running.
+    assert retried.json()["ingestion_job_id"] != str(failed_job.id)
+    indexed = await count_active_points(
+        qdrant_client, stack.collection, tenant_id=principal.tenant_id
+    )
+    assert indexed == retried.json()["chunks_indexed"]
+
+
+async def test_get_file_for_another_tenants_file_returns_404_not_403(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """404, never 403: a 403 would confirm the file exists to a stranger."""
+    stack = await make_stack()
+    created = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+    other = await provision_principal(db_session)
+
+    response = await stack.client.get(
+        f"/v1/files/{created.json()['file_id']}", headers=other.headers
+    )
+
+    assert response.status_code == 404
+    assert _envelope(response)["code"] == "not_found"
+
+
+async def test_upload_to_an_unregistered_domain_is_rejected_before_any_job_is_written(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain="never-registered"), headers=principal.headers
+    )
+
+    assert response.status_code == 400
+    assert _envelope(response)["code"] == "unknown_domain"
+    jobs = await db_session.execute(
+        select(IngestionJob).where(IngestionJob.tenant_id == principal.tenant_id)
+    )
+    assert jobs.scalars().all() == []
+
+
+async def test_upload_at_capacity_is_rejected_with_503_and_retry_after(
+    make_stack: StackFactory, committed_sessionmaker: async_sessionmaker[AsyncSession]
+) -> None:
+    """ADR-0017 rejects rather than queues, so the second concurrent upload
+    must fail fast with a backoff hint instead of waiting for a slot.
+
+    This is the one test that needs two requests genuinely in flight at once,
+    so it runs on `committed_sessionmaker` -- see that fixture for why the
+    shared-connection default cannot serve concurrent sessions.
+    """
+    async with committed_sessionmaker() as session:
+        principal = await provision_principal(session)
+    stack = await make_stack(
+        ingestion={"max_concurrency": 1},
+        dense_delay_seconds=0.4,
+        sessionmaker=committed_sessionmaker,
+    )
+
+    first, second = await asyncio.gather(
+        stack.client.post(
+            "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+        ),
+        stack.client.post(
+            "/v1/files",
+            **upload_payload(domain=principal.domain, filename="other.csv", data=b"a,b\n1,2\n"),
+            headers=principal.headers,
+        ),
+    )
+
+    statuses = sorted([first.status_code, second.status_code])
+    assert statuses == [201, 503]
+    rejected = first if first.status_code == 503 else second
+    assert _envelope(rejected)["code"] == "ingestion_at_capacity"
+    assert rejected.headers["Retry-After"] == "1"
+
+
+async def test_upload_exceeding_the_timeout_returns_504_and_writes_a_terminal_job(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """A timeout must never leave a job stuck in `running` (ADR-0016)."""
+    stack = await make_stack(ingestion={"timeout_seconds": 0.05}, dense_delay_seconds=0.5)
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 504
+    assert _envelope(response)["code"] == "ingestion_timeout"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "timeout"
+
+
+async def test_upload_of_an_unparseable_document_returns_400_and_a_failed_job(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+
+    response = await stack.client.post(
+        "/v1/files",
+        **upload_payload(domain=principal.domain, filename="broken.docx", data=CORRUPT_DOCX_BYTES),
+        headers=principal.headers,
+    )
+
+    assert response.status_code == 400
+    assert _envelope(response)["code"] == "validation_error"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "parse_failed"
+
+
+async def test_upload_returns_502_when_qdrant_indexing_fails(
+    make_stack: StackFactory, principal: Principal, db_session: AsyncSession
+) -> None:
+    """A real Qdrant error, not a fake: the collection the deployment step
+    should have created is missing, which is the failure an operator actually
+    hits when `qdrant_bootstrap` was skipped.
+    """
+    stack = await make_stack(
+        collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
+    )
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=principal.domain), headers=principal.headers
+    )
+
+    assert response.status_code == 502
+    assert _envelope(response)["code"] == "index_error"
+    job = await _latest_job(db_session, tenant_id=principal.tenant_id)
+    assert job.status == "failed"
+    assert job.error_code == "index_failed"
+
+
+async def test_upload_without_the_files_write_scope_is_forbidden(
+    make_stack: StackFactory, db_session: AsyncSession
+) -> None:
+    stack = await make_stack()
+    reader = await provision_principal(db_session, scopes=["domains:read"])
+
+    response = await stack.client.post(
+        "/v1/files", **upload_payload(domain=reader.domain), headers=reader.headers
+    )
+
+    assert response.status_code == 403
+    assert _envelope(response)["code"] == "missing_scope"
+
+
+async def test_readyz_reports_every_dependency_ready_against_real_services(
+    make_stack: StackFactory,
+) -> None:
+    """`/readyz` is dependency readiness, `/healthz` is process health."""
+    stack = await make_stack()
+
+    ready = await stack.client.get("/readyz")
+    healthy = await stack.client.get("/healthz")
+
+    assert healthy.status_code == 200
+    assert ready.status_code == 200
+    assert ready.json() == {"postgres": True, "minio": True, "qdrant": True}
+
+
+async def test_readyz_is_unready_when_the_chunks_collection_is_missing(
+    make_stack: StackFactory,
+) -> None:
+    """A reachable but unbootstrapped Qdrant is deliberately not ready --
+    uploads to it would 502 (see the indexing test above).
+    """
+    stack = await make_stack(
+        collection=f"never_bootstrapped_{uuid.uuid4().hex}", bootstrap_collection=False
+    )
+
+    response = await stack.client.get("/readyz")
+
+    assert response.status_code == 503
+    assert response.json()["qdrant"] is False

tests/fakes.py

@@ -32,6 +32,9 @@ class FakeDenseEmbedder:
 
     name: str
     dimensions: int = 4
+    value: float = 0.0
+    """Component value of every returned vector. Non-zero where a real Qdrant
+    has to score the result, since a zero vector has no direction to compare."""
     model_version: str = "fake-dense-v1"
     calls: list[list[str]] = field(default_factory=list)
     fail_next: bool = False
@@ -45,7 +48,7 @@ class FakeDenseEmbedder:
         if self.fail_next:
             self.fail_next = False
             raise RuntimeError("simulated embedder failure")
-        return [[0.0] * self.dimensions for _ in texts]
+        return [[self.value] * self.dimensions for _ in texts]
 
 
 @dataclass

tests/integration/minio/conftest.py

@@ -1,39 +0,0 @@
-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/postgres/conftest.py

@@ -1,98 +0,0 @@
-from collections.abc import AsyncIterator, Iterator
-
-import pytest
-import pytest_asyncio
-from alembic.command import upgrade
-from alembic.config import Config
-from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
-from testcontainers.community.postgres import PostgresContainer
-
-from src.config import PostgresSettings
-from src.infrastructure.postgres.database import create_engine
-
-
-def _alembic_config(database_url: str) -> Config:
-    config = Config("alembic.ini")
-    config.set_main_option("sqlalchemy.url", database_url)
-    return config
-
-
-def _settings_from_url(url: str) -> PostgresSettings:
-    # testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
-    # PostgresSettings builds its own dsn from parts, so parse the parts back out.
-    without_scheme = url.split("://", 1)[1]
-    creds, hostpart = without_scheme.split("@", 1)
-    user, password = creds.split(":", 1)
-    hostport, db = hostpart.split("/", 1)
-    host, port = hostport.split(":", 1)
-    return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
-
-
-@pytest.fixture(scope="session")
-def postgres_container() -> Iterator[PostgresContainer]:
-    with PostgresContainer("postgres:17", driver="asyncpg") as container:
-        yield container
-
-
-@pytest.fixture(scope="session")
-def postgres_url(postgres_container: PostgresContainer) -> str:
-    """The container's URL, pinned to IPv4.
-
-    Testcontainers reports the host as `localhost`, which resolves to `::1`
-    before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
-    IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
-    address until its connect timeout instead of falling back to the second --
-    the connection does not fail, it hangs.
-    """
-    return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
-
-
-@pytest.fixture(scope="session")
-def migrated_postgres_url(postgres_url: str) -> str:
-    """The container's URL, after Alembic has created the schema on it once."""
-    upgrade(_alembic_config(postgres_url), "head")
-    return postgres_url
-
-
-@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:
-        yield engine
-    finally:
-        await engine.dispose()
-
-
-@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.
-
-    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"
-        )
-        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_provisioning.py

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

tests/integration/qdrant/conftest.py

@@ -1,52 +0,0 @@
-import uuid
-from collections.abc import AsyncIterator, Iterator
-
-import pytest
-import pytest_asyncio
-from qdrant_client import AsyncQdrantClient
-from testcontainers.community.qdrant import QdrantContainer
-
-from src.config import QdrantSettings
-from src.infrastructure.qdrant.client import create_client
-
-# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
-# testcontainers default image trails it far enough that the client emits an
-# incompatibility warning, and testing against a version we do not deploy is
-# the wrong signal anyway.
-_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
-
-
-@pytest.fixture(scope="session")
-def qdrant_container() -> Iterator[QdrantContainer]:
-    with QdrantContainer(image=_QDRANT_IMAGE) as container:
-        yield container
-
-
-@pytest.fixture(scope="session")
-def qdrant_url(qdrant_container: QdrantContainer) -> str:
-    """The container's REST URL, pinned to IPv4.
-
-    Same gotcha as postgres_url/minio_settings (see their conftests):
-    testcontainers reports the host as `localhost`, which resolves to `::1`
-    first, but Docker publishes the mapped port on IPv4 only. The IPv6 SYN is
-    dropped rather than refused, so the client hangs until its timeout instead
-    of falling back to the second address -- the connection does not fail, it
-    hangs.
-    """
-    host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
-    return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
-
-
-@pytest.fixture
-def qdrant_settings(qdrant_url: str) -> QdrantSettings:
-    """Settings naming a collection unique to this test (ADR-0016 isolation)."""
-    return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
-
-
-@pytest_asyncio.fixture(loop_scope="session")
-async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
-    client = create_client(qdrant_settings)
-    try:
-        yield client
-    finally:
-        await client.close()

tests/support/containers.py

@@ -0,0 +1,205 @@
+"""Disposable real infrastructure for integration and e2e tests (ADR-0016).
+
+Registered as a plugin from the root `tests/conftest.py` rather than living in
+a per-boundary conftest, because `tests/e2e/` needs the same Postgres, MinIO,
+and Qdrant containers that `tests/integration/*/` does, and `pytest_plugins`
+is only honoured in the root conftest.
+
+Nothing here is autouse: the container fixtures are session-scoped but lazy,
+so `pytest -m unit` still runs without Docker.
+"""
+
+import uuid
+from collections.abc import AsyncIterator, Iterator
+
+import pytest
+import pytest_asyncio
+from alembic.command import upgrade
+from alembic.config import Config
+from qdrant_client import AsyncQdrantClient
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
+from testcontainers.community.minio import MinioContainer
+from testcontainers.community.postgres import PostgresContainer
+from testcontainers.community.qdrant import QdrantContainer
+
+from src.config import MinioSettings, PostgresSettings, QdrantSettings
+from src.infrastructure.minio.client import create_client as create_minio_client
+from src.infrastructure.postgres.database import create_engine
+from src.infrastructure.qdrant.client import create_client as create_qdrant_client
+
+_MINIO_BUCKET = "test-source-files"
+
+# Pinned to match the `qdrant-client` major/minor in pyproject.toml. The
+# testcontainers default image trails it far enough that the client emits an
+# incompatibility warning, and testing against a version we do not deploy is
+# the wrong signal anyway.
+_QDRANT_IMAGE = "qdrant/qdrant:v1.19.0"
+
+
+# --- Postgres ---------------------------------------------------------------
+
+
+def _alembic_config(database_url: str) -> Config:
+    config = Config("alembic.ini")
+    config.set_main_option("sqlalchemy.url", database_url)
+    return config
+
+
+def _postgres_settings_from_url(url: str) -> PostgresSettings:
+    # testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
+    # PostgresSettings builds its own dsn from parts, so parse the parts back out.
+    without_scheme = url.split("://", 1)[1]
+    creds, hostpart = without_scheme.split("@", 1)
+    user, password = creds.split(":", 1)
+    hostport, db = hostpart.split("/", 1)
+    host, port = hostport.split(":", 1)
+    return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
+
+
+@pytest.fixture(scope="session")
+def postgres_container() -> Iterator[PostgresContainer]:
+    with PostgresContainer("postgres:17", driver="asyncpg") as container:
+        yield container
+
+
+@pytest.fixture(scope="session")
+def postgres_url(postgres_container: PostgresContainer) -> str:
+    """The container's URL, pinned to IPv4.
+
+    Testcontainers reports the host as `localhost`, which resolves to `::1`
+    before `127.0.0.1`. Docker publishes the mapped port on IPv4 only, and the
+    IPv6 SYN is dropped rather than refused, so asyncpg blocks on the first
+    address until its connect timeout instead of falling back to the second --
+    the connection does not fail, it hangs.
+    """
+    return postgres_container.get_connection_url().replace("@localhost:", "@127.0.0.1:")
+
+
+@pytest.fixture(scope="session")
+def migrated_postgres_url(postgres_url: str) -> str:
+    """The container's URL, after Alembic has created the schema on it once."""
+    upgrade(_alembic_config(postgres_url), "head")
+    return postgres_url
+
+
+@pytest.fixture(scope="session")
+def postgres_settings(migrated_postgres_url: str) -> PostgresSettings:
+    """Settings an application component can be constructed from directly."""
+    return _postgres_settings_from_url(migrated_postgres_url)
+
+
+@pytest_asyncio.fixture(scope="session", loop_scope="session")
+async def postgres_engine(postgres_settings: PostgresSettings) -> AsyncIterator[AsyncEngine]:
+    engine = create_engine(postgres_settings)
+    try:
+        yield engine
+    finally:
+        await engine.dispose()
+
+
+@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.
+
+    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"
+        )
+        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
+
+
+# --- MinIO ------------------------------------------------------------------
+
+
+@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:
+    """Settings for the container, with the bucket already created.
+
+    Bucket creation belongs here rather than in a separate autouse fixture:
+    autouse in a *globally registered* plugin would pull a MinIO container
+    into every unit test run.
+    """
+    config = minio_container.get_config()
+    # Pinned to IPv4 for the same reason as postgres_url above: `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:")
+    settings = MinioSettings(
+        endpoint=endpoint,
+        access_key=config["access_key"],
+        secret_key=config["secret_key"],
+        secure=False,
+        bucket=_MINIO_BUCKET,
+    )
+    client = create_minio_client(settings)
+    if not client.bucket_exists(settings.bucket):
+        client.make_bucket(settings.bucket)
+    return settings
+
+
+# --- Qdrant -----------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def qdrant_container() -> Iterator[QdrantContainer]:
+    with QdrantContainer(image=_QDRANT_IMAGE) as container:
+        yield container
+
+
+@pytest.fixture(scope="session")
+def qdrant_url(qdrant_container: QdrantContainer) -> str:
+    """The container's REST URL, pinned to IPv4.
+
+    Same gotcha as postgres_url/minio_settings above: testcontainers reports
+    the host as `localhost`, which resolves to `::1` first, but Docker
+    publishes the mapped port on IPv4 only. The IPv6 SYN is dropped rather
+    than refused, so the client hangs until its timeout instead of falling
+    back to the second address -- the connection does not fail, it hangs.
+    """
+    host = qdrant_container.get_container_host_ip().replace("localhost", "127.0.0.1")
+    return f"http://{host}:{qdrant_container.get_exposed_port(6333)}"
+
+
+@pytest.fixture
+def qdrant_settings(qdrant_url: str) -> QdrantSettings:
+    """Settings naming a collection unique to this test (ADR-0016 isolation)."""
+    return QdrantSettings(url=qdrant_url, collection=f"chunks_{uuid.uuid4().hex}")
+
+
+@pytest_asyncio.fixture(loop_scope="session")
+async def qdrant_client(qdrant_settings: QdrantSettings) -> AsyncIterator[AsyncQdrantClient]:
+    client = create_qdrant_client(qdrant_settings)
+    try:
+        yield client
+    finally:
+        await client.close()

tests/unit/cli/test_provision_tenant.py

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