DealDocumentScreening/docs/ARCHITECTURE.md

ARCHITECTURE — «Контракт-чек» production refactor

Audience: a fresh agent (future me) picking this repo up cold. This doc is the single source of truth for the refactor. It supersedes the stage-1 architecture in IMPLEMENTATION_PLAN.md and the relevant tickets in TICKETS.md (T-E1-008/011/012/015 and anything mentioning arq, Selectel S3, or the MODE=... dispatcher). The stage-0 prototype and its tickets (T-E0-*) remain valid. BUSINESS_IDEA.md (product/business) is unchanged.

One-line summary: take the stage-0 prototype (PDF/DOCX → text → Ollama Cloud → markdown report) and rebuild it as a production, event-driven, multi-container application: a FastAPI api, three RabbitMQ-consuming workers (extract/OCR, deterministic prescreen, and LLM-analyze, split for CPU vs I/O profiles), a Telegram bot adapter, backed by Postgres + RabbitMQ + MinIO + Redis, observable via structlog + Sentry + Prometheus/Grafana + OpenTelemetry, behind Nginx + certbot. Monorepo, six fine-tuned Docker images (srv/ — api + three workers + bot + prototype), shared contract_check.core domain package. Compose now, k8s-ready later. HA-ready (quorum queues, WAL archiving Postgres), not HA-running on one VPS — see §10.

---

1. Why this refactor (the pivot)

The prototype proved the LLM path end-to-end (T-E0-005). The original plan (IMPLEMENTATION_PLAN.md) carried that into production with three assumptions:

1. arq + Redis as the job queue.
2. Selectel S3 (cloud, RF-located) as object storage.
3. One shared Docker image with a MODE=api|worker|bot runtime dispatcher.

The owner has explicitly pivoted away from all three:

1. Queue → RabbitMQ, and the worker becomes event-driven (pipeline fan-out: extract stage, then analyze stage), not a single in-process job.
2. Storage → MinIO (local/self-hosted S3), not Selectel.
3. Images → one Dockerfile per service, deps fine-tuned per image; the MODE dispatcher is killed (one image = one entrypoint).

Additional decisions locked during the architecture questionnaire:

• Three worker pools, not one: worker-extract (pymupdf + Tesseract, CPU), worker-prescreen (hybrid heuristic → optional LLM-fallback metadata extraction, CPU but tiny), and worker-analyze (LLM, I/O). The prescreen stage routes low-risk contracts to manual/auto-approval and high-risk contracts to deep LLM analysis, protecting the expensive LLM worker from trivial documents.
• LLM behind a provider interface (core/llm/port.py), Ollama Cloud as the first realization — swappable later for self-hosted Ollama / GigaChat.
• Refund policy is a runtime switch (REFUND_POLICY=all|infra_only) with a failure-class taxonomy; credit refund stays idempotent.
• Observability is full, not logs-only: structlog+correlation_id, Sentry, Prometheus+Grafana, OpenTelemetry.
• Payments (ЮKassa) live behind a provider port (core/billing/port.py), webhook-driven state machine, plans/subscriptions + credit top-ups, and a worker-billing scheduler (renewals/expiry/reconciliation). Disabled by default (YOOKASSA_ENABLED=false → 503, catalog stays readable).
• Web SPA deferred; api/worker/bot land first.
• Landing is incremental, verified green at each step.

Non-goals for this refactor (do NOT build these now): multi-tenant orgs, React SPA, K8s manifests, RAG/vector DB, template generation, E-sign/Gosuslugi.

---

2. System overview

Hexagonal (ports & adapters). The core owns all state and side effects (DB, S3, MQ, LLM, credits). Adapters (bot, future web/cli) are thin HTTP clients to the api and touch nothing but the api.

                        ┌───────────────────── HTTPS (Nginx + certbot) ─────────────────────┐
                        │                                                                    │
   Telegram ───►  bot (aiogram)  ──HTTP──►  api (FastAPI)  ──publish──►  RabbitMQ (direct)
    (adapter, HTTP-only)                      (core)                          │
                                              │  owns: Postgres, MinIO,          ▼
                                              │  Redis, credits, tokens     extract.q ──► worker-extract
                                              │  (core)                      (quorum)     (core, CPU: pymupdf+tesseract)
                                              ▼                                               │ publish PrescreenRequested
                                         Postgres                                            ▼
                                         MinIO ◄──── read/write ────►  prescreen.q ──► worker-prescreen
                                          Redis (rate limit/sess)      (quorum)        (hybrid heuristic router)
                                                                                        │ publish AnalyzeRequested
                                                                                        ▼
                                                                                  analyze.q ──► worker-analyze
                                                                                   (quorum)     (core, I/O: LLM provider)
                                                                                              │ save Report, status=done
                                                                                              ▼
                                                                                           Postgres

Data plane:

• Postgres — source of truth (users, documents, reports, jobs, tokens, invoices). Replication-ready (§10).
• MinIO — raw doc blobs + extracted text blobs, TTL-purged.
• RabbitMQ — durable job pipeline, quorum queues, TTL retry, DLQ.
• Redis — rate limiting (future), sessions (future). Idle for now but in compose; do not use it as a job queue.

Control plane:

• api is the only writer to Postgres/MinIO for user-initiated mutations. Workers write their own stage rows and reports but never create users or move credits except via the idempotent refund_credit helper.

---

3. Locked decisions (cheat sheet)

Area	Decision	Replaces
Repo	Monorepo, shared contract_check.core, 7 Dockerfiles in srv/<svc>/	one-image MODE dispatch
Scope now	api + worker-extract + worker-prescreen + worker-analyze + worker-billing + worker-notify + bot (+ prototype benchmark)	(web later)
Queue	RabbitMQ, direct exchange, pipeline fan-out	arq + Redis
Retry	TTL retry queues, exponential backoff, final DLQ	—
Workers	Three pools (extract/OCR + prescreen + analyze/LLM)	single arq worker
Storage	MinIO, users/{uid}/docs/{did}.{ext}	Selectel S3
Upload	Proxy through API (multipart)	—
Doc retention	TTL purge of raw docs after N days	—
Redis	Kept (rate limit/sessions future)	(no longer the queue)
Auth	Per-adapter service_tokens, revocable	single SERVICE_TOKEN
Report delivery	Polling + SSE (GET /reports/{id}/events, web-контракт analysisResultSchema); webhook later	—
Sync /analyze	No	—
Doc status	Fine-grained queued→extracting→prescreening→ocr→analyzing→done|failed	coarse status
Extra tables	jobs, service_tokens, invoices(stub)	3-table plan
Report storage	JSONB + markdown column	—
Durability	HA-ready (quorum queues, WAL archive)	—
Observability	structlog+corr, Sentry, Prom/Grafana, OTel	docker logs
LLM	Provider port + Ollama Cloud adapter	direct client
Refund	Policy switch all|infra_only + failure classes	refund-all
Edge	Nginx + certbot	—
Deploy	Compose now, k8s-ready later	—
Prototype	Kept as standalone benchmark	—
Tests	pytest+respx unit + testcontainers integration	—
Python	3.13 (was 3.14) — wheel availability	py3.14
Landing	Incremental, green per step	—

---

4. Repository layout (target)

Existing modules migrate as annotated. New code is marked (new).

