2026-08-24 01:00:11 +03:00

75 KiB

Raw Blame History

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:

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

The owner has explicitly pivoted away from all three:

Queue → RabbitMQ, and the worker becomes event-driven (pipeline fan-out: extract stage, then analyze stage), not a single in-process job.
Storage → MinIO (local/self-hosted S3), not Selectel.
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) deferred to a follow-up; the invoices table ships empty in the initial migration for forward compatibility.
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, ЮKassa integration, 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`, 6 Dockerfiles in `srv/<svc>/`	one-image `MODE` dispatch
Scope now	api + worker-extract + worker-prescreen + worker-analyze + bot (+ prototype benchmark)	(web/payments 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 now (fine-grained stage), SSE/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.yml          (planned — scrape api + workers :9100/:9101/:9102)
│       ├── otel-collector-config.yaml (planned)
│       ├── tempo.yaml              (planned — trace storage)
│       └── 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)
│   │   │   ├── 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

api publishes DocumentUploaded to contracts.x rk extract (persistent, correlation_id UUID in both message id and headers["x-correlation-id"]).
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.
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.
worker-analyze consumes from analyze.q with prefetch=3 (mirrors Ollama Pro concurrency). On success: save report, status=done, ack. On failure: retry.
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.
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).
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 (STUB — no ЮKassa logic this refactor; forward-compatible schema)
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')),
    provider    TEXT,                             -- 'yookassa'
    external_id TEXT,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    paid_at     TIMESTAMPTZ
);
CREATE INDEX invoices_user_idx ON invoices (user_id, created_at DESC);

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.

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:

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

Cluster RabbitMQ to 3 nodes across hosts (quorum queues auto-replicate).
Add Postgres replica + Patroni (automated failover).
Run ≥2 api, ≥2 worker-extract, ≥2 worker-analyze behind the same Rabbit/PG/MinIO (stateless services scale horizontally for free).
Move MinIO to distributed mode (≥4 nodes, erasure coding).
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`
`SENTRY_DSN`	(empty)	if set, init sentry
`OTEL_EXPORTER_OTLP_ENDPOINT`	(empty)	OTel collector
`OTEL_SERVICE_NAME`	per-service	overridden in each service settings

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. Observability behind --profile obs. 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
   bot:             build srv/bot/Dockerfile; depends_on api healthy (NOT pg/rabbit)

  # ── OBSERVABILITY (profile: obs) ── PLANNED (T-E1-010)
  prometheus:      prom/prometheus; scrape api:9100, extract:9101, analyze:9102
  grafana:         grafana/grafana; provisioned datasources + starter dashboards
  otel-collector:  otel/opentelemetry-collector-contrib; receives OTLP
  tempo:           grafana/tempo; trace storage

  # ── EDGE (profile: edge) ── PLANNED (T-E1-009)
  nginx:           nginx:alpine; reverse proxy → api; TLS via certbot
  certbot:         certbot/certbot; renew cron sidecar

volumes: { pgdata, pgwal, redisdata, rabbitmq, minio }

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.

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

Provisioned Prometheus + Tempo datasources.
Starter dashboard: queue depth (rabbitmq_queue_messages), job duration histogram, LLM tokens/min, refund rate, 429/fallback rate.

OpenTelemetry (`core/telemetry.py`)

OTLP exporter to otel-collector → Tempo.
Auto-instrument FastAPI (api), httpx (all outbound, incl. Ollama calls).
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/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…"). SSE/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,healthz,readyz,metrics} → api:8000.
TLS via certbot; cert files bind-mounted; renew cron sidecar.
Webhook target for future ЮKassa: https://<domain>/api/v1/webhooks/yookassa.

Single-VPS deploy

VPS (Hetzner/Selectel), Docker + compose installed.
Clone, cp .env.example .env, fill secrets (DB, Rabbit, MinIO, Ollama, Sentry, tokens).
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.
docker compose --profile services --profile obs --profile edge up -d.
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) — next focused iteration after core lands.
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

75 KiB Raw Blame History Unescape Escape