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>
test(e2e): add the Compose smoke test of the running web process
2026-08-20 22:26:58 +03:30 · 2026-08-20 22:26:47 +03:30 · 2026-08-20 22:26:36 +03:30 · 2026-08-20 22:26:25 +03:30 · 2026-08-20 22:26:12 +03:30 · 2026-08-20 22:26:02 +03:30
23 changed files with 1766 additions and 203 deletions
--- a/CLAUDE.md
+++ b/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
+Also working: the operator runbook (`docs/runbook.md`), tenant/API-key/domain
-`src/agent/`. That maps to plan 001 Phases 1-5 done, Phase 6 (runbook, e2e
+provisioning (`uv run python -m src.cli.provision_tenant` — the third deployment
-tests, Compose smoke test) not started.
+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
+- Integration **and e2e** tests use **Testcontainers** (never a developer's
-  services or Langfuse-owned storage/credentials) — this is the standard
+  local services or Langfuse-owned storage/credentials) — this is the standard
-  automated mechanism, not Docker Compose. Isolate data per test via unique
+  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
--- a/README.md
+++ b/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:
--- a/docs/plans/001-ingestion-vertical-slice.md
+++ b/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.
--- a/docs/runbook.md
+++ b/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.
--- a/scripts/smoke.sh
+++ b/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"
--- a/src/api/errors.py
+++ b/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",
--- a/src/application/tenants/init.py
+++ b/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"]
--- a/src/application/tenants/provisioning.py
+++ b/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),
    )
--- a/src/cli/provision_tenant.py
+++ b/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()
--- a/src/infrastructure/postgres/repositories/api_keys.py
+++ b/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
--- a/src/infrastructure/postgres/repositories/tenants.py
+++ b/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
--- a/tests/conftest.py
+++ b/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]:
--- a/tests/e2e/conftest.py
+++ b/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
--- a/tests/e2e/test_compose_smoke.py
+++ b/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"]
--- a/tests/e2e/test_ingestion_slice.py
+++ b/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
--- a/tests/fakes.py
+++ b/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
--- a/tests/integration/minio/conftest.py
+++ b/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)
--- a/tests/integration/postgres/conftest.py
+++ b/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
--- a/tests/integration/postgres/test_provisioning.py
+++ b/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")
--- a/tests/integration/qdrant/conftest.py
+++ b/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()
--- a/tests/support/containers.py
+++ b/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()
--- a/tests/unit/cli/init.py
+++ b/tests/unit/cli/init.py
--- a/tests/unit/cli/test_provision_tenant.py
+++ b/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"])
Author	SHA1	Message	Date
Ali Zarinkolah	3d9269e54f	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>	2026-08-20 22:26:58 +03:30
Ali Zarinkolah	7c1fe79f1c	test(e2e): add the Compose smoke test of the running web process Why: - ADR-0016 reserves Compose for a serialized smoke test of the running web process. Nothing else exercises the deployment steps, a real HTTP server, or the real logging configuration, which every in-process test no-ops. Changes: - scripts/smoke.sh brings up Compose, runs both bootstrap steps, provisions a throwaway tenant, starts uvicorn, and drives the test against it - the test skips unless SMOKE_BASE_URL is set, so `uv run pytest` never invokes Compose; it asserts the ADR-0011 JSON log sink Impact: - a pre-release gate, not a per-PR one Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-20 22:26:47 +03:30
Ali Zarinkolah	1b873e5a6f	test(e2e): cover the ingestion slice against real Postgres, MinIO, and Qdrant Why: - the slice's reliability invariants (ADR-0016) had no end-to-end coverage. Changes: - 13 Testcontainers-based tests: duplicate upload, retry after a failed job, cross-tenant 404, unregistered domain, capacity 503, timeout 504, parse 400, real Qdrant 502, missing scope 403, and both readiness states - only the dense embedders are faked (ADR-0016 bars live providers); they return the pinned 768/3072 dimensions - the capacity test uses a committing sessionmaker, since the shared-connection fixture cannot serve concurrent sessions Impact: - runs in the default `uv run pytest`; needs Docker, like every integration test Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-20 22:26:36 +03:30
Ali Zarinkolah	133f565704	feat(tenant): add operator provisioning for tenants, API keys, and domains Why: - nothing over HTTP could create the first tenant: every /v1 route needs an API key, and a key cannot exist before its tenant. The service was unusable by a human without hand-written SQL. Changes: - add `provision_tenant`, owning tenant reuse-or-create, key generation and hashing, and domain registration in one transaction - expose it as `python -m src.cli.provision_tenant`, alongside `alembic upgrade head` and `qdrant_bootstrap` - add `tenants.get_by_slug`/`create` and `api_keys.create` - log `tenant.provisioned` / `api_key.provisioned` with the key prefix only Impact: - a third deployment step; the plaintext key is printed once and never logged or stored (ADR-0011, ADR-0009) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-20 22:26:25 +03:30
Ali Zarinkolah	c7a5b69c0a	fix(api): use the non-deprecated 422 status constant Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-20 22:26:12 +03:30
Ali Zarinkolah	2c72688440	refactor(test): share container fixtures across the integration and e2e suites Why: - tests/e2e/ cannot reach fixtures defined in a per-boundary conftest, and `pytest_plugins` is only honoured in the root conftest. Changes: - move the Postgres/MinIO/Qdrant container fixtures into tests/support/containers.py and register it as a root plugin - fold MinIO bucket creation into `minio_settings`; an autouse fixture in a globally registered plugin would pull a container into unit runs - add a `postgres_settings` fixture so a component can be built from it directly Impact: - no behavior change; `pytest -m unit` still needs no Docker Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-08-20 22:26:02 +03:30