# Phase 6: Performance & Production Hardening — Pattern Map

**Mapped:** 2026-06-02
**Files analyzed:** 12
**Analogs found:** 9 / 12

---

## File Classification

| New/Modified File | Role | Data Flow | Closest Analog | Match Quality |
|-------------------|------|-----------|----------------|---------------|
| `backend/deps/utils.py` | utility | request-response | `backend/deps/utils.py` (self) | self-update |
| `backend/main.py` | config/wiring | request-response | `backend/main.py` (self) | self-update |
| `backend/api/auth.py` | controller | request-response | `backend/api/auth.py` (self) | self-update |
| `backend/api/documents.py` | controller | CRUD | `backend/api/auth.py` | role-match |
| `backend/api/cloud.py` | controller | request-response | `backend/api/auth.py` | role-match |
| `backend/services/logging.py` | service | event-driven | `backend/services/auth.py` | role-match |
| `backend/config.py` | config | — | `backend/config.py` (self) | self-update |
| `backend/load_tests/locustfile.py` | test | request-response | `backend/tests/conftest.py` | partial-match |
| `backend/Dockerfile` | config | — | `backend/Dockerfile` (self) | self-update |
| `docker-compose.yml` | config | — | `docker-compose.yml` (self) | self-update |
| `docker/loki/loki-config.yaml` | config | — | none | no-analog |
| `docker/loki/promtail-config.yaml` | config | — | none | no-analog |
| `RUNBOOK.md` | documentation | — | none | no-analog |

---

## Pattern Assignments

### `backend/deps/utils.py` (utility, request-response) — D-11

**Change type:** Replace function body in-place. The function `get_client_ip` already exists and is imported by every router that does audit logging (`auth.py`, `documents.py`, etc.). The body must be replaced with trusted-proxy CIDR logic. Do NOT rename or add a second function.

**Current body** (`backend/deps/utils.py` lines 10–22):
```python
def get_client_ip(request: Request) -> Optional[str]:
    """Extract best-effort client IP from request for audit logging.

    TRUST BOUNDARY: X-Forwarded-For is a client-controlled header and can be
    forged by any caller. ...
    """
    return request.headers.get("X-Forwarded-For") or (
        request.client.host if request.client else None
    )
```

**Replacement pattern** (from RESEARCH.md Pattern 3):
```python
import ipaddress
from typing import Optional
from fastapi import Request

_TRUSTED_PROXY_NETS = [
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("::1/128"),
]

def _is_trusted_proxy(host: str) -> bool:
    try:
        addr = ipaddress.ip_address(host)
        return any(addr in net for net in _TRUSTED_PROXY_NETS)
    except ValueError:
        return False

def get_client_ip(request: Request) -> Optional[str]:
    """Extract client IP with trusted-proxy CIDR check (D-11).

    If the direct peer (request.client.host) is a trusted proxy, read the
    leftmost address from X-Forwarded-For. Otherwise ignore forwarded headers
    and return the direct peer IP — prevents header spoofing from external clients.
    """
    direct_peer = request.client.host if request.client else None
    if direct_peer and _is_trusted_proxy(direct_peer):
        xff = request.headers.get("X-Forwarded-For")
        if xff:
            return xff.split(",")[0].strip()
    return direct_peer
```

**Existing import callers** (no changes required in these files — they already import the right name):
- `backend/api/auth.py` line 34: `from deps.utils import get_client_ip`
- (all other routers that call `get_client_ip(request)`)

**TRUSTED_PROXY_CIDRS config hook:** The list `_TRUSTED_PROXY_NETS` should be built from `settings.trusted_proxy_cidrs` (added in `config.py`) rather than hardcoded. The hardcoded list above is the safe default; read from config on module import after `settings` is available.

---

### `backend/main.py` (config/wiring, request-response) — D-01, D-12

**Change type:** Add `CorrelationIDMiddleware` class, import `setup_logging`, wire `account_limiter` state.

**Existing middleware pattern** (`backend/main.py` lines 24–131) — copy exactly for the new raw-ASGI middleware class:

```python
# Existing pattern for BaseHTTPMiddleware (lines 25–42):
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        response.headers["Content-Security-Policy"] = "..."
        return response

# Existing app.add_middleware() calls (lines 108–131):
app.state.limiter = auth_limiter          # line 109
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)  # line 110
app.add_middleware(SlowAPIMiddleware)      # line 111
app.add_middleware(SecurityHeadersMiddleware)  # line 119
app.add_middleware(CORSMiddleware, ...)    # line 122-128
app.add_middleware(OriginValidationMiddleware) # line 131
```