DealDocumentScreening/
├── ARCHITECTURE.md                 (this file)
├── BUSINESS_IDEA.md                (unchanged — product/business)
├── IMPLEMENTATION_PLAN.md          (stage-0 valid; stage-1+ superseded here)
├── TICKETS.md                      (T-E0-* valid; T-E1-* superseded here)
├── README.md                       (rewrite: how to run the stack)
├── pyproject.toml                  (rewrite: core + per-service extras + dev group)
├── uv.lock                         (regenerated)
├── .env.example                    (rewrite: full env list, §11)
├── docker-compose.yml              (rewrite: infra + services, profiles, §12)
├── srv/                            (new — one Dockerfile per service, deps-tuned)
│   ├── api/Dockerfile              (new)
│   ├── worker-extract/Dockerfile   (new, tesseract layer)
│   ├── worker-analyze/Dockerfile   (new, lean)
│   ├── bot/Dockerfile              (new)
│   └── prototype/Dockerfile        (new, preserves stage-0 CLI)
├── deploy/                         (PLANNED — not yet built; lands with §20 steps 6 / E1-009..010)
│   ├── nginx/
│   │   ├── templates/contract-check.conf.template  (reverse proxy + TLS)
│   │   └── certbot-init.sh         (initial cert + nginx reload)
│   └── observability/
│       ├── prometheus/prometheus.yml  (scrape api + workers :9100..:9105)
│       ├── otel-collector-config.yaml (OpenObserve profile: receiver + forwarder)
│       ├── loki-config.yaml        (Grafana/Loki profile)
│       ├── tempo.yaml              (planned — trace storage for Grafana stack)
│       └── grafana/provisioning/
│           ├── datasources/        (prometheus + tempo)
│           └── dashboards/         (starter: queue depth, job latency, LLM tokens)
├── migrations/                     (new — alembic)
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
│       ├── 0001_initial.py         (6 tables, §7)
│       └── 0002_api_keys.py        (api_keys, api_key_requests — B2B, §15)
├── src/contract_check/
│   ├── __init__.py
│   ├── core/                       (new — shared domain, imported by every service image)
│   │   ├── __init__.py
│   │   ├── config.py               (pydantic-settings: base + per-service, §11)
│   │   ├── logging.py              (structlog JSON + correlation_id contextvar)
│   │   ├── telemetry.py            (OTel SDK init, FastAPI/asyncio instrumentation)
│   │   ├── sentry.py               (sentry_sdk init helper)
│   │   ├── metrics.py              (prometheus_client registry + counters/hists)
│   │   ├── api_keys.py             (B2B key gen/hash/verify — sha256 + hmac.compare_digest)
│   │   ├── rate_limit.py           (token-bucket per api_key_id; Memory + Redis backends)
│   │   ├── redis_client.py         (async Redis client from redis_url)
│   │   ├── db/
│   │   │   ├── __init__.py
│   │   │   ├── models.py           (SQLAlchemy 2 decl: User, PasskeyCredential, Document, Report, PrescreenResult, Job, ServiceToken, ApiKey, ApiKeyRequest, Invoice)
│   │   │   ├── session.py          (async_sessionmaker, engine)
│   │   │   └── enums.py            (DocStatus, JobStatus, QueueName, RoutingDecision, FailureClass — as plain str constants)
│   │   ├── mq/
│   │   │   ├── __init__.py
│   │   │   ├── topology.py         (exchange/queue/rk constants + declare_all())
│   │   │   ├── publisher.py        (aio-pika RobustChannel, publisher confirms)
│   │   │   ├── consumer.py         (base Consumer class: connect/prefetch/handle/nack-retry)
│   │   │   └── messages.py         (pydantic: DocumentUploaded, DocumentExtracted)
│   │   ├── s3/
│   │   │   ├── port.py             (Storage Protocol: put/get/stat/delete/presign)
│   │   │   └── minio_storage.py    (MinIO adapter, bucket init, key builders)
│   │   ├── llm/
│   │   │   ├── port.py             (LLMProvider Protocol + AnalysisResult dataclass)
│   │   │   ├── ollama_cloud.py     (ports prototype llm_client: repair-loop + 429 fallback)
│   │   │   └── factory.py          (provider selection by LLM_PROVIDER env)
│   │   ├── extraction/             (new — DocumentExtractor port + adapters)
│   │   │   ├── port.py             (ExtractedDocument, DocumentExtractor Protocol, errors)
│   │   │   ├── factory.py          (mime/suffix → adapter; SUPPORTED_SUFFIXES gate)
│   │   │   ├── adapters/           (pdf_pymupdf, docx_mammoth, rtf_striprtf, txt_chardet, ocr_tesseract)
│   │   │   └── __init__.py         (extract_document orchestrator with OCR fallback)
│   │   ├── analysis/               (existing; now consumed by prototype + ocr adapter)
│   │   │   ├── __init__.py
│   │   │   ├── extractor.py        (← from contract_check/extractor.py; prototype use)
│   │   │   ├── chunker.py          (← from contract_check/chunker.py; + chunk_markdown)
│   │   │   ├── checklist.py        (← from contract_check/checklist.py)
│   │   │   ├── report_schema.py    (← from contract_check/report_schema.py)
│   │   │   ├── ocr.py              (pytesseract via pymupdf rasterize)
│   │   │   └── analyzer.py         (orchestrate chunk→LLM→merge→dedupe→sort→markdown)
│   │   ├── credits.py              (reserve_credit atomic UPDATE; refund_credit idempotent + policy)
│   │   └── tokens.py               (service-token hash/verify; FastAPI dependency)
│   ├── api/                        (new — FastAPI image)
│   │   ├── __init__.py
│   │   ├── __main__.py             (uvicorn entrypoint)
│   │   ├── app.py                  (create_app: middleware, routes, lifespan)
│   │   ├── deps.py                 (db session, s3, publisher, service-token dep, api-key dep, rate-limit)
│   │   ├── services.py             (shared upload→reserve→MinIO→publish logic; used by documents + b2b)
│   │   ├── routes/
│   │   │   ├── health.py           (/healthz, /readyz)
│   │   │   ├── documents.py        (POST /api/v1/documents — upload→MinIO→publish, reserve)
│   │   │   ├── reports.py          (GET /api/v1/reports/{id} — 202+stage or 200+md; /events — SSE)
│   │   │   ├── me.py               (GET /api/v1/me — credits balance)
│   │   │   ├── metrics.py          (/metrics — prometheus)
│   │   │   └── b2b.py              (X-API-Key: POST /analyze, GET /b2b/reports, /b2b/usage, /b2b/keys CRUD)
│   │   └── middleware.py           (correlation_id inject, request metrics, Sentry)
│   ├── worker_extract/             (new — CPU image: pymupdf + tesseract)
│   │   ├── __init__.py
│   │   ├── __main__.py             (entrypoint: start consumer + metrics server)
│   │   ├── handler.py              (handle DocumentUploaded: MinIO dl → extraction factory → OCR fallback → upload .txt (markdown) → publish PrescreenRequested)
│   │   └── consumer.py             (wire handler into core.mq.consumer base)
│   ├── worker_prescreen/           (new — lean CPU image: hybrid heuristic + optional LLM fallback)
│   │   ├── __init__.py
│   │   ├── __main__.py             (entrypoint: metrics server on :9104)
│   │   ├── config.py               (PrescreenSettings, prefetch tunable)
│   │   ├── handler.py              (handle PrescreenRequested: download text → hybrid extract → route → persist prescreen_results → publish AnalyzeRequested/auto-approve/manual-review)
│   │   ├── consumer.py             (wire handler into core.mq.consumer base)
│   │   ├── extractor.py            (PrescreenContractMeta + MetadataExtractor protocol + compat shim)
│   │   ├── extractor_heuristic.py  (Stage 1: deterministic keyword/positional extractor, regex-free)
│   │   ├── extractor_llm.py        (Stage 2: dict → validated PrescreenContractMeta)
│   │   ├── extractor_hybrid.py     (orchestrator: heuristic → optional LLM fallback)
│   │   └── router.py               (decision logic: auto_approve / manual_review / deep_analysis)
│   ├── worker_analyze/             (new — I/O image: LLM provider)
│   │   ├── __init__.py
│   │   ├── __main__.py
│   │   ├── handler.py              (handle AnalyzeRequested: text dl→chunk→LLM (with prescreen context)→validate→save report→done)
│   │   └── consumer.py
│   ├── bot/                        (new — aiogram adapter, HTTP-only)
│   │   ├── __init__.py
│   │   ├── __main__.py
│   │   ├── config.py               (BotSettings: BOT_TOKEN, API_URL, BOT_SERVICE_TOKEN, poll tuning)
│   │   ├── client.py               (ApiClient: typed httpx wrapper; ApiError/NoCreditsError/...)
│   │   └── handlers.py             (/start, doc upload→POST api, poll→send report)
│   └── prototype/                  (moved from contract_check/prototype.py — kept as benchmark)
│       ├── __init__.py
│       └── __main__.py             (stage-0 CLI, imports updated to core.analysis.*)
└── tests/
    ├── conftest.py                 (fixtures: testcontainers pg/rabbit/minio)
    ├── unit/
    │   ├── test_chunker.py
    │   ├── test_extraction_adapters.py
    │   ├── test_extraction_factory.py
    │   ├── test_extractor.py
    │   ├── test_prescreen_extractor.py
    │   ├── test_prescreen_router.py
    │   ├── test_checklist_report.py
    │   ├── test_llm_ollama_cloud.py (respx: 200 / 429-fallback / repair)
    │   ├── test_credits.py
    │   ├── test_messages.py
    │   ├── test_rate_limit.py
    │   ├── test_bot_client.py
    │   └── test_bot_boundary.py     (static AST: bot must not import core.db/s3/llm/mq/credits)
    └── integration/
        ├── conftest.py             (seed service-token + api-key; testcontainers wiring)
        ├── test_upload_pipeline.py (POST /documents → message on extract.q or prescreen.q)
        ├── test_extract_worker.py  (DocumentUploaded → PrescreenRequested published)
        ├── test_analyze_worker.py  (AnalyzeRequested → Report saved, status done)
        ├── test_credits_db.py      (reserve/refund against real Postgres)
        └── test_b2b_api.py         (X-API-Key: 202 → poll report; 401/429 branches)

Import rule (hexagonal boundary, enforced in review):

