kite/.planning/phases/06-performance-production-hardening/06-PATTERNS.md

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):

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):

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:

# 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):

# 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):

# 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:

# 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:

# 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):

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:

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:

@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:

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):

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):

"""
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:

"""
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):

# 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()):

# 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):

# 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):

"""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):

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):

# 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:

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):

# 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):

environment:
  # ... existing vars ...
  - LOG_LEVEL=${LOG_LEVEL:-INFO}
  - LOG_JSON=${LOG_JSON:-false}

New services block (D-02):

  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):

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)

# 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.

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)

# 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

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

# 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

@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