**New CorrelationIDMiddleware** — use raw ASGI (NOT BaseHTTPMiddleware) to avoid response buffering. Must be registered LAST so it runs FIRST in the request chain (Starlette reverse-insertion order):

```python
# Add import block additions to main.py:
import uuid
import time
import structlog
from starlette.types import ASGIApp, Receive, Scope, Send

logger = structlog.get_logger()

class CorrelationIDMiddleware:
    """Generate per-request correlation ID; bind to structlog contextvars.

    Uses raw ASGI (not BaseHTTPMiddleware) to avoid response-body buffering.
    Register LAST so it runs FIRST (Starlette reverse-insertion order).
    """
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        correlation_id = str(uuid.uuid4())
        start_ns = time.perf_counter_ns()

        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            correlation_id=correlation_id,
            path=scope.get("path", ""),
            method=scope.get("method", ""),
        )

        async def send_with_header(message):
            if message["type"] == "http.response.start":
                headers = list(message.get("headers", []))
                headers.append(
                    (b"x-correlation-id", correlation_id.encode())
                )
                message = {**message, "headers": headers}
            await send(message)

        await self.app(scope, receive, send_with_header)
        duration_ms = (time.perf_counter_ns() - start_ns) / 1_000_000
        structlog.contextvars.bind_contextvars(duration_ms=round(duration_ms, 2))
```

**Lifespan hook — call setup_logging first** (`backend/main.py` lines 67–101 — insert before the `yield`):
```python
# In lifespan(), before the existing minio init:
from services.logging import setup_logging
setup_logging(
    json_logs=settings.log_json,
    log_level=settings.log_level,
)
```

**account_limiter wiring** — add alongside existing `app.state.limiter = auth_limiter`:
```python
# In main.py, alongside app.state.limiter = auth_limiter (line 109):
from services.rate_limiting import account_limiter  # or wherever it lives
# account_limiter decorators work independently; no app.state wiring needed
# SlowAPIMiddleware only tracks the limiter assigned to app.state
app.state.limiter = auth_limiter  # existing — drives SlowAPIMiddleware
```

**Middleware registration order** — CorrelationIDMiddleware added last (runs first), per existing Starlette convention documented at line 113–116:
```python
# After all existing app.add_middleware() calls, add last:
app.add_middleware(CorrelationIDMiddleware)
```

---

### `backend/api/auth.py` (controller, request-response) — D-11, D-13

**Change type:** Replace `key_func=get_remote_address` with `key_func=get_client_ip`. Two-line change.

**Current limiter declaration** (`backend/api/auth.py` lines 37–44):
```python
from slowapi import Limiter
from slowapi.util import get_remote_address

router = APIRouter(prefix="/api/auth", tags=["auth"])

# IP-level rate limiter (SEC-02 — 10 req/min on register/login/refresh)
limiter = Limiter(key_func=get_remote_address)
```

**Replacement**:
```python
from slowapi import Limiter
from deps.utils import get_client_ip  # replace get_remote_address import

router = APIRouter(prefix="/api/auth", tags=["auth"])

# IP-level rate limiter with trusted-proxy key function (D-11, D-13)
limiter = Limiter(key_func=get_client_ip)
```

**Existing `@limiter.limit()` decorators remain unchanged** — they are already on the right endpoints (`lines 97–98`, `170–171`, `300–301`, `538–539`, `621–622`). Per D-13 the limits themselves (10/minute, 5/hour) are preserved.

---

### `backend/api/documents.py` (controller, CRUD) — D-12

**Change type:** Add `@account_limiter.limit("100/minute")` decorator and `request.state.current_user = current_user` binding to authenticated endpoints.

**Existing endpoint pattern** (`backend/api/documents.py` lines 88–101) — the handler signature and dependency injection to copy from:
```python
@router.post("/upload-url")
async def request_upload_url(
    body: UploadUrlRequest,
    session: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_regular_user),
):
```

**New pattern with per-account rate limiting**:
```python
from backend.services.rate_limiting import account_limiter  # shared module

@router.get("/")
@account_limiter.limit("100/minute")
async def list_documents(
    request: Request,                          # Request must be first positional arg for slowapi
    current_user: User = Depends(get_regular_user),
    session: AsyncSession = Depends(get_db),
    ...
):
    request.state.current_user = current_user  # MUST be first line — exposes user to key_func
    structlog.contextvars.bind_contextvars(user_id=str(current_user.id))
    ...
```

