# Implementation Roadmap — Phase 2 (Prescreen) & Remaining Work > Based on `document-extraction-spec.md`, `needle-prescreen-integration.md`, > and the Phase 0 spike report (`docs/SPIKE_PHASE0.md`). > Current checkpoint: Phase 1 extraction refactor is complete (`docs/PHASE1_HANDOFF.md`). ## Status overview | Phase | Scope | Status | |---|---|---| | Phase 0 | Needle/RustFS spikes | Done (`docs/SPIKE_PHASE0.md`) | | Phase 1 | Extraction layer refactor | Done (`docs/PHASE1_HANDOFF.md`) | | Phase 2 | Prescreen stage | **Done** (`worker_prescreen/`, migrations `0006_prescreen.py` … `0008_add_manual_review_status.py`) | | Phase 3 | Storage modernization / RustFS watch | **Planned below** | | Follow-up | Admin/web, payments, heavy OCR | **Backlog** | ## Phase 2 — Prescreen stage between extract and analyze ### Why Insert a fast, local decision layer after extraction and before the expensive LLM analysis so that: - obvious low-risk contracts can short-circuit to a lightweight report - high-risk / high-value contracts are routed to deep LLM analysis - the heavy LLM worker is no longer the only path for every document - per-stage queue metrics, retry, and DLQ are clean and independent ### Key correction from the original prescreen spec The original spec assumed **Needle 2** (`cactus-needle`) as the prescreen extractor. The Phase 0 spike proved the base model is English-only and fails on Russian contracts with `confidence=0.0`. Fine-tuning would also disable the calibrated confidence head the routing design depends on. Therefore Phase 2 uses a **deterministic regex+pydantic extractor** for RU/BY contracts. Contract boilerplate is highly templated, so regex gives: - zero marginal cost - no hallucination risk - missing fields naturally map to `manual_review` - `confidence_score` becomes a deterministic **field-coverage ratio** A small-LLM fallback can be added later for fields regex misses. ### 2.1 RabbitMQ topology changes Add to `core/mq/topology.py` using the existing direct-exchange + TTL retry/DLQ pattern: ```text contracts.x ├─ extract ─► extract.q (existing) ├─ prescreen ─► prescreen.q (NEW) └─ analyze ─► analyze.q (existing) contracts.retry.x ├─ retry.prescreen ─► prescreen.retry.q (NEW classic delay queue, DLX→contracts.x[prescreen]) prescreen.dlq (NEW quorum) ``` Constants: `QUEUE_PRESCREEN`, `RK_PRESCREEN`, update `_RETRY_QUEUE_FOR`, `DLQ_FOR`, `declare_all()`. ### 2.2 Database migration `0006_prescreen.py` New table `prescreen_results`: ```sql 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, -- extracted metadata contract_type VARCHAR(64), party_a TEXT, party_b TEXT, total_amount DECIMAL(18, 2), currency VARCHAR(8), start_date DATE, end_date DATE, has_penalty_clause BOOLEAN, has_termination_clause BOOLEAN, has_arbitration BOOLEAN, confidence_score DECIMAL(4, 3), -- coverage ratio 0.000–1.000 -- routing routing_decision VARCHAR(32) NOT NULL, -- auto_approve | manual_review | deep_analysis -- metrics / audit prescreened_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), processing_ms INTEGER, extractor_version VARCHAR(32) DEFAULT 'regex-v1', -- auto-approve lightweight output auto_summary TEXT, auto_findings JSONB DEFAULT '[]', -- retry tracking error_message TEXT, retry_count INTEGER DEFAULT 0 ); CREATE INDEX idx_prescreen_document ON prescreen_results(document_id); CREATE INDEX idx_prescreen_routing ON prescreen_results(routing_decision); CREATE INDEX idx_prescreen_confidence ON prescreen_results(confidence_score); ``` Also: ```sql ALTER TYPE document_status ADD VALUE 'prescreening' AFTER 'extracted'; ALTER TYPE jobs.queue ADD VALUE 'prescreen' AFTER 'analyze'; ALTER TABLE reports ADD COLUMN prescreen_result_id UUID REFERENCES prescreen_results(id), ADD COLUMN prescreen_meta JSONB DEFAULT NULL; ``` Use `TEXT+CHECK` instead of Postgres enums if the project convention prefers additive migrations (the existing schema uses both). Decision to make during implementation. ### 2.3 Message schema additions Add to `core/mq/messages.py`: ```python class PrescreenRequested(PipelineMessage): """worker-extract → contracts.x[prescreen] → worker-prescreen.""" text_s3_key: str filename: str class PrescreenCompleted(PipelineMessage): """worker-prescreen result; also the payload forwarded to analyze.q.""" text_s3_key: str prescreened_at: datetime contract_type: str | None party_a: str | None party_b: str | None total_amount: float | None currency: str | None start_date: str | None end_date: str | None has_penalty_clause: bool | None has_termination_clause: bool | None has_arbitration: bool | None confidence_score: float | None # 0..1 coverage ratio routing_decision: Literal["auto_approve", "manual_review", "deep_analysis"] auto_summary: str | None auto_findings: list[dict] = Field(default_factory=list) error_message: str | None class AnalyzeRequested(PipelineMessage): """worker-prescreen → contracts.x[analyze] → worker-analyze.""" text_s3_key: str filename: str prescreen_meta: PrescreenCompleted ``` Keep `DocumentExtracted` for backward compatibility; `worker-extract` may still publish it for observability, but the prescreen stage is the main forward path. ### 2.4 New service: `worker-prescreen` Directory: ```text src/contract_check/worker_prescreen/ __init__.py __main__.py consumer.py handler.py router.py extractor.py # regex+pydantic RU/BY contract extractor config.py ``` #### `extractor.py` Deterministic extractor built with `re` + Pydantic model. Target fields: ```python class PrescreenContractMeta(BaseModel): contract_type: str | None party_a: str | None party_b: str | None total_amount: float | None currency: str | None start_date: str | None end_date: str | None has_penalty_clause: bool | None has_termination_clause: bool | None has_arbitration: bool | None ``` Heuristics: - `contract_type`: map first matched keyword (`договор поставки`, `договор оказания услуг`, `договор подряда`, `договор аренды`, `договор купли-продажи`, etc.) - `party_a` / `party_b`: extract first two legal entities after patterns like `Общество с ограниченной ответственностью «(.*?)»`, `Акционерное общество «(.*?)»`, `Индивидуальный предприниматель (.*?)`. - `total_amount`: find `составляет ([\d\s,.]+) (рублей|руб|USD|EUR|€|\$)` and parse as float. - `currency`: normalize to `RUB`, `USD`, `EUR`, `BYN`. - `start_date` / `end_date`: match `с (\d{2}\.\d{2}\.\d{4})`, `от (\d{2}\.\d{2}\.\d{4})`, `действует с (\d{2}\.\d{2}\.\d{4}) по (\d{2}\.\d{2}\.\d{4})`. - `has_penalty_clause`: presence of `неустойка`, `штраф`, `пеня`, `0,1%`/`% за каждый день просрочки`. - `has_termination_clause`: presence of `расторгнуть`, `расторжение`, `одностороннему порядке`. - `has_arbitration`: presence of `арбитражный суд`, `Арбитражный суд`, `Международный коммерческий арбитражный суд`. `confidence_score = matched_fields / total_fields`. Test on synthetic and a few real contracts; tune false-positive/negative rate. #### `router.py` ```python class PrescreenRouter: def __init__( self, confidence_threshold: float = 0.75, high_value_threshold: float = 100_000, ) -> None: ... def decide(self, meta: PrescreenContractMeta) -> str: if confidence < threshold or missing mandatory fields: return "manual_review" if total_amount > high_value_threshold: return "deep_analysis" if has_penalty_clause or has_arbitration: return "deep_analysis" return "auto_approve" ``` `auto_approve` is **disabled by default** via env `PRESCREEN_AUTO_APPROVE=false` until accuracy is proven; when disabled, `auto_approve` decisions are mapped to `manual_review`. #### `handler.py` - idempotency check on `documents.status` - set `status = 'prescreening'`, `jobs.status = 'running'` - download extracted Markdown from MinIO - run regex extractor in thread pool (`asyncio.to_thread`) - route - persist `prescreen_results` - publish next message: - `deep_analysis` → `contracts.x[analyze]` with `AnalyzeRequested` - `auto_approve` → `contracts.x[report.completed]` with lightweight summary (only when `PRESCREEN_AUTO_APPROVE=true`) - `manual_review` → `contracts.x[review.manual]`; no further processing - ack #### `consumer.py` `Consumer[PrescreenRequested]` for `prescreen.q`, wired like `AnalyzeConsumer`. #### `config.py` - `PRESCREEN_ENABLED` — if false, handler immediately publishes `AnalyzeRequested` (passthrough, preserving analytics) - `PRESCREEN_AUTO_APPROVE` - `PRESCREEN_CONFIDENCE_THRESHOLD` - `PRESCREEN_HIGH_VALUE_THRESHOLD` ### 2.5 Dockerfile + compose - `srv/worker-prescreen/Dockerfile`: lean image, no tesseract/pymupdf, only core deps + regex engine. - Add `worker-prescreen` to `docker-compose.yml` profile `services`. - Memory limit 128M (regex is tiny). ### 2.6 Changes to existing services #### `worker-extract` After publishing `DocumentExtracted` (backward compat), also publish `PrescreenRequested`: ```python await publisher.publish( PrescreenRequested( correlation_id=..., document_id=..., user_id=..., text_s3_key=extracted_key, filename=payload.filename, ), routing_key=RK_PRESCREEN, ) ``` Gate behind `PRESCREEN_ENABLED` (default true after Phase 2). #### `worker-analyze` Change consumer to `AnalyzeRequested`: ```python class AnalyzeConsumer(Consumer[AnalyzeRequested]): queue = "analyze.q" message_model = AnalyzeRequested ``` Enrich prompt with `prescreen_meta` (§6.3 of the original spec). #### `api` - Update document status endpoint to include prescreen block. - Status lifecycle: `queued → extracting → prescreening → analyzing | manual_review | done`. - Optional: expose `?mode=fast` vs `?mode=full` for B2B. ### 2.7 Metrics Add to `core/metrics.py` and expose from `worker-prescreen`: - `prescreen_requests_total{source}` - `prescreen_duration_seconds{decision}` - `prescreen_routing_decisions_total{decision,contract_type}` - `prescreen_confidence_distribution` - `prescreen_errors_total{error_type}` - `prescreen_dlq_messages_total` ### 2.8 Tests - Unit: `tests/unit/test_prescreen_extractor.py` — regex patterns on synthetic contracts, coverage scoring. - Unit: `tests/unit/test_prescreen_router.py` — routing matrix with all flag combinations. - Integration: `tests/integration/test_prescreen_worker.py`: - `PrescreenRequested` → `AnalyzeRequested` for `deep_analysis` - `PrescreenRequested` → `review.manual` for `manual_review` - bypass mode (`PRESCREEN_ENABLED=false`) → passthrough `AnalyzeRequested` - Update `test_extract_worker.py` to assert `PrescreenRequested` is also published. - Update `test_analyze_worker.py` to consume `AnalyzeRequested` and include prescreen metadata. ### 2.9 DoD ```bash make lint && make typecheck && make test-unit make test-integration ``` ## Phase 3 — Storage watch / RustFS evaluation ### Goal Keep MinIO as the production default but maintain a credible migration path to RustFS when its Lifecycle + KMS features reach GA. No application code changes are required because S3 is already behind `core.s3.port.py`. ### Why not switch now From the Phase 0 RustFS spike: - Lifecycle/KMS/distributed mode are marked **"Under Testing"** upstream. - SSE-S3 objects written by MinIO are **not readable by RustFS** today — a migration would require re-putting objects through the app. - RustFS is weeks old; MinIO is battle-tested for this exact compose setup. - Real advantages exist: Apache 2.0 license (vs MinIO AGPL), no telemetry, RF data-sovereignty friendly. ### Actions 1. Add an optional `docker-compose.rustfs.yml` overlay for local experiments. 2. Run the full integration suite against RustFS quarterly or when upstream announces Lifecycle/KMS GA. 3. If/when adopted: - pin exact RustFS image tag - set `S3_SERVER_SIDE_ENCRYPTION=false` during transition - re-encrypt objects via the app after cutover - update `minio-init` service or replace with `rustfs-init` ## Backlog / follow-up work (not tied to Phase 2) | Item | Rationale | |---|---| | Admin/web SPA | Deferred by original architecture scope. | | ЮKassa payments | Stub `invoices` table exists; payment logic deferred. | | Heavy OCR adapters (marker, paddleocr, easyocr, img2table) | Spec deferral. Consider separate `worker-extract-heavy` queue or cloud API. | | Email extractor (MSG/EML) | Spec deferral; only relevant for B2B corporate email ingestion. | | Multi-tenancy | Explicit non-goal of original refactor. Do not add `tenant_id` until orgs are a real requirement. | | Fine-tuned RU Needle model | Revisit if `cactus-needle` ships a multilingual or RU-tuned base with calibrated confidence. | ## Appendix: original spec alignment ### `document-extraction-spec.md` | Requirement | Phase 1 status | |---|---| | `DocumentExtractor` Protocol / ABC | Done (`core/extraction/port.py`) | | `ExtractedDocument` dataclass / Pydantic | Done | | `ExtractorFactory` with MIME detection | Done | | `DocxExtractor` (`mammoth`) | Done | | `RtfExtractor` (`striprtf`) | Done | | `EncodingDetector` (`chardet`) | Done | | `PyMuPDFExtractor` + `has_tables` | Done | | `TesseractOCRAdapter` + `is_structured=False` | Done | | Metrics by format | Done | | Update architecture docs | Done | | Heavy adapters deferred | Backlog | ### `needle-prescreen-integration.md` | Requirement | Phase 2 plan above | |---|---| | Separate `prescreen.q` stage | Yes | | `PrescreenRequested` / `PrescreenCompleted` messages | Yes, corrected to extend `PipelineMessage` | | `prescreen_results` table | Yes, adjusted columns | | `worker-prescreen` service | Yes, regex-based instead of Needle | | Routing decisions | Yes, conservative default | | `worker-extract` publishes prescreen | Yes | | `worker-analyze` consumes `AnalyzeRequested` | Yes | | API status endpoint enriched | Yes | | Dockerfile/compose for prescreen worker | Yes | | Rollback / bypass flag | Yes | | Multi-tenant / credit split | **Rejected** — not in current scope; integer credit model stays | ## Ordering recommendation 1. Phase 2 DB migration + RabbitMQ topology. 2. Phase 2 messages + `worker_prescreen` skeleton (no-op passthrough first). 3. Phase 2 regex extractor + router + tests. 4. Wire `worker-extract` → `prescreen.q` and `worker-analyze` → `analyze.q`. 5. End-to-end integration test. 6. Phase 3 RustFS spike only after upstream GA announcement. --- **Next action:** implement Phase 2 step 1 (migration + topology) if approved.