DealDocumentScreening/docs/PHASES_2_PLUS_ROADMAP.md

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	Planned below
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:

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:

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:

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:

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:

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:

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

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:

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:

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

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.