**Key constraint:** `Request` must appear as the first parameter after `self` for slowapi decorators to work. Review each existing endpoint signature — `request: Request` may need to be added or moved to first position.

---

### `backend/api/cloud.py` (controller, request-response) — D-12

**Change type:** Same per-account rate limiting pattern as `documents.py`. The cloud router uses the same `Depends(get_regular_user)` pattern visible at `backend/api/cloud.py` lines 29–47.

**Existing endpoint signature pattern** (`backend/api/cloud.py` lines 44–47):
```python
from deps.auth import get_regular_user
from deps.db import get_db

router = APIRouter(prefix="/api/cloud", tags=["cloud"])
```

**Apply the same decorator/binding pattern** as `documents.py` above to each endpoint that uses `Depends(get_regular_user)`.

---

### `backend/services/logging.py` (service, event-driven) — D-01

**Change type:** New file. No existing analog for a structlog setup module. The closest structural analog is `backend/services/auth.py` (pure Python service, no FastAPI coupling, single module with module-level init).

**Analog structure** (`backend/services/auth.py` lines 1–45):
```python
"""
Auth service — pure Python, no FastAPI coupling.
...
"""
from __future__ import annotations
import logging
# ... imports ...
from config import settings

logger = logging.getLogger(__name__)

# Module-level init (PasswordHash instance)
_pwd = PasswordHash([Argon2Hasher()])

def hash_password(plain: str) -> str: ...
def verify_password(plain: str, hashed: str) -> bool: ...
```

**New file pattern** — mirror the module-level docstring, `from __future__ import annotations`, import from `config.settings`, expose a single entry-point function:
```python
"""
Structured logging setup — pure Python, no FastAPI coupling.

Call setup_logging() once in main.py lifespan before the yield.
Bridges stdlib loggers (uvicorn, sqlalchemy, celery) through the same
structlog JSON processor chain.
"""
from __future__ import annotations

import logging
import structlog

from config import settings


def setup_logging(json_logs: bool = False, log_level: str = "INFO") -> None:
    """Configure structlog with ProcessorFormatter bridge for stdlib loggers.

    Parameters match settings.log_json and settings.log_level so callers
    can pass settings values directly.
    """
    timestamper = structlog.processors.TimeStamper(fmt="iso")

    shared_processors = [
        structlog.contextvars.merge_contextvars,      # MUST be first
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.stdlib.ExtraAdder(),
        timestamper,
        structlog.processors.StackInfoRenderer(),
    ]
    if json_logs:
        shared_processors.append(structlog.processors.format_exc_info)

    structlog.configure(
        processors=shared_processors + [
            structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
        ],
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

    log_renderer = (
        structlog.processors.JSONRenderer()
        if json_logs
        else structlog.dev.ConsoleRenderer()
    )

    formatter = structlog.stdlib.ProcessorFormatter(
        foreign_pre_chain=shared_processors,
        processors=[
            structlog.stdlib.ProcessorFormatter.remove_processors_meta,
            log_renderer,
        ],
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(log_level.upper())

    # Route uvicorn logs through structlog; suppress access log (re-emitted by middleware)
    for name in ("uvicorn", "uvicorn.error"):
        logging.getLogger(name).handlers.clear()
        logging.getLogger(name).propagate = True
    logging.getLogger("uvicorn.access").handlers.clear()
    logging.getLogger("uvicorn.access").propagate = False
```

---

### `backend/config.py` (config) — D-01, D-11

**Change type:** Add three new settings fields to the existing `Settings` class. Pattern: follow the existing field declaration style (`backend/config.py` lines 1–74) — typed field with default, grouped by phase/feature with a comment.

**Existing field pattern** (`backend/config.py` lines 48–71):
```python
# AI classification defaults (Phase 3 — D-13, D-15)
system_prompt: str = ""
default_ai_provider: str = "ollama"
default_ai_model: str = "llama3.2"

# Cloud Storage (Phase 5)
cloud_creds_key: str = "CHANGEME-32-bytes-padded!!"
google_client_id: str = ""
```