• core/* may import anything.
• api/*, worker_extract/*, worker_prescreen/*, worker_analyze/* import only core/*.
• bot/* imports only httpx + its own modules. It must NOT import core.db, core.s3, core.llm, core.mq, core.credits. It speaks to the api over HTTP only. (Add a ruff/flake8 import-forbidden rule or a unit test that asserts this.)
• prototype/* imports core.analysis.* but is otherwise standalone (no DB/MQ).

---

5. RabbitMQ topology (full spec)

No plugins required. Pure direct exchanges + TTL/DLX for retry. All declarations in core/mq/topology.py and idempotent (declare on every service start).

Exchanges (direct)

Name	Type	Purpose
contracts.x	direct	Main exchange. Routing keys: extract, prescreen, analyze.
contracts.retry.x	direct	DLX target of main queues; routing keys: retry.extract, retry.prescreen, retry.analyze.

Queues

Name	Type	Args	Bound (exchange / rk)	Consumers
extract.q	quorum	x-dead-letter-exchange=contracts.retry.x, x-dead-letter-routing-key=retry.extract	contracts.x / extract	worker-extract
prescreen.q	quorum	x-dead-letter-exchange=contracts.retry.x, x-dead-letter-routing-key=retry.prescreen	contracts.x / prescreen	worker-prescreen
analyze.q	quorum	x-dead-letter-exchange=contracts.retry.x, x-dead-letter-routing-key=retry.analyze	contracts.x / analyze	worker-analyze
extract.retry.q	classic	x-dead-letter-exchange=contracts.x, x-dead-letter-routing-key=extract; per-message expiration	contracts.retry.x / retry.extract	none (delay slot)
prescreen.retry.q	classic	x-dead-letter-exchange=contracts.x, x-dead-letter-routing-key=prescreen; per-message expiration	contracts.retry.x / retry.prescreen	none (delay slot)
analyze.retry.q	classic	x-dead-letter-exchange=contracts.x, x-dead-letter-routing-key=analyze; per-message expiration	contracts.retry.x / retry.analyze	none (delay slot)
extract.dlq	quorum	—	—	none (manual requeue)
prescreen.dlq	quorum	—	—	none (manual requeue)
analyze.dlq	quorum	—	—	none (manual requeue)

Flow & retry mechanics

1. api publishes DocumentUploaded to contracts.x rk extract (persistent, correlation_id UUID in both message id and headers["x-correlation-id"]).
2. worker-extract consumes from extract.q with prefetch=1..N (config, default 1 — extraction is CPU-bound). On success: upload extracted Markdown to MinIO (text/markdown), publish PrescreenRequested to contracts.x rk prescreen (is_structured/has_tables metadata included), ack. On failure: see retry below.
3. worker-prescreen consumes from prescreen.q with prefetch=1. It runs a hybrid two-stage extractor: Stage 1 is a deterministic regex-free heuristic over keyword dictionaries and positional windows; Stage 2 is an LLM fallback (extract_prescreen on core/llm/port.py) triggered only when heuristic confidence is below PRESCREEN_LLM_FALLBACK_THRESHOLD and the kill-switch PRESCREEN_LLM_FALLBACK_ENABLED is true. It computes a weighted confidence_score, routes to auto_approve (lightweight report, disabled by default), manual_review (terminal DB state for future admin SPA), or deep_analysis. For deep_analysis it publishes AnalyzeRequested to contracts.x rk analyze carrying prescreen_meta. On failure: retry. The v1 regex implementation is kept behind PRESCREEN_KEEP_REGEX=true for one release as a rollback safety net.
4. worker-analyze consumes from analyze.q with prefetch=3 (mirrors Ollama Pro concurrency). On success: save report, status=done, ack. On failure: retry.
5. Retry (nack): the base consumer does not use basic.nack(requeue=True) (instant re-redelivery, no backoff). Instead it publishes a copy of the message to the matching *.retry.q with:
   • headers["x-attempt"] = attempt + 1
   • expiration = str(int(BASE_DELAY_MS * 2 ** attempt)) (e.g. BASE_DELAY_MS=2000 → 2s, 4s, 8s, 16s, 32s)
   • then basic_ack the original. When the retry queue's per-message TTL expires, its DLX bounces the message back to contracts.x with rk extract/prescreen/analyze → re-enters the main quorum queue. Clean, plugin-free, exponential.
6. Poison (max attempts): when headers["x-attempt"] >= MAX_ATTEMPTS (default 5), the consumer publishes to extract.dlq / prescreen.dlq / analyze.dlq instead of retry, acks the original, sets jobs.dlq=true, jobs.last_failure_class, transitions documents.status=failed, and calls refund_credit(document_id, failure_class) per policy (§8).
7. Manual requeue: a CLI script scripts/mq_requeue.py (or mgmt UI) moves a DLQ message back to the main exchange with reset attempt.

Prefetch & concurrency

• worker-extract: RabbitMQ prefetch caps concurrent extraction jobs (CPU bound; default 1, tune to CPU count).
• worker-prescreen: prefetch=1; Stage 1 heuristic is cheap but routing decisions must be ordered per document.
• worker-analyze: prefetch caps concurrent LLM jobs (default 3, mirrors Ollama Pro). Additionally keep the in-process asyncio.Semaphore from the prototype llm_client — prefetch caps jobs, the semaphore caps parallel chunk requests within a job. Both are needed for long multi-chunk contracts.

Idempotency

Every handler starts by re-reading documents.status (and jobs row) by correlation_id/document_id. If already terminal (done/failed) or the job is mid-flight by another consumer, ack and exit — do not re-run the LLM, do not double-refund. This preserves the existing idempotency invariant from T-E1-008 across the queue pivot.

Message schemas (core/mq/messages.py, pydantic v2)

class DocumentUploaded(BaseModel):
    correlation_id: UUID
    document_id: UUID
    user_id: UUID
    s3_key: str  # users/{uid}/docs/{did}.{ext}
    filename: str
    mime: str
    attempt: int = 0


class DocumentExtracted(BaseModel):
    correlation_id: UUID
    document_id: UUID
    user_id: UUID
    extracted_s3_key: str  # users/{uid}/docs/{did}.txt
    char_count: int
    ocr_used: bool
    is_structured: bool = False
    has_tables: bool = False
    attempt: int = 0


class PrescreenRequested(BaseModel):
    correlation_id: UUID
    document_id: UUID
    user_id: UUID
    text_s3_key: str  # users/{uid}/docs/{did}.txt
    filename: str
    char_count: int = 0
    ocr_used: bool = False
    is_structured: bool = False
    has_tables: bool = False
    attempt: int = 0


class PrescreenCompleted(BaseModel):
    correlation_id: UUID
    document_id: UUID
    user_id: UUID
    text_s3_key: str
    filename: str
    prescreened_at: str
    contract_type: str | None = None
    party_a: str | None = None
    party_b: str | None = None
    total_amount: float | None = None
    currency: str | None = None
    start_date: str | None = None
    end_date: str | None = None
    has_penalty_clause: bool | None = None
    has_termination_clause: bool | None = None
    has_arbitration: bool | None = None
    confidence_score: float | None = None
    routing_decision: str  # auto_approve | manual_review | deep_analysis
    auto_summary: str | None = None
    auto_findings: list[dict] = []
    extractor_version: str = "heuristic-v2"  # heuristic-v2 | hybrid-llm-v1
    attempt: int = 0


class AnalyzeRequested(BaseModel):
    correlation_id: UUID
    document_id: UUID
    user_id: UUID
    extracted_s3_key: str
    filename: str
    char_count: int = 0
    ocr_used: bool = False
    is_structured: bool = False
    has_tables: bool = False
    prescreen_meta: PrescreenCompleted | None = None
    attempt: int = 0

RabbitMQ headers: x-correlation-id, x-attempt, x-origin (api | worker-extract | worker-prescreen). content_type=application/json, delivery_mode=2 (persistent). Validate on consume with the pydantic model; on validation error → .dlq immediately with failure_class=infra.

---

6. MinIO (object storage)

Bucket & keys

• One bucket per env: contract-check-docs (configurable, S3_BUCKET).
• Layout: users/{user_id}/docs/{document_id}.{ext} (original upload) and users/{user_id}/docs/{document_id}.txt (extracted text, written by worker-extract, read by worker-analyze).
• Key builders in core/s3/minio_storage.py: original_key(uid, did, ext), extracted_key(uid, did). Never construct keys ad-hoc.

Upload path (proxy through API)

POST /api/v1/documents (multipart): the api validates mime/size, reserves the credit (§8), writes the blob to MinIO via the Storage port, creates the documents row (status=queued), inserts a jobs row, publishes DocumentUploaded, returns 202 + {document_id, correlation_id}. Adapters never see S3 credentials. Presigned URLs are a future optimization only.

Initialization

A one-shot minio-init service in compose runs mc alias set ... && mc mb ... && mc anonymous set none ... on boot. Idempotent. Creates the bucket before api/worker start. Alternatively core/s3/minio_storage.py does ensure_bucket() lazily on first use — keep both; lazy is the safety net.

Retention (152-ФЗ friendliness)

• MinIO ILM lifecycle rule: expire objects under users/*/docs/* after DOC_RETENTION_DAYS (default 7, configurable). Configured via mc ilm add/queue in minio-init, or via the MinIO console. The extracted .txt may share the rule or a longer one (TEXT_RETENTION_DAYS, default 30).
• The report (findings + markdown) lives in Postgres, not MinIO, so it survives the raw-doc purge. This is the 152-ФЗ lever: raw contract text leaves the system on a schedule, only the structured findings stay.

Encryption / versioning

• SSE-S3 (server-side encryption with MinIO-managed keys) on for production env, off for dev — toggle via S3_SERVER_SIDE_ENCRYPTION.
• Versioning off (documents are write-once; versioning adds cost and complicates TTL expiry).

---

7. Postgres schema & data access (Alembic migrations + repositories)

Six core tables plus additive migrations. status/queue/adapter/role columns are TEXT + CHECK (not Postgres enums) so migrations are additive — matches the convention noted in IMPLEMENTATION_PLAN.md §1.2.

7.1 Data-access layer

All production SQL lives behind the repository package:

src/contract_check/core/db/repositories/
├── api_keys.py          (ApiKeyRepository)
├── credits.py           (CreditsRepository — thin facade)
├── documents.py         (DocumentRepository)
├── jobs.py              (JobRepository)
├── passkeys.py          (PasskeyRepository)
├── prescreen_results.py (PrescreenResultRepository)
├── reports.py           (ReportRepository)
├── service_tokens.py    (ServiceTokenRepository)
└── users.py             (UserRepository)

Rules:

• A repository receives an AsyncSession; it never commits.
• Callers own transactions and commits.
• Raw text() is allowed only inside repositories or migrations.
• api/deps.py keeps lightweight dependency helpers but no direct SQL; it calls UserRepository, ApiKeyRepository, etc.
• core/credits.py remains the public facade for credit operations; it now delegates to CreditsRepository.

7.2 Migrations

Migrations (hand-written, async env.py):

• 0001 initial schema: users/documents/reports/jobs/service_tokens/invoices
• 0002 api_keys + api_key_requests (B2B)
• 0003 webUI auth columns on users (email, password_hash, reset tokens, is_active)
• 0004 users.role for the admin panel ('user'|'admin', default 'user')
• 0005 Telegram bot hardening columns on users
• 0006 prescreen_results table; expand documents.status and jobs.queue enums

-- users
CREATE TABLE users (
    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    telegram_id   BIGINT UNIQUE,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    credits_left  INTEGER NOT NULL DEFAULT 0,
    email         TEXT UNIQUE,                     -- added by 0003
    password_hash TEXT,                              -- added by 0003
    password_reset_token_hash TEXT,                  -- added by 0003
    password_reset_expires_at TIMESTAMPTZ,           -- added by 0003
    is_active     BOOLEAN NOT NULL DEFAULT TRUE,     -- added by 0003
    role          TEXT NOT NULL DEFAULT 'user'       -- added by 0004
                  CHECK (role IN ('user','admin')),
    CONSTRAINT users_credits_nonneg CHECK (credits_left >= 0),
    CONSTRAINT users_identity_present CHECK (telegram_id IS NOT NULL OR email IS NOT NULL)
);
CREATE INDEX users_role_idx ON users (role);         -- added by 0004

-- documents
CREATE TABLE documents (
    id                 UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id            UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    s3_key             TEXT NOT NULL,
    extracted_s3_key   TEXT,
    filename           TEXT NOT NULL,
    mime               TEXT NOT NULL,
    bytes              BIGINT NOT NULL DEFAULT 0,
    status             TEXT NOT NULL DEFAULT 'queued'
                       CHECK (status IN ('queued','extracting','prescreening','ocr','analyzing','done','failed')),
    stage              TEXT,                       -- sub-stage / human label
    refunded           BOOLEAN NOT NULL DEFAULT FALSE,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at         TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX documents_user_created_idx ON documents (user_id, created_at DESC);
CREATE INDEX documents_status_idx ON documents (status);

-- reports
CREATE TABLE reports (
    id                   UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id          UUID NOT NULL UNIQUE REFERENCES documents(id) ON DELETE CASCADE,
    content_json         JSONB NOT NULL,
    markdown             TEXT NOT NULL,
    model_used           TEXT,
    prompt_tokens        INTEGER NOT NULL DEFAULT 0,
    eval_tokens          INTEGER NOT NULL DEFAULT 0,
    latency_ms           INTEGER NOT NULL DEFAULT 0,
    prescreen_result_id  UUID REFERENCES prescreen_results(id) ON DELETE SET NULL,
    prescreen_meta       JSONB,
    created_at           TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- prescreen_results (added by 0006)
CREATE TABLE prescreen_results (
    id                     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id            UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    correlation_id         UUID NOT NULL,
    contract_type          TEXT,
    party_a                TEXT,
    party_b                TEXT,
    total_amount           NUMERIC(18, 2),
    currency               TEXT,
    start_date             DATE,
    end_date               DATE,
    has_penalty_clause     BOOLEAN,
    has_termination_clause BOOLEAN,
    has_arbitration        BOOLEAN,
    confidence_score       NUMERIC(4, 3) CHECK (confidence_score >= 0 AND confidence_score <= 1),
    routing_decision       TEXT NOT NULL DEFAULT 'manual_review'
                           CHECK (routing_decision IN ('auto_approve','manual_review','deep_analysis')),
    prescreened_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
    processing_ms          INTEGER,
    extractor_version      TEXT NOT NULL DEFAULT 'heuristic-v2'
                           CHECK (extractor_version IN ('regex-v1','heuristic-v2','hybrid-llm-v1')),
    auto_summary           TEXT,
    auto_findings          JSONB NOT NULL DEFAULT '[]',
    error_message          TEXT,
    retry_count            INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0)
);
CREATE INDEX ix_prescreen_results_document_id ON prescreen_results (document_id);
CREATE INDEX prescreen_results_routing_idx ON prescreen_results (routing_decision);
CREATE INDEX prescreen_results_confidence_idx ON prescreen_results (confidence_score);

-- jobs (RabbitMQ correlation; one document has up to 3 jobs: extract + prescreen + analyze)
CREATE TABLE jobs (
    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id         UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
    correlation_id      UUID NOT NULL,
    queue               TEXT NOT NULL CHECK (queue IN ('extract','prescreen','analyze')),
    attempts            INTEGER NOT NULL DEFAULT 0,
    max_attempts        INTEGER NOT NULL DEFAULT 5,
    last_failure_class  TEXT,
    last_error          TEXT,
    dlq                 BOOLEAN NOT NULL DEFAULT FALSE,
    status              TEXT NOT NULL DEFAULT 'pending'
                        CHECK (status IN ('pending','running','retrying','dlq','done')),
    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX jobs_correlation_idx ON jobs (correlation_id);
CREATE INDEX jobs_document_idx ON jobs (document_id);

-- service_tokens (per-adapter auth)
CREATE TABLE service_tokens (
    id           UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name         TEXT NOT NULL UNIQUE,            -- "bot-prod", "web-prod"
    token_hash   TEXT NOT NULL,                   -- sha256 hex of the bearer secret
    adapter      TEXT NOT NULL CHECK (adapter IN ('bot','web','cli')),
    revoked      BOOLEAN NOT NULL DEFAULT FALSE,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    last_used_at TIMESTAMPTZ
);

-- invoices (billing, migration 0011; money = kopecks int)
CREATE TABLE invoices (
    id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id           UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    amount            INTEGER NOT NULL,                 -- kopecks
    status            TEXT NOT NULL DEFAULT 'draft'
                      CHECK (status IN ('draft','pending','succeeded','cancelled','refunded')),
    kind              VARCHAR(12) CHECK (kind IN ('topup','subscription','renewal')),
    credits_purchased INT,                              -- topup: pack size
    subscription_id   UUID,                             -- subscription/renewal origin
    provider          TEXT,                             -- 'yookassa'
    external_id       TEXT,                             -- ЮKassa payment id
    confirmation_url  TEXT,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
    paid_at           TIMESTAMPTZ
);
CREATE INDEX invoices_user_idx ON invoices (user_id, created_at DESC);

-- user_profiles (1:1 passive settings, migration 0011)
CREATE TABLE user_profiles (
    user_id         UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
    language        VARCHAR(4),          -- 'ru' | 'be' | 'en' (stored passively)
    timezone        VARCHAR(64),         -- IANA name
    notif_prefs     JSONB NOT NULL DEFAULT '{"report_ready": true, "security": true, "marketing": false}',
    dashboard_prefs JSONB NOT NULL DEFAULT '{"severity_filter": "all", "per_page": 10, "density": "comfortable"}',
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- plans (lookup, seeded)
CREATE TABLE plans (
    code           VARCHAR(16) PRIMARY KEY,   -- free | lite | pro | max
    name           TEXT NOT NULL,
    price_kopecks  INT NOT NULL,              -- 0 / 49000 / 149000 / 390000
    monthly_quota  INT NOT NULL,              -- 0 / 5 / 20 / 60
    is_active      BOOLEAN NOT NULL DEFAULT TRUE,
    sort           INT NOT NULL
);

-- subscriptions (one active per user, partial unique index)
CREATE TABLE subscriptions (
    id                   UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id              UUID NOT NULL REFERENCES users(id),
    plan_code            VARCHAR(16) NOT NULL REFERENCES plans(code),
    status               VARCHAR(12) NOT NULL
                         CHECK (status IN ('active','past_due','expired','cancelled')),
    current_period_start TIMESTAMPTZ NOT NULL,
    current_period_end   TIMESTAMPTZ NOT NULL,
    auto_renew           BOOLEAN NOT NULL DEFAULT FALSE,   -- opt-in, default OFF
    origin_invoice_id    UUID REFERENCES invoices(id),
    created_at           TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at           TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX uq_subscriptions_active
    ON subscriptions (user_id) WHERE status IN ('active','past_due');

-- quota_usage (per-period plan usage ledger; UNIQUE(document_id) = idempotency)
CREATE TABLE quota_usage (
    id              BIGSERIAL PRIMARY KEY,
    user_id         UUID NOT NULL,
    subscription_id UUID NOT NULL REFERENCES subscriptions(id),
    document_id     UUID NOT NULL REFERENCES documents(id),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (document_id)
);

-- credit_events (append-only credit ledger; balance_after self-verifying)
CREATE TABLE credit_events (
    id            BIGSERIAL PRIMARY KEY,
    user_id       UUID NOT NULL,
    delta         INT NOT NULL CHECK (delta <> 0),
    kind          VARCHAR(16) NOT NULL CHECK (kind IN
                  ('signup','admin_grant','topup','reserve','refund_auto',
                   'clawback','quota_refund','hold_clear')),
    document_id   UUID,
    invoice_id    UUID,
    balance_after INT NOT NULL CHECK (balance_after >= 0),
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- users: + billing_hold BOOLEAN NOT NULL DEFAULT FALSE (refund-clawback gate,
-- uploads answer 402 "billing hold" until an admin clears it)

Alembic rules: env.py uses the async engine; migrations are hand-written (autogenerate used only to draft, never committed as-is); up and down both clean; CI runs alembic upgrade head then alembic downgrade base on a throwaway DB.

---

8. Credits & refund policy (billing invariants)

The reserve-on-enqueue invariant from T-E1-003 survives the queue pivot unchanged. RabbitMQ only changes who runs the job, not when the credit moves.

Reserve (api, on enqueue, synchronous)

async def reserve_credit(session, user_id) -> bool:
    row = await session.execute(
        text("""
            UPDATE users SET credits_left = credits_left - 1
            WHERE id = :u AND credits_left > 0
            RETURNING credits_left
        """),
        {"u": user_id},
    )
    return row.first() is not None

Atomic, never negative. If False, api returns 402 Payment Required synchronously — no MinIO write, no message published.

Refund (worker, on terminal failure, idempotent)

Guarded by documents.refunded so a duplicate message / requeue cannot double-refund:

NON_REFUNDABLE_INFRA_ONLY = {"extraction_failed"}  # user-garbage input


async def refund_credit(session, document_id, failure_class, policy) -> bool:
    if policy == "infra_only" and failure_class in NON_REFUNDABLE_INFRA_ONLY:
        return False  # user pays for undetectable garbage
    result = await session.execute(
        text("""
            UPDATE users
            SET credits_left = credits_left + 1
            WHERE id = (SELECT user_id FROM documents
                        WHERE id = :d AND refunded = FALSE)
            RETURNING id
        """),
        {"d": document_id},
    )
    if result.first() is None:
        return False  # already refunded (idempotent no-op)
    await session.execute(
        text("UPDATE documents SET refunded = TRUE WHERE id = :d"),
        {"d": document_id},
    )
    return True

Failure classes (core/db/enums.py)

Class	Meaning	Refundable under infra_only?
extraction_failed	pymupdf couldn't extract (>100 chars expected, got less) and OCR also failed	No (likely user garbage)
ocr_failed	Tesseract raised	Yes
llm_quota	429 / quota on both primary and fallback	Yes
llm_invalid_output	model returned unrepairable JSON	Yes
llm_timeout	Ollama timed out past retries	Yes
infra	DB/S3/MQ connectivity, unhandled exception	Yes
unknown	anything not classified	Yes

Set on the jobs.last_failure_class column. REFUND_POLICY env selects all (default) or infra_only.

---

8a. Billing: plans, payments, refunds (ЮKassa, migration 0011)

Money is integer kopecks everywhere; RUB decimals exist only at the ЮKassa HTTP boundary. Domain code lives in core/billing/ (port + yookassa adapter + quota + fulfillment + refunds), mirroring core/llm/.

Payment provider port

core/billing/port.py defines PaymentProvider: create_payment, get_payment, refund — all idempotence-keyed (key = invoice uuid). The ЮKassa adapter is httpx-based and respx-testable. YOOKASSA_ENABLED=false ⇒ build_payment_provider raises PaymentsDisabled ⇒ mutating billing routes answer 503 while GET /billing/plans stays readable (dev degrade).

Invoice state machine (webhook-driven, idempotent)

pending ──payment.succeeded──► succeeded   (topup: +credits ledger event;
        │                                  subscription: grant active period)
        ├──payment.canceled───► cancelled
        └──reconcile (pending >15m)──► re-fetch via API, apply same edges
succeeded ──refund──► refunded             (clawback + billing_hold, §8b)

Webhook trust model (api/routes/webhooks.py): ЮKassa notification Basic auth (shop id + secret, constant-time compare), then re-fetch the payment via the REST API — the push payload's amount is never trusted. Applying is delegated to core/billing/fulfillment.apply_payment_status, the same function the worker-billing reconciliation sweep uses; replays are no-ops.

Subscription lifecycle (worker-billing, 60 s tick)

• auto_renew = TRUE and period_end − 3d < now and no pending renewal invoice → create renewal invoice + payment.
• period_end < now: renewal succeeded → roll period forward (active); else → past_due (7-day grace) → expired.
• auto_renew = FALSE (default): expire at period end. Monthly quota resets on period roll — no rollover.

Upload quota ordering (D7)

billing_hold = TRUE?           → 402 "billing hold"  (hard stop, admin clears)
active subscription & quota rows this period < plan.monthly_quota
                               → insert quota_usage row (source='quota')
else credits_left > 0          → atomic credit reserve (source='credits')
else                           → 402 "no credits available"

Implemented in core/billing/quota.reserve_document_slot; gated behind PLANS_ENABLED (false = legacy credits-only).

Refund policy (D10) — core/billing/refunds.py

Usage = credit reserve events + quota_usage rows created after the invoice's paid_at (FIFO approximation).

• within REFUND_WINDOW_DAYS (14) and usage ≤ REFUND_FULL_USAGE_THRESHOLD (20%) of purchased ⇒ full refund;
• else ⇒ proportional: max(0, amount − used_docs × PRICE_PER_DOC_KOPECKS);
• already refunded / not succeeded / outside window / zero result ⇒ 409.

Execution: provider refund → invoice refunded → clawback (remove remaining purchased credits / cancel subscription + drop current-period quota rows, clawback ledger events) → if the balance went negative or a subscription was pulled back, set users.billing_hold = TRUE → uploads 402 until an admin clears it (audit trail: log lines, not ledger — delta <> 0 CHECK forbids zero-delta events).

Admin panel

/admin/invoices (global list, status/kind filters) + user-card billing blocks: invoices, credit-events tail, hold badge + "снять hold", manual refund (same service as the API route) with a full-refund override for support cases. ЮKassa credentials are env-only — never entered in the panel.

---

9. LLM provider port + adapters

The prototype's llm_client.py (bearer httpx, format: json-schema, 5xx backoff, 429→fallback, repair-loop) is excellent. It becomes the first adapter behind a port so we can later swap in self-hosted Ollama, GigaChat, or YandexGPT without touching analyzer.py.

Registered providers are selected by LLM_PROVIDER:

• ollama_cloud (default)
• yandex_gpt

Port (core/llm/port.py)

from typing import Protocol, Sequence
from dataclasses import dataclass


@dataclass(slots=True)
class AnalysisResult:
    findings: list  # list[Finding] (from core.analysis.report_schema)
    model_used: str
    fell_back: bool
    repaired: bool
    prompt_tokens: int
    eval_tokens: int
    latency_sec: float


class LLMProvider(Protocol):
    async def analyze(
        self, text: str, *, checklist: str, extra_context: str = ""
    ) -> AnalysisResult: ...
    async def aclose(self) -> None: ...

Adapter (core/llm/ollama_cloud.py)

Move the existing OllamaCloudClient + ChatResult + _QuotaError + _run_with_fallback + _run_repair_loop + _post_chat verbatim, then add an analyze(text, *, checklist) method that:

1. chunks the text via core.analysis.chunker.chunk_markdown (heading-aware when the extractor produced structured Markdown; falls back to plain chunking),
2. appends extra_context (prescreen metadata for AnalyzeRequested) to each user prompt,
3. builds the system/user prompts (from prototype.SYSTEM_PROMPT + build_user_prompt, lifted into core/analysis/analyzer.py),
4. fans out chunk requests under the existing asyncio.Semaphore,
5. merges/dedupes/sorts findings (lift dedupe_findings + sort_findings into core/analysis/analyzer.py),
6. returns AnalysisResult.

Map adapter-internal failures to FailureClass:

• _QuotaError escapes → llm_quota
• ValidationError after repair → llm_invalid_output
• httpx.TimeoutException → llm_timeout
• anything else → infra

Factory (core/llm/factory.py)

def build_llm_provider(settings) -> LLMProvider:
    match settings.llm_provider:
        case "ollama_cloud":
            return OllamaCloudProvider(settings)
        case "yandex_gpt":
            return YandexGPTProvider(settings)
        case _:
            raise ValueError(f"unknown LLM_PROVIDER={settings.llm_provider!r}")

LLM_PROVIDER env (default ollama_cloud). Additional providers register here.

---

10. Durability & HA posture (single VPS, HA-ready)

Confirmed posture: HA-ready on one VPS, not HA-running. True mirrored queues and Postgres replication require multiple hosts; on one VPS we get durability (survives reboot/crash) and recoverability (point-in-time), and the path to true HA is a topology change, not a code change.

RabbitMQ — quorum queues

• extract.q, prescreen.q, analyze.q, extract.dlq, prescreen.dlq, analyze.dlq are x-queue-type: quorum. Quorum queues persist every message to disk and use Raft. On a single node they are durable; the moment a 3-node RabbitMQ cluster is added (compose: rabbitmq@rmq1/2/3), the same queues replicate with no application code change.
• Retry queues are classic (transient delay slots; safe to lose on catastrophic failure — the originating message is already acked and tracked in jobs).
• Enable publisher confirms on the api publisher (await confirm per publish; if nacked/timeout, fail the request before telling the user 202). A paid job's message must never silently vanish.

Postgres — replication-ready

• wal_level=replica, archive_mode=on, archive_command to a mounted volume (or pg_backrest later). Gives PITR.
• A physical replication slot + hot standby is a documented add-host step (add postgres-replica service, primary_conninfo). Patroni/repmgr when automated failover is wanted.
• Daily pg_dump cron in compose (sidecar or host cron) — baseline backup.

Volumes & restart

• Named volumes for pgdata, rabbitmq, minio, redis, prometheus, grafana, tempo. Bind-mount only ./deploy configs and cert dirs.
• restart: unless-stopped on every long-running service.
• depends_on: condition: service_healthy everywhere with real healthchecks (pg pg_isready, rabbit rabbitmq-diagnostics ping, minio mc ready, redis redis-cli ping).

Scale-out checklist (when leaving one VPS)

1. Cluster RabbitMQ to 3 nodes across hosts (quorum queues auto-replicate).
2. Add Postgres replica + Patroni (automated failover).
3. Run ≥2 api, ≥2 worker-extract, ≥2 worker-analyze behind the same Rabbit/PG/MinIO (stateless services scale horizontally for free).
4. Move MinIO to distributed mode (≥4 nodes, erasure coding).
5. Reconsider Nginx → managed LB.

None of 1–5 requires touching core/ application code — only compose/infra.

---

11. Configuration (core/config.py, pydantic-settings)

12-factor: all config via env. Typed via pydantic-settings. Base Settings

• per-service subclasses. No hardcoded model names, prompts, delays.

Base (shared by all services)

Env	Default	Notes
ENV	dev	dev/staging/prod — toggles TLS, sentry sample rate
LOG_LEVEL	INFO	structlog level
LOG_FORMAT	json	json (prod/staging) or console (dev) — explicit override of env-based default
APP_VERSION	unknown	added to every log line; set at build/deploy time
DATABASE_URL	(required)	postgresql+asyncpg://...
REDIS_URL	redis://redis:6379/0	rate limit / sessions (future)
RABBITMQ_URL	(required)	amqp://guest:guest@rabbitmq:5672//
MQ_PREFETCH_EXTRACT	1	CPU-bound
MQ_PREFETCH_ANALYZE	3	mirrors Ollama Pro concurrency
MQ_PREFETCH_PRESCREEN	1	deterministic heuristic is tiny but keep ordered
MQ_MAX_ATTEMPTS	5	before DLQ
MQ_RETRY_BASE_MS	2000	exponential base
S3_ENDPOINT_URL	(required)	MinIO URL
S3_ACCESS_KEY / S3_SECRET_KEY	(required)
S3_BUCKET	contract-check-docs
S3_REGION	us-east-1	MinIO default
S3_SERVER_SIDE_ENCRYPTION	false	true in prod
DOC_RETENTION_DAYS	7	MinIO ILM expiry
TEXT_RETENTION_DAYS	30	extracted .txt expiry
REFUND_POLICY	all	all | infra_only
PLANS_ENABLED	false	subscription quota logic; false = credits-only legacy
YOOKASSA_ENABLED	false	payments master switch; false → billing mutations 503
YOOKASSA_SHOP_ID / YOOKASSA_SECRET_KEY	(empty)	ЮKassa merchant credentials
YOOKASSA_RETURN_BASE_URL	(empty)	/pay/{invoice_id} base for return links
PRICE_PER_DOC_KOPECKS	19900	pay-per-doc price (199 ₽)
BILLING_RETURN_JWT_SECRET	(empty → JWT_SECRET)	signs /pay/{id} tokens
BILLING_RETURN_TOKEN_TTL_MINUTES	15	pay-page token TTL
REFUND_WINDOW_DAYS	14	full-refund window (§8a)
REFUND_FULL_USAGE_THRESHOLD	0.20	usage ratio for full refund
SENTRY_DSN	(empty)	if set, init sentry
OTEL_EXPORTER_OTLP_ENDPOINT	(empty)	OTLP endpoint; with profile observer use http://otel-collector:4318
OTEL_SERVICE_NAME	per-service	overridden in each service settings
OPENOBSERVE_AUTH_TOKEN	default creds base64	Basic auth token the collector uses to forward to OpenObserve
OTEL_COLLECTOR_OTLP_GRPC_PORT	4317	host port for collector OTLP/gRPC receiver
OTEL_COLLECTOR_OTLP_HTTP_PORT	4318	host port for collector OTLP/HTTP receiver
OTEL_COLLECTOR_METRICS_PORT	8889	host port for collector self-metrics
GRAFANA_PORT	3000	host port for the Grafana UI (profile obs)
GRAFANA_ADMIN_USER	admin	initial Grafana admin user
GRAFANA_ADMIN_PASSWORD	admin	initial Grafana admin password
GRAFANA_ROOT_URL	http://localhost:3000	external Grafana URL for callbacks

LLM

Env	Default	Notes
LLM_PROVIDER	ollama_cloud	factory selection: ollama_cloud, yandex_gpt
OLLAMA_HOST	(required if provider=ollama_cloud)	https://ollama.com for Ollama Cloud; http://localhost:11434 for self-hosted
OLLAMA_API_KEY	(required)	bearer
OLLAMA_MODEL	qwen2.5:14b	primary; must be a model available at the configured host
OLLAMA_FALLBACK_MODEL	qwen2.5:7b	429 fallback; must also exist on the host
OLLAMA_TEMPERATURE	0.2
OLLAMA_NUM_PREDICT	3072
OLLAMA_TIMEOUT	120	seconds
OLLAMA_MAX_CONCURRENCY	3	in-process semaphore (per analyze worker)
CHUNK_SIZE_CHARS	10000	chunker
YANDEXGPT_API_KEY	(required if provider=yandex_gpt)	API key for Yandex Cloud Foundation Models
YANDEXGPT_FOLDER_ID	(required if provider=yandex_gpt)	Yandex Cloud folder ID (gpt://<folder_id>/<model>)
YANDEXGPT_MODEL	yandexgpt-lite	primary YandexGPT model URI name
YANDEXGPT_FALLBACK_MODEL	(empty)	optional fallback model name
YANDEXGPT_BASE_URL	https://llm.api.cloud.yandex.net	completion API base URL
YANDEXGPT_COMPLETION_PATH	/foundationModels/v1/completion	completion API path
YANDEXGPT_TEMPERATURE	0.2
YANDEXGPT_MAX_TOKENS	3072	max completion tokens
YANDEXGPT_TIMEOUT	120	seconds
YANDEXGPT_MAX_CONCURRENCY	3	in-process semaphore (per analyze worker)

Prescreen

Env	Default	Notes
PRESCREEN_ENABLED	true	Run the prescreen stage between extraction and analysis.
PRESCREEN_AUTO_APPROVE	false	Allow auto-approval of low-risk contracts (disabled until accuracy proven).
PRESCREEN_CONFIDENCE_THRESHOLD	0.75	Minimum weighted field-coverage score to route away from manual_review.
PRESCREEN_HIGH_VALUE_THRESHOLD	100000	Amount (in contract currency) above which contract is always sent to deep analysis.
PRESCREEN_LLM_FALLBACK_ENABLED	false	Kill-switch for the Stage-2 LLM fallback in the hybrid extractor. Flip after burn-in.
PRESCREEN_LLM_FALLBACK_THRESHOLD	0.75	Heuristic confidence below this triggers the LLM fallback (keep ≤ confidence threshold).
PRESCREEN_LLM_MAX_CHARS	20000	Cap on characters sent to the LLM fallback (cost control).
PRESCREEN_KEEP_REGEX	false	Temporary: run the legacy regex-v1 extractor via extract_contract_meta shim.

API

Env	Default	Notes
API_HOST	0.0.0.0
API_PORT	8000
API_METRICS_PORT	9100
B2B_DEFAULT_RATE_LIMIT_RPS	3	per API key; mirrors Ollama Pro concurrency, overridable per api_keys.rate_limit_rps
CORS_ORIGINS	(empty, future web)

Auth (JWT + Telegram identity verification + webUI + admin panel)

Env	Default	Notes
TELEGRAM_BOT_TOKEN	(empty)	used by API to verify Login Widget / Mini App signatures
JWT_SECRET	(required)	HS256 secret for signing user JWTs; generate with openssl rand -hex 32
JWT_ALGORITHM	HS256
JWT_ACCESS_TTL_MINUTES	1440 (24h)	access-token lifetime; tune per env
JWT_REFRESH_TTL_DAYS	30	refresh-token lifetime for webUI sessions
WEB_AUTH_ENABLED	true	toggle /api/v1/auth/{register,login,logout,forgot-password,reset-password}
WEB_APP_BASE_URL	http://localhost:5173	base URL of the future web SPA; used for password-reset links
PASSWORD_RESET_TTL_MINUTES	60	reset-link validity
PASSWORD_MIN_LENGTH	8	enforced at register, reset, and in the /admin create form
WEB_ADMIN_ENABLED	true	toggle the /admin/* server-rendered management UI
ADMIN_REQUIRED_ROLE	admin	users.role value required to enter /admin (must match the DB CHECK constraint)

Bot

Env	Default
BOT_TOKEN	(required)
API_URL	http://api:8000
BOT_SERVICE_TOKEN	(required — bearer for adapter auth to /api/v1/auth/telegram/bot, looked up against service_tokens)

Prototype (standalone benchmark image)

Same Ollama env as above; no DB/MQ/S3 env needed.

---

12. Docker — images & compose

Per-service Dockerfile pattern (uv, multi-stage, py3.13)

Dockerfiles live in srv/<service>/Dockerfile (one per service). Common shape (shown for api):

# syntax=docker/dockerfile:1
FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS builder
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=never \
    UV_PROJECT_ENVIRONMENT=/app/.venv
WORKDIR /app
COPY pyproject.toml uv.lock ./
COPY README.md ./
COPY src ./src
RUN --mount=type=cache,target=/root/.cache/uv \
    uv sync --frozen --no-default-groups --group api

FROM python:3.13-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PATH=/app/.venv/bin:$PATH
WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
# Tesseract ONLY in srv/worker-extract/Dockerfile:
# RUN apt-get update && apt-get install -y --no-install-recommends \
#     tesseract-ocr tesseract-ocr-rus tesseract-ocr-eng \
#  && rm -rf /var/lib/apt/lists/*
EXPOSE 8000
CMD ["python", "-m", "contract_check.api"]

Per-service differences (uv uses PEP 735 dependency-groups — see pyproject.toml):

Dockerfile	Extra installed	Runtime apt	CMD	Expose
srv/api/Dockerfile	--group api	none	python -m contract_check.api	8000, 9100
srv/worker-extract/Dockerfile	--group extract	tesseract-ocr, -rus, -eng, libmagic1	python -m contract_check.worker_extract	9101
srv/worker-prescreen/Dockerfile	--group prescreen	none	python -m contract_check.worker_prescreen	9104
srv/worker-analyze/Dockerfile	--group analyze	none	python -m contract_check.worker_analyze	9102
srv/bot/Dockerfile	--group bot	none	python -m contract_check.bot	—
srv/prototype/Dockerfile	--group prototype	none	python -m contract_check.prototype	—

The bot image is the leanest (no DB driver, no S3 client, no pymupdf). The analyze image has httpx but no tesseract/pymupdf. The extract image is the heaviest (tesseract + language packs + libmagic1). This is the "fine-tuned deps per service" payoff.

pyproject.toml dependency-groups (actual — PEP 735)

The project uses [dependency-groups] (not [project.optional-dependencies]). Core deps (installed in every image) stay minimal; service-specific libs live in groups so each Docker image runs uv sync --no-default-groups --group <svc>. Shared db/mq/s3/obs groups are include-group-ed by the service groups that need them. Sketch (see pyproject.toml for the authoritative list):

[project]
name = "contract-check"
requires-python = ">=3.13"
dependencies = [
    "pydantic>=2.7", "pydantic-settings>=2.3", "structlog>=24.1",
    "python-dotenv>=1.0", "httpx[http2]>=0.27",
]

[dependency-groups]
db  = ["sqlalchemy>=2.0", "asyncpg>=0.29", "alembic>=1.13"]
mq  = ["aio-pika>=9.4"]
s3  = ["minio>=7.2"]
obs = ["prometheus-client>=0.20", "sentry-sdk>=2",
       "opentelemetry-sdk>=1.24", "opentelemetry-exporter-otlp>=1.24"]

api      = [{ include-group = "db" }, { include-group = "mq" },
           { include-group = "s3" }, { include-group = "obs" },
           "fastapi>=0.110", "uvicorn[standard]>=0.29", "python-multipart>=0.0.9",
           "redis>=5.0",
           "opentelemetry-instrumentation-fastapi>=0.45b0",
           "opentelemetry-instrumentation-asgi>=0.45b0"]
extract  = [{ include-group = "db" }, { include-group = "mq" },
           { include-group = "s3" }, { include-group = "obs" },
           "pymupdf>=1.24", "pytesseract>=0.3.10", "pillow>=10",
           "mammoth>=1.8", "striprtf>=0.0.26", "chardet>=5.2",
           "python-magic>=0.4.27"]
analyze  = [{ include-group = "db" }, { include-group = "mq" },
           { include-group = "s3" }, { include-group = "obs" },
           "opentelemetry-instrumentation-httpx>=0.45b0"]
prescreen = [{ include-group = "db" }, { include-group = "mq" },
            { include-group = "s3" }, { include-group = "obs" }]
bot      = ["aiogram>=3.4"]
prototype = ["pymupdf>=1.24", "python-docx>=1.1"]
dev = [{ include-group = "api" }, { include-group = "extract" },
       { include-group = "prescreen" }, { include-group = "analyze" },
       { include-group = "bot" }, { include-group = "prototype" },
       "pytest>=8", "pytest-asyncio>=0.23", "respx>=0.21", "ruff>=0.5",
       "mypy>=1.10", "anyio>=4", "aiosqlite>=0.20",
       "testcontainers[rabbitmq,postgres,minio]>=4", "asgi-lifespan>=2.1.0"]

docker-compose.yml structure

One file, profiles. Default (docker compose up) = infra only. Services behind --profile services. Two observability stacks are mutually exclusive: --profile obs (Grafana + Loki + Prometheus) or --profile observer (OpenObserve + OTel Collector). Edge behind --profile edge.

services:
  # ── INFRA (default profile) ──
  postgres:        postgres:18-alpine, wal_level=replica, archive, healthcheck, volumes (pgdata, pgwal); host port 15432
  redis:           redis:8-alpine, aof, healthcheck, volume; host port 17379
  rabbitmq:        rabbitmq:4-management-alpine, healthcheck, volume; AMQP 5672, mgmt UI 15672
  minio:           minio/minio, healthcheck, volume, console on :9001; S3 API :9000
  minio-init:      one-shot: mc alias + mb + ilm rule; depends_on minio healthy

  # ── SERVICES (profile: services) ──
  api:             build srv/api/Dockerfile; depends_on pg/rabbit/minio-init healthy;
                   healthcheck /healthz; ports 8000,9100
   worker-extract:  build srv/worker-extract/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9101
   worker-prescreen: build srv/worker-prescreen/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9104
   worker-analyze:  build srv/worker-analyze/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9102
   worker-billing:  build srv/worker-billing/Dockerfile; depends_on pg/minio-init healthy; 9105
   worker-notify:   build srv/worker-notify/Dockerfile; depends_on rabbit healthy; 9103
   bot:             build srv/bot/Dockerfile; depends_on api healthy (NOT pg/rabbit)

  # ── OBSERVABILITY — GRAFANA STACK (profile: obs) ──
  loki:            grafana/loki; filesystem-backed single-node log store
  promtail:        grafana/promtail; ships Docker container logs into Loki
  grafana:         grafana/grafana; provisioned Loki datasource + logs dashboard
  prometheus:      prom/prometheus; scrapes service /metrics endpoints

  # ── OBSERVABILITY — OPENOBSERVE STACK (profile: observer) ──
  openobserve:     public.ecr.aws/zinclabs/openobserve:latest; UI + OTLP backend on :5080, gRPC on :5081
  otel-collector:  otel/opentelemetry-collector-contrib:latest; OTLP receiver 4317/4318,
                   scrapes /metrics from services, forwards logs/metrics/traces to OpenObserve

  # ── EDGE (profile: edge) ──
  nginx:           nginx:alpine; reverse proxy → api/openobserve/grafana; TLS via certbot
  certbot:         certbot/certbot; renew cron sidecar

volumes: { pgdata, pgwal, redisdata, rabbitmq, minio, loki-data, grafana-data, prometheus-data, openobserve-data }

bot depends on api healthy (not on infra) — it speaks HTTP to the api, enforcing the adapter boundary even in dependency ordering.

---

13. Observability

Structured logging (core/logging.py)

• structlog with JSON renderer in prod, console renderer in dev.
• A correlation_id contextvars.ContextVar is the spine of every log line.
• api middleware reads X-Correlation-ID header or mints a UUID4, sets the contextvar, and the api publisher stamps headers["x-correlation-id"] on every published message.
• workers read headers["x-correlation-id"] on consume and set the same contextvar before handling. So a single upload's logs trace api → rabbit → worker-extract → worker-analyze → DB under one ID.
• OTel baggage/span context propagates the same ID for distributed traces.
• The same correlation_id is parsed by the Loki datasource as a derived field, so you can click through from any log line to every other log line with the same id.

Sentry (core/sentry.py)

• sentry_sdk.init(dsn=SENTRY_DSN, environment=ENV, traces_sample_rate=...) in each service's entrypoint.
• FastAPI integration in api, asyncio integration in workers.
• Sample rate: 1.0 in dev, 0.1 in prod (env-driven).

Prometheus (core/metrics.py + /metrics)

• api: http_requests_total, http_request_duration_seconds, documents_uploaded_total, credits_reserved_total, mq_publish_total.
• worker-extract: extract_duration_seconds{format=...}, extraction_total{format,structured}, expose on :9101.
• worker-prescreen: prescreen_duration_seconds{decision=...}, prescreen_runs_total{decision,contract_type}, prescreen_confidence, prescreen_fallback_runs_total{outcome=used|failed|skipped|disabled}, prescreen_extraction_stage_seconds{stage=heuristic|llm}, expose on :9104.
• worker-analyze: analyze_duration_seconds, llm_tokens_total{kind=prompt|eval}, llm_fell_back_total, expose on :9102.
• Workers run prometheus_client.start_http_server(port) in a background thread alongside the async consumer.

Grafana + Loki

• obs profile adds loki, promtail, grafana to compose.
• Promtail discovers all compose containers via the Docker socket and pushes their stdout/stderr to Loki; labels include service, project, profile, and container.
• Logs retain the JSON format emitted by structlog in staging/prod and the pretty console format in dev; Loki stores the raw line.
• Provisioned datasource Loki and a starter Contract Check — Logs dashboard at http://localhost:${GRAFANA_PORT:-3000}/d/contract-check-logs.
• Dashboard has a service filter and a free-text search box; error words are highlighted as annotations. Use correlation_id values to trace one upload across api → worker-extract → worker-prescreen → worker-analyze.

Prometheus + Tempo (Grafana stack, planned enhancements)

• prometheus in the obs profile scrapes :9100/:9101/:9102/:9104/:9105/:9103 metrics exposed by the services.
• Tempo will receive OTLP traces via the OpenTelemetry collector when added.
• Starter dashboards for queue depth, job latency, LLM tokens/min, refund rate, and 429/fallback rate will ship with the metrics/traces datasources.

OpenObserve + OTel Collector (observer profile)

• otel-collector runs in the observer profile and listens for OTLP on 4317 (gRPC) and 4318 (HTTP).
• Services export OTLP traces, metrics, and logs to http://otel-collector:4318 (OTEL_EXPORTER_OTLP_ENDPOINT).
• The collector also scrapes Prometheus /metrics from api:9100 and each worker (:9101/:9102/:9104/:9105/:9103).
• A dedicated otlphttp/openobserve exporter forwards all signals to http://openobserve:5080/api/default using OPENOBSERVE_AUTH_TOKEN.
• OpenObserve UI is exposed under /openobserve/ via nginx.

OpenTelemetry (core/telemetry.py)

• OTLP exporter sends traces, metrics, and logs to the endpoint configured in OTEL_EXPORTER_OTLP_ENDPOINT (collector in observer profile, Tempo in obs).
• Auto-instrument FastAPI (api), httpx (all outbound, incl. Ollama calls).
• configure_logging routes structlog through stdlib logging so the OTEL log handler captures application logs with their structured fields.
• Spans carry the same correlation_id as logs. Trace from HTTP request → RabbitMQ publish → consume → LLM call is one trace tree.

---

14. Testing strategy

Unit (tests/unit/, fast, no I/O)

• test_chunker.py — chunk_text + chunk_markdown.
• test_extraction_adapters.py, test_extraction_factory.py — adapters and factory.
• test_extractor.py — legacy prototype extractor path remains valid.
• test_llm_ollama_cloud.py — respx mocks: 200 happy, 429→fallback, invalid-JSON→repair→success, invalid-JSON→repair→fail, timeout.
• test_credits_policy.py — reserve_credit never negative; refund_credit idempotent (double-call refunds once); infra_only skips extraction_failed; concurrency: 5 parallel reserves against 1 credit → exactly 1 succeeds.
• test_messages.py — pydantic round-trip + header validation.
• adapter-boundary test: bot/ must not import any core.db/core.s3/ core.llm/core.mq/core.credits symbol (static AST check).

Integration (tests/integration/, testcontainers, slow)

• conftest.py fixtures start real Postgres + RabbitMQ + MinIO via testcontainers (one container each per session), apply migrations, wire to a throwaway MinIO bucket and RabbitMQ vhost.
• test_upload_pipeline.py — POST /api/v1/documents (TestClient) → credit reserved, blob in MinIO, DocumentUploaded lands on extract.q (assert via aio-pika consumer).
• test_extract_worker.py — publish DocumentUploaded, run worker-extract handler in-process → PrescreenRequested published, extracted Markdown in MinIO, documents.status=advanced to prescreening.
• test_analyze_worker.py — publish AnalyzeRequested with respx-mocked Ollama, run worker-analyze handler → report row in Postgres with status=done, markdown contains disclaimer.
• test_retry_and_dlq.py — force Ollama 500 repeatedly → message cycles retry queues with growing TTL, lands on analyze.dlq after MAX_ATTEMPTS, documents.status=failed, credit refunded per policy.

Marks: @pytest.mark.integration excluded from the default fast run; CI runs both. pytest -q (unit) < 5s; integration as a separate stage.

DoD per step

Every landing step ends with all three green:

ruff check . && mypy src && pytest -q          # unit, fast
pytest -m integration -q                       # integration, CI only

---

15. API surface (FastAPI)

JSON API routes live under /api/v1. The server-rendered admin panel is mounted at /admin/* and is the only HTML surface in the api; it reuses the user JWT for authentication (stored in an HttpOnly cookie).

Three JSON auth modes:

• User JWT (Authorization: Bearer <jwt>) — common token for bot users, Telegram Login Widget users, and Telegram Mini App users. Issued by /api/v1/auth/telegram/* after verifying the Telegram identity proof (core/auth.py). User endpoints rely on api/deps.py:require_current_user.
• Service token (Authorization: Bearer <token>) — adapter-level auth, validated against service_tokens (core/tokens.py). Used only by the bot adapter to call /api/v1/auth/telegram/bot and exchange a verified telegram_id for a user JWT.
• B2B API key (X-API-Key) — validated against api_keys.key_hash (core/api_keys.py + api/deps.py:require_api_key) with per-key token-bucket rate-limit (core/rate_limit.py). Used by external B2B clients.

Health/metrics exempt from auth.

Auth endpoints (api/routes/auth/) — telegram.py, password.py, passkeys.py, magic_links.py

Method	Path	Auth	Behavior
POST	/api/v1/auth/telegram/bot	service token	bot exchanges verified telegram_id for a user JWT
POST	/api/v1/auth/telegram/web	—	verify Telegram Login Widget payload → issue user JWT
POST	/api/v1/auth/telegram/miniapp	—	verify Mini App initData HMAC → issue user JWT
POST	/api/v1/auth/register	—	email/password → new user + JWT pair (requires WEB_AUTH_ENABLED)
POST	/api/v1/auth/login	—	email/password → JWT pair (requires WEB_AUTH_ENABLED)
POST	/api/v1/auth/logout	refresh token	revoke refresh (requires WEB_AUTH_ENABLED)
POST	/api/v1/auth/forgot-password	—	enqueue reset email (requires WEB_AUTH_ENABLED)
POST	/api/v1/auth/reset-password	reset token	rotate password (requires WEB_AUTH_ENABLED)
GET	/api/v1/auth/me	user JWT	introspect JWT claims

User endpoints (user JWT)

Method	Path	Auth	Behavior
GET	/healthz	none	200 liveness (process alive)
GET	/readyz	none	200 readiness (DB+Rabbit+MinIO reachable)
GET	/metrics	none	Prometheus exposition
POST	/api/v1/documents	user JWT	multipart → reserve credit → MinIO put → row queued → publish DocumentUploaded → 202 {document_id, correlation_id}. 402 if no credit. 400 bad mime/size.
GET	/api/v1/documents/{id}	user JWT	status + stage + filename (for polling UI)
GET	/api/v1/reports/{document_id}	user JWT	202 {status, stage} while not done; 200 {markdown, findings, ...} when done
GET	/api/v1/reports/{document_id}/events	user JWT (header или ?access_token= — для нативного EventSource)	SSE-стрим статусов: безымянные кадры с payload analysisResultSchema (camelCase, pending/processing/completed/failed); терминальные completed/failed закрывают стрим; служебные события timeout/error; keep-alive комментарии между кадрами
GET	/api/v1/me	user JWT	{telegram_id, credits_left}

B2B endpoints (X-API-Key, api/routes/b2b.py)

Method	Path	Auth	Behavior
POST	/api/v1/analyze	api key	multipart → reserve owner's credit → publish → 202 {document_id, correlation_id}; rate-limited (429), 401 bad/revoked key
GET	/api/v1/b2b/reports/{document_id}	api key	scoped to key owner; 202 {status, stage} or 200 {markdown, findings, ...}
GET	/api/v1/b2b/usage	api key	{monthly_quota, monthly_used, requests_this_month, resets_at} for the authenticating key
POST	/api/v1/b2b/keys	user JWT	create key; raw api_key returned once (only hash stored)
GET	/api/v1/b2b/keys	user JWT	list owner's keys
POST	/api/v1/b2b/keys/{id}/revoke	user JWT	revoke (instant auth disable)
GET	/api/v1/b2b/keys/{id}/usage	user JWT	per-month request counts

Admin panel (api/admin/)

Mounted at /admin/*, gated by WEB_ADMIN_ENABLED and by the ADMIN_REQUIRED_ROLE check on users.role. Stack: FastAPI + Jinja2 + HTMX (client-side via CDN). Reuses the same access JWT, but stored in an HttpOnly cookie (cc_admin_token) so browsers can navigate it.

Method	Path	Behavior
GET/POST	/admin/login	email/password → cookie; non-admin roles rejected
POST	/admin/logout	clear cookie
GET	/admin / /admin/	redirect to /admin/users if logged in, else /admin/login
GET	/admin/users	paginated, searchable user list
GET	/admin/users/new	create-user form
POST	/admin/users	create a user (email, password, role, credits, optional telegram_id)
GET	/admin/users/{user_id}	user detail + document stats
POST	/admin/users/{user_id}	update role / active flag / credits
POST	/admin/users/{user_id}/toggle-active	ban/unban (HTMX partial swap)
GET	/admin/users/{user_id}/credits	bump credits by delta (HTMX partial swap)

To grant admin access: set users.role = 'admin' (or the value of ADMIN_REQUIRED_ROLE) for the target account. The Makefile has helpers: make admin-promote EMAIL=... and make admin-list.

No synchronous /analyze (locked). Adapters poll /reports/{id}; the fine-grained stage field powers a progress signal in the bot ("Extracting text…", "Analyzing…"). Web clients can instead subscribe to the SSE stream GET /api/v1/reports/{id}/events: every frame carries the normalized analysisResultSchema payload (pending/processing/completed/failed, issues, derived riskScore/summary) — see api/schemas/analysis_result.py. Webhook added later.

---

16. Worker internals

worker-extract (worker_extract/handler.py)

on DocumentUploaded(msg):
    if documents.status by msg.document_id is terminal: ack; return
    set status=extracting (or ocr), jobs.status=running
    download s3_key from MinIO → temp file
    try: text = extractor.extract_text(tmp)
         ocr_used = False
    except ExtractionError (<100 chars, likely scan):
         set status=ocr
         text = ocr.ocr_pdf(tmp); ocr_used = True
         if still <100: raise with failure_class=extraction_failed
    upload extracted_key=text to MinIO
    publish DocumentExtracted(correlation_id, document_id, extracted_s3_key,
                              char_count=len(text), ocr_used, attempt)
    ack
on failure: consumer-base retry/dlq logic (§5), refund per policy (§8)

worker-analyze (worker_analyze/handler.py)

on DocumentExtracted(msg):
    if documents.status terminal: ack; return
    set status=analyzing, jobs.status=running
    text = s3.get(extracted_s3_key)
    result = llm_provider.analyze(text, checklist=checklist_for_prompt())
    validate result.findings via ReportPayload (repair already in provider)
    markdown = analyzer.render_markdown(result.findings, ...)
    save report (content_json, markdown, tokens, latency, model_used)
    set documents.status=done
    ack
on failure: classify failure_class, retry/dlq, refund per policy

LLM provider failures bubble as FailureClass; the consumer-base catches and routes to retry/DLQ/refund.

Concurrency reminders

• worker-extract: RabbitMQ prefetch caps jobs (CPU). Tesseract/pymupdf are sync → run in asyncio.to_thread/threadpool within the handler.
• worker-analyze: RabbitMQ prefetch (3) caps jobs; the provider's internal asyncio.Semaphore(OLLAMA_MAX_CONCURRENCY=3) caps parallel chunk LLM calls within a single multi-chunk contract. Both layers matter.

---

17. Bot adapter (bot/)

aiogram 3. Pure HTTP client to the api. Forbidden imports: core.db, core.s3, core.llm, core.mq, core.credits, sqlalchemy, minio, aio_pika, pymupdf. Enforced by a unit test (§14).

Flow:

• /start → GET /api/v1/me → greeting + credit balance.
• User sends PDF/DOCX → forward multipart to POST /api/v1/documents with Authorization: Bearer $BOT_SERVICE_TOKEN. React to 402 (no credit, polite refusal), 400 (bad format), 202 (acknowledged).
• Poll GET /api/v1/reports/{id} with backoff; surface stage as a status message ("Extracting…", "Analyzing…").
• On done: send markdown report; if >4096 chars, send as .md attachment. Every report carries the disclaimer (bot appends if api omitted).

No business logic, no direct state. The hexagonal rule.

---

18. Edge & deployment

Nginx + certbot (deploy/nginx/)

• Nginx reverse-proxies /{api,v1,pay,healthz,readyz,metrics} → api:8000.
• TLS via certbot; cert files bind-mounted; renew cron sidecar.
• ЮKassa webhook target: https://<domain>/api/v1/webhooks/yookassa (configure in the ЮKassa merchant dashboard; Basic auth = shop id + secret).

Single-VPS deploy

1. VPS (Hetzner/Selectel), Docker + compose installed.
2. Clone, cp .env.example .env, fill secrets (DB, Rabbit, MinIO, Ollama, Sentry, tokens).
3. Seed a service_tokens row for the bot (hash of BOT_SERVICE_TOKEN) via a one-shot python -m contract_check.api seed-token bot bot-prod.
4. docker compose --profile services --profile obs --profile edge up -d.
5. pg_dump cron for backups.

k8s-ready posture

• Every service exposes a healthcheck (/healthz) and is configurable purely via env (12-factor). No code change needed to move to k8s; only manifests/Helm charts (future).

---

19. Conventions & rules of the road

• Lint/types/tests green per step: ruff check . && mypy src && pytest.
• No comments in code (house style — existing prototype has none).
• Hexagonal boundary: adapters never import core state/infra. Enforced by a unit test.
• Reserve-on-enqueue invariant is sacred: credit moves on POST /documents, never in the worker.
• Refund idempotency is sacred: documents.refunded guards it.
• Config via env, never hardcoded model/prompt/delay.
• Every report carries the disclaimer "не заменяет юриста" — render in analyzer.render_markdown, assert in tests.
• Migrations hand-written, up and down both clean.
• One image = one entrypoint; MODE dispatcher is dead.
• Publisher confirms on; a published-but-unconfirmed message fails the HTTP request.
• Correlation ID flows HTTP → message header → logs → traces.

---

20. Landing sequence (incremental, green per step)

Each step is a verifiable unit. Do not start step N+1 until N is green (ruff && mypy && pytest + relevant integration).

Step 1 — Infra + core skeleton + initial migration

• Compose: postgres (replication-ready), redis, rabbitmq (mgmt UI exposed), minio, minio-init. All with healthchecks + volumes. Default profile = infra only.
• src/contract_check/core/: config, logging, telemetry, sentry, metrics, db/models (all 10 tables), db/session, mq/topology, mq/messages, s3/port, s3/minio_storage, llm/port, analysis/* (migrate extractor/chunker/checklist/ report_schema from prototype), credits, tokens.
• Alembic: init + initial migration (6 tables).
• Migrate prototype.py → prototype/__main__.py (update imports to core.analysis.*).
• DoD: docker compose up infra healthy; alembic upgrade head + downgrade base clean; ruff && mypy && pytest green (unit tests ported for chunker/extractor/credits/messages).

Step 2 — api image (done)

• srv/api/Dockerfile, src/contract_check/api/*: app factory, middleware (correlation_id, request metrics, Sentry), routes (health, ready, metrics, documents upload→MinIO→publish with confirms, reports poll with stage, me, b2b.py — X-API-Key endpoints), deps (db session, s3, publisher, service-token, api-key, rate-limit), services.py (shared upload logic).
• Reserve-on-enqueue; 402 path.
• Seed-token one-shot CLI.
• DoD: POST /api/v1/documents reserves credit, writes MinIO, publishes DocumentUploaded (verified on extract.q); unit + integration (test_upload_pipeline, test_b2b_api) green.

Step 3 — worker-extract image (done)

• srv/worker-extract/Dockerfile (tesseract layer), worker_extract/*: handler + consumer wiring + metrics server on :9101.
• core/analysis/ocr.py.
• Retry + DLQ via base consumer; failure classification; status updates.
• DoD: test_extract_worker + test_retry_and_dlq green; real PDF in MinIO → extracted text published on analyze.q.

Step 4 — worker-analyze image (done)

• srv/worker-analyze/Dockerfile, worker_analyze/*: handler + consumer + metrics on :9102.
• core/llm/ollama_cloud.py (ported) + factory; analyzer orchestration; report save with JSONB+markdown; refund-on-fail per policy.
• DoD: test_analyze_worker green (respx-mocked Ollama); end-to-end upload→report in integration; disclaimer present.

Step 5 — bot image (done)

• srv/bot/Dockerfile, bot/*: aiogram handlers, HTTP-only.
• Boundary test (no forbidden imports).
• DoD: local polling: send PDF → get report; boundary test green.

Step 6 (later) — payments, web, dashboards

• ЮKassa + invoices logic + recurring (separate effort).
• React SPA + Telegram Login + SSE/webhook delivery.
• Grafana dashboards polished; Tempo/Jaeger trace UI.

---

21. Open questions / deferred

• Payments (ЮKassa) — done (plans/subscriptions/top-ups/refunds, core/billing/, worker-billing; see §8a and .scratch/user-profile-billing/).
• Web SPA + Telegram Login auth — stage 2.
• B2B API keys + rate limit — stage 3 (done; Redis + core/api_keys.py + core/rate_limit.py + api/routes/b2b.py).
• Multi-host HA — when load justifies; path documented in §10.
• Self-hosted Ollama / GigaChat — behind the provider port when 152-ФЗ forces it or Ollama Cloud overage bites.
• OCR backend (Yandex Vision) — behind an OCRBackend port later.
• DLQ admin UI — a small management route/script for requeue; mgmt UI suffices initially.

---

22. Quick reference — file-to-rule index

• "What's the queue topology?" → §5, core/mq/topology.py
• "How does a credit move?" → §8, core/credits.py
• "What statuses can a document have?" → §7, core/db/enums.py
• "How is a job retried?" → §5 (retry mechanics), core/mq/consumer.py
• "How do I add a new LLM provider?" → §9, core/llm/factory.py
• "What can the bot import?" → §4 (boundary), §17, tests/unit/test_bot_boundary.py
• "Where do env vars live?" → §11, core/config.py, .env.example
• "How is this deployed?" → §18
• "How do we scale to HA?" → §10
• "What's the next thing to build?" → §20, step 1
• "How does the B2B API work?" → §15, api/routes/b2b.py, core/api_keys.py
• "How is rate limiting enforced?" → §15, core/rate_limit.py
• "Where are the Dockerfiles?" → §12, srv/<service>/Dockerfile