**New fields to append** (after the Cloud Storage block, before `settings = Settings()`):
```python
# Observability (Phase 6 — D-01)
log_level: str = "INFO"           # LOG_LEVEL env var; passed to setup_logging()
log_json: bool = False             # LOG_JSON env var; True in production

# Rate limiting (Phase 6 — D-11)
# Comma-separated list of trusted proxy CIDRs; requests from these may set X-Forwarded-For
trusted_proxy_cidrs: list[str] = [
    "127.0.0.0/8",
    "172.16.0.0/12",
    "192.168.0.0/16",
    "::1/128",
]
```

**env_list_separator** — already set to `","` in `model_config` (line 11), so `TRUSTED_PROXY_CIDRS=127.0.0.0/8,172.16.0.0/12` is parsed correctly out of the box.

---

### `backend/load_tests/locustfile.py` (test, request-response) — D-04, D-05, D-06

**Change type:** New file in new directory `backend/load_tests/`. No existing Locust file. Closest analog is the auth fixture pattern in `backend/tests/conftest.py`.

**Auth pattern from conftest.py** — the login flow the Locust `on_start()` replicates (`backend/tests/conftest.py` lines 186–226):
```python
# auth_user fixture shows the login payload shape:
token = create_access_token(str(user_id), "user")
headers = {"Authorization": f"Bearer {token}"}

# Locust replicates the same via HTTP POST:
resp = self.client.post(
    "/api/auth/login",
    json={"email": TEST_EMAIL, "password": TEST_PASSWORD},
)
self.access_token = resp.json().get("access_token", "")
```

**New file pattern** (from RESEARCH.md Pattern 7):
```python
"""Locust load test for DocuVault — D-04, D-05, D-06.

Run:
    locust --headless --users 50 --spawn-rate 10 --run-time 5m \\
           --host http://localhost:8000 \\
           --csv backend/load_tests/results \\
           -f backend/load_tests/locustfile.py

Prerequisites: a user with TEST_EMAIL/TEST_PASSWORD must exist in the DB.
Create via: POST /api/auth/register (on_start handles this — registers if not exists).
"""
import os
from locust import HttpUser, task, between, events

TEST_EMAIL = os.environ.get("LOAD_TEST_EMAIL", "loadtest@example.com")
TEST_PASSWORD = os.environ.get("LOAD_TEST_PASSWORD", "Loadtest123!@#")

class DocuVaultUser(HttpUser):
    wait_time = between(0.5, 2.0)
    access_token: str = ""

    def on_start(self):
        # Register if not exists (catches 409 Conflict silently)
        self.client.post(
            "/api/auth/register",
            json={"handle": "loadtestuser", "email": TEST_EMAIL, "password": TEST_PASSWORD},
        )
        resp = self.client.post(
            "/api/auth/login",
            json={"email": TEST_EMAIL, "password": TEST_PASSWORD},
        )
        if resp.status_code == 200:
            self.access_token = resp.json().get("access_token", "")
        else:
            self.environment.runner.quit()

    def _auth_headers(self):
        return {"Authorization": f"Bearer {self.access_token}"}

    @task(5)
    def list_documents(self):
        self.client.get("/api/documents/", headers=self._auth_headers())

    @task(2)
    def upload_document(self):
        # NOTE: confirm upload endpoint shape against documents.py before finalizing
        # If two-step presigned flow: POST /upload-url → PUT to MinIO → POST /{id}/confirm
        from io import BytesIO
        data = b"%PDF-1.4 1 0 obj<</Type/Catalog>>endobj"
        self.client.post(
            "/api/documents/upload",
            files={"file": ("test.pdf", BytesIO(data), "application/pdf")},
            headers=self._auth_headers(),
        )

    @task(1)
    def refresh_token(self):
        self.client.post("/api/auth/refresh")


@events.quitting.add_listener
def check_sla(environment, **kwargs):
    stats = environment.runner.stats.total
    if stats.fail_ratio > 0.01:
        environment.process_exit_code = 1
    elif stats.get_response_time_percentile(0.95) > 200:
        environment.process_exit_code = 1
    elif stats.get_response_time_percentile(0.99) > 500:
        environment.process_exit_code = 1
```

**Also create:** `backend/load_tests/__init__.py` (empty) so pytest does not discover this directory.

**Credentials security:** TEST_EMAIL and TEST_PASSWORD read from env vars — never hardcoded in version-controlled files.

---

### `backend/Dockerfile` (config) — D-07, D-08, D-09

**Change type:** Full replacement of single-stage build with multi-stage.

**Current file** (`backend/Dockerfile` lines 1–16):
```dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
    tesseract-ocr libgl1 libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
```

**Replacement pattern** (from RESEARCH.md Pattern 5, D-07):
```dockerfile
# Stage 1: builder — installs Python packages as root
FROM python:3.12-slim AS builder
WORKDIR /build
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    && rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: runtime — non-root appuser, no build tools
FROM python:3.12-slim AS runtime
# Runtime system deps (tesseract-ocr, libgl1, libglib2.0-0 are required at runtime)
RUN apt-get update && apt-get install -y --no-install-recommends \
    tesseract-ocr \
    libgl1 \
    libglib2.0-0 \
    && rm -rf /var/lib/apt/lists/*
COPY --from=builder /install /usr/local
RUN groupadd --gid 1000 appgroup && \
    useradd --uid 1000 --gid appgroup --shell /bin/sh --no-create-home appuser
WORKDIR /app
COPY --chown=appuser:appgroup . .
USER appuser
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
```

**Note on `--prefix=/install`:** Verify that `pip install --prefix=/install` followed by `COPY --from=builder /install /usr/local` correctly populates Python site-packages in the runtime stage. An alternative is `pip install --target=/install` with `PYTHONPATH` set. The prefix approach is preferred. Verify in Wave 0 smoke test: `docker run --rm docuvault-backend:latest python -c "import structlog"`.

---

### `docker-compose.yml` (config) — D-08, D-09, D-02

**Change type:** Modify existing service definitions for `backend` and `celery-worker`; add `loki`, `promtail`, `grafana` services; add named volumes.

**Existing service definition pattern** (`docker-compose.yml` lines 49–80) — the `backend` service to extend:
```yaml
backend:
  build: ./backend
  ports:
    - "8000:8000"
  volumes:
    - ./backend:/app
  environment:
    - DATABASE_URL=${DATABASE_URL}
    # ... (existing env vars) ...
  command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
  depends_on:
    postgres:
      condition: service_healthy
    ...
```

**Hardening additions for `backend` and `celery-worker`** (D-08, D-09):
```yaml
# Add these keys to both backend and celery-worker service definitions:
read_only: true
tmpfs:
  - /tmp:mode=1777        # world-writable; appuser (uid=1000) can write; covers tempfile.NamedTemporaryFile
cap_drop:
  - ALL
security_opt:
  - no-new-privileges:true
labels:
  logging: "promtail"     # Promtail docker_sd_configs filter label (D-02)
```

**New env vars for `backend` service** (D-01):
```yaml
environment:
  # ... existing vars ...
  - LOG_LEVEL=${LOG_LEVEL:-INFO}
  - LOG_JSON=${LOG_JSON:-false}
```

**New services block** (D-02):
```yaml
  loki:
    image: grafana/loki:latest
    ports:
      - "3100:3100"
    volumes:
      - ./docker/loki/loki-config.yaml:/etc/loki/local-config.yaml
      - loki_data:/loki
    command: -config.file=/etc/loki/local-config.yaml

  promtail:
    image: grafana/promtail:latest
    volumes:
      - ./docker/loki/promtail-config.yaml:/etc/promtail/config.yaml
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - /var/run/docker.sock:/var/run/docker.sock
    command: -config.file=/etc/promtail/config.yaml
    depends_on:
      - loki

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
    volumes:
      - grafana_data:/var/lib/grafana
    depends_on:
      - loki
```

**New volumes** (append to existing `volumes:` block):
```yaml
volumes:
  postgres_data:     # existing
  minio_data:        # existing
  loki_data:         # new
  grafana_data:      # new
```

**Celery-beat exclusion:** D-08 says `read_only: true` applies to "FastAPI and Celery worker services" — the `celery-beat` service writes `celerybeat-schedule` to its working directory. Do NOT apply `read_only: true` to `celery-beat`. If hardening is desired later, add `--schedule /tmp/celerybeat-schedule` to its command.

---

## Shared Patterns

### Existing Limiter declaration (all auth endpoint rate limiting)

**Source:** `backend/api/auth.py` lines 37–44
**Apply to:** `backend/api/auth.py` (replace `get_remote_address` with `get_client_ip`)

```python
# Current (to be replaced):
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)

# Replacement:
from deps.utils import get_client_ip
limiter = Limiter(key_func=get_client_ip)
```

### Per-account rate limiter (second Limiter instance)

**Source:** RESEARCH.md Pattern 4
**Apply to:** `backend/api/documents.py`, `backend/api/cloud.py`
**Where to define it:** A shared module — `backend/api/rate_limiting.py` or `backend/main.py` — so both document and cloud routers import the same instance.

```python
from slowapi import Limiter
from fastapi import Request

def _account_key(request: Request) -> str:
    user = getattr(request.state, "current_user", None)
    if user is None:
        return request.client.host if request.client else "anonymous"
    return str(user.id)

account_limiter = Limiter(key_func=_account_key)
```

```python
# Usage in each authenticated endpoint:
@router.get("/")
@account_limiter.limit("100/minute")
async def list_documents(
    request: Request,                              # must be first positional param
    current_user: User = Depends(get_regular_user),
    ...
):
    request.state.current_user = current_user      # expose to key_func — MUST be first line
    structlog.contextvars.bind_contextvars(user_id=str(current_user.id))
    ...
```

### Structlog logger usage in route handlers

**Source:** RESEARCH.md Code Examples section
**Apply to:** `backend/api/documents.py`, `backend/api/cloud.py`, any router that has authenticated endpoints

```python
import structlog
log = structlog.get_logger()

async def some_endpoint(..., current_user: User = Depends(get_regular_user)):
    structlog.contextvars.bind_contextvars(user_id=str(current_user.id))
    log.info("event.name", field=value)
```

### Pydantic Settings field pattern

**Source:** `backend/config.py` lines 13–72
**Apply to:** All new env vars in `backend/config.py`

```python
# Pattern: typed field + default value + inline comment with phase reference
field_name: type = default_value  # ENV_VAR_NAME env var; description (Phase N — Decision ref)
```

### Async test client with auth headers

**Source:** `backend/tests/conftest.py` lines 186–226
**Apply to:** `backend/tests/test_logging.py`, `backend/tests/test_rate_limiting.py`

```python
@pytest_asyncio.fixture
async def auth_user(db_session: AsyncSession):
    # Returns: {"user": User, "token": str, "headers": {"Authorization": "Bearer <token>"}}
    ...

# Usage in test:
async def test_something(async_client, auth_user):
    resp = await async_client.get("/api/documents/", headers=auth_user["headers"])
    assert resp.status_code == 200
```

---

## No Analog Found

Files with no close match in the codebase (planner should use RESEARCH.md patterns directly):

| File | Role | Data Flow | Reason |
|------|------|-----------|--------|
| `docker/loki/loki-config.yaml` | config | — | No YAML service configs exist in the repo; use RESEARCH.md Pattern 6 (loki-config.yaml section) verbatim |
| `docker/loki/promtail-config.yaml` | config | — | No YAML service configs exist in the repo; use RESEARCH.md Pattern 6 (promtail-config.yaml section) verbatim |
| `RUNBOOK.md` | documentation | — | No operational runbook exists; D-14 describes content fully |

---

## Critical Notes for Planner

1. **`get_client_ip` is a body replacement, not a new function.** CLAUDE.md mandates one canonical definition in `deps/utils.py`. Every router already imports it by name — the import chain stays intact.

2. **`CorrelationIDMiddleware` must use raw ASGI, not `BaseHTTPMiddleware`.** The existing `SecurityHeadersMiddleware` and `OriginValidationMiddleware` use `BaseHTTPMiddleware` — do not copy that pattern for `CorrelationIDMiddleware`. See RESEARCH.md Anti-Patterns section.

3. **Locust must NOT be in `requirements.txt`.** It is a dev/external tool. Add to `requirements-dev.txt` or run from a host virtualenv. The locustfile has zero imports from the application codebase.

4. **`read_only: true` excludes `celery-beat`** (Pitfall 7 in RESEARCH.md). The D-08 scope is "FastAPI and Celery worker services" only.

5. **`tmpfs: - /tmp:mode=1777`** is required (not just `/tmp`). Without `mode=1777`, appuser (uid=1000) cannot write to the tmpfs-mounted `/tmp` — `services/extractor.py` uses `tempfile.NamedTemporaryFile()` which writes to `/tmp`.

6. **Wave 0 assumption to verify:** `request.state.current_user` set as first line of handler body must be read correctly by slowapi's `key_func` before it increments the counter. Write a unit test (`test_rate_limiting.py::test_account_limiter_key`) before applying the decorator to all endpoints.

---

## Metadata

**Analog search scope:** `backend/api/`, `backend/services/`, `backend/deps/`, `backend/tests/`, `docker-compose.yml`, `backend/Dockerfile`, `backend/config.py`
**Files scanned:** 13
**Pattern extraction date:** 2026-06-02