doc intelligence

# valoswiss-doc-intelligence (33°)

**Macro-categoria**: 📄 DOMINI SINGOLI (entry Wave 5)
**Scope**: OCR + table extraction + entity recognition da documenti finanziari/legali (PDF, scansioni TIFF, foto smartphone) tramite Docling Python sidecar + fallback LlamaParse cloud.
**Born**: 2026-05-03 (W1 sidecar Python + W2 NestJS module + W3 frontend admin queue + W4 vault auto-trigger + W5 entity recognition specialized + Persona pack DOC-OPS)
**Owner downstream**: ADVISOR (lettura output) · SUPERVISOR/ADMIN (queue + retry + cost ledger) · DOC-OPS (operatore back-office triage qualità)
**Last aligned**: 2026-05-03 V20

---

## §0 · Pre-flight check (entry rituale dell'agente)

Prima di ogni intervento, verifica in quest'ordine:

1. **Branch + working tree**
   ```bash
   cd ~/git/valoswiss && git status --short && git log -3 --oneline
   ```
2. **Sidecar Python health**
   ```bash
   curl -s http://127.0.0.1:8891/healthz | jq .
   ```
   Deve ritornare `{"status":"ok","version":"...","engine":"docling","useCloudFallback":true|false}`. Se 502/connection refused → sidecar PM2 down: `pm2 list | grep doc-intel-py`.
3. **NestJS proxy health**
   ```bash
   curl -s http://127.0.0.1:4010/api/doc-intel/health -H "Cookie: valo_token=<dev-token>"
   ```
   Deve ritornare `{ sidecar:{status:'ok'}, circuitBreaker:{state:'closed', failures:0}, queueDepth:N }`.
4. **Prisma schema sync**
   ```bash
   cd apps/api && npx prisma migrate status
   ```
   Verifica che la model `DocumentParse` + enum `DocumentParseStatus` + enum `DocumentParseEngine` siano applicati (idempotent migration V15-doc-intel).
5. **Tenant configs**: `tenants/ws.json` e `tenants/az.json` devono avere `"docIntelligence": true` subito dopo `documentVault`.
6. **Persona pack**: `apps/api/src/common/persona-packs/persona-packs.constants.ts` deve avere `'docIntelligence'` in `defaultModules` per `ADVISOR` + `RELATIONSHIP_MANAGER` + `SUPERVISOR` + `ADMIN` (NON in PROSPECT/RETAIL_CLIENT/AFFLUENT_CLIENT/UHNW_CLIENT/FAMILY_OFFICE_PRINCIPAL — PII privacy).
7. **Module registry**: `apps/web/src/lib/module-registry.ts` deve esporre entry `docIntelligence` con `sidebarSection: 'OPERARE'` (sotto OPERATIVITÀ DOCUMENTALE), `requiredRole: 'ADVISOR'`, `personaHint: 'documental'`, icon `📄`.
8. **R-Audit gate**: prima di qualsiasi commit su file CRITICAL (vedi §6.1), eseguire `npx tsx scripts/r-audit.ts <file> --validate-business-logic`.

Se uno qualunque dei 7 punti fallisce, **fermati e annota la deviazione** prima di procedere — la 3-Point Registration V16 è invariante non negoziabile (vedi `feedback_new_module_registration.md`).

---

## §1 · Aree di competenza

### 1.1 Pipeline parsing 4-stage
1. **Stage 1 — Ingest**: NestJS riceve `documentVaultId` o upload diretto, calcola SHA256 fingerprint, persiste `DocumentParse` con status `PENDING`, enqueue async job al sidecar Python.
2. **Stage 2 — Layout analysis** (Docling primary): `services/doc-intel-py` chiama Docling pipeline → DLNet layout segmentation + TableFormer table structure recognition (97.9% accuracy IBM Research benchmark). Output `DoclingDocument` JSON intermediate.
3. **Stage 3 — Entity recognition**: extractor specializzati per categoria documento:
   - `bank-statement` → KV pairs (IBAN, BIC, periodo, balance) + line items table
   - `invoice` → header KV (numero, data, partita IVA, totale, IVA) + righe articoli
   - `kyc-passport` → MRZ parser ICAO 9303 + foto crop bounding box
   - `contract` → clausole semantic chunking + parties + dates + signatures detection
   - `portfolio-statement` → holdings table normalization (ISIN/quantity/price/value)
4. **Stage 4 — Output structured JSON**: persiste in `DocumentParse.outputJson` (JSONB Postgres) + bounding box per ogni field con `confidence ∈ [0,1]` + chiamata callback NestJS `POST /admin/doc-intel/internal/persist-result` loopback-only.

### 1.2 Engine selection logic
| Trigger | Engine selezionato | Modello |
|---|---|---|
| PDF nativo testo (estraibile) | `docling-fast` | layout-only DLNet |
| PDF scansione + tabelle complesse | `docling-vlm` | SmolDocling-256M VLM |
| Immagine (JPEG/PNG/HEIC) | `docling-vlm` | SmolDocling-256M VLM |
| TIFF multi-page bancario | `docling-tableformer` | TableFormer transformer |
| Cloud burst (sidecar down OR file >50MB) | `llama-parse` | LlamaIndex API premium |
| Multi-parser ensemble (audit critical) | `ensemble` | docling + llama-parse + tesseract → vote |

Override env: `DOC_INTEL_ENGINE_DEFAULT` (default `docling-fast`), `DOC_INTEL_ENABLE_CLOUD_FALLBACK` (default `1`).

### 1.3 Output structured JSON (esempio bank statement)
```jsonc
{
  "documentParseId": "uuid",
  "documentVaultId": "uuid",
  "engine": "docling-fast",
  "version": "docling-2.x.y",
  "category": "bank-statement",
  "tenantSlug": "ws",
  "pages": 4,
  "headerKv": {
    "iban":   { "value": "CH93 0076 2011 6238 5295 7", "page": 1, "bbox": [120, 80, 480, 110], "confidence": 0.98 },
    "period": { "value": "2026-04-01..2026-04-30",     "page": 1, "bbox": [120, 130, 380, 160], "confidence": 0.95 }
  },
  "tables": [
    {
      "page": 2,
      "title": "Movimenti del periodo",
      "headers": ["Data", "Descrizione", "Addebiti", "Accrediti", "Saldo"],
      "rows": [
        ["2026-04-02", "Bonifico SEPA", "-1500.00", null, "23450.20"],
        ["2026-04-05", "Stipendio",     null,       "3800.00", "27250.20"]
      ],
      "cellsBbox": [...],
      "tableConfidence": 0.92
    }
  ],
  "entities": {
    "ibans":   [{ "value": "CH93...", "kind": "primary",  "confidence": 0.98 }],
    "amounts": [{ "value": 1500.00, "currency": "CHF", "kind": "debit",  "page": 2 }],
    "dates":   [{ "value": "2026-04-02", "kind": "transaction", "page": 2 }]
  },
  "qualityScore": 0.91,
  "warnings": []
}
```

### 1.4 Persona visibility
- **DOC-OPS** (ws+az): operatore back-office triage qualità + retry con altro engine + manual KV correction
- **ADVISOR** / **RELATIONSHIP_MANAGER**: lettura output structured (read-only) per docs cliente-scoped
- **SUPERVISOR/ADMIN**: cross-tenant + queue stats + cost ledger doc-intel + retry forzato + engine override
- **CLIENT/PROSPECT/RETAIL_CLIENT/AFFLUENT_CLIENT/UHNW_CLIENT/FAMILY_OFFICE_PRINCIPAL**: NEGATO assoluto — output OCR contiene PII grezza (IBAN, MRZ, etc.) gestione solo dietro KYC vault PII pipeline

### 1.5 Multi-tenant ws+az
- Tenant scoping via `tenantSlug` su ogni job + RLS Postgres su `DocumentParse`
- Queue partitioning weighted: ws 60% / az 40% baseline (override SUPERVISOR via admin endpoint)
- Cost ledger per tenant via callback `record-external` riusando endpoint TradingAgents-style

---

## §2 · Pattern di codice (riferimenti architetturali)

### 2.1 Python sidecar — `services/doc-intel-py/app.py`
```python
from __future__ import annotations
from contextlib import asynccontextmanager
from fastapi import FastAPI, BackgroundTasks
from .runner import run_parse_async, get_job_status, get_health
from .contracts import ParseRequest, ParseResponse, JobStatus

@asynccontextmanager
async def lifespan(app: FastAPI):
    # warmup Docling pipeline lazy (evita 2GB import time se non usato)
    app.state.warmed = False
    yield

app = FastAPI(title="valoswiss-doc-intel", version="1.0.0", lifespan=lifespan)

@app.get("/healthz")
async def healthz() -> dict:
    return get_health()

@app.post("/parse", response_model=ParseResponse)
async def parse_sync(req: ParseRequest) -> ParseResponse:
    """Sync 5-30s — usato per file piccoli native PDF text."""
    return await run_parse_async(req, sync=True)

@app.post("/parse-async")
async def parse_async(req: ParseRequest, bg: BackgroundTasks) -> dict:
    """Async — ritorna job_id immediato, advisor poll status."""
    job_id = await run_parse_async(req, sync=False, bg=bg)
    return {"jobId": job_id, "status": "PENDING"}

@app.get("/jobs/{job_id}")
async def jobs(job_id: str) -> JobStatus:
    return get_job_status(job_id)
```

### 2.2 Docling client wrapper — `services/doc-intel-py/docling_client.py`
```python
from __future__ import annotations
from pathlib import Path
import subprocess, json, tempfile

# Lazy import: docling pesa ~2GB con modelli VLM caricati
_DOCLING_AVAILABLE = None
_DOCLING_CONVERTER = None

def _ensure_docling():
    global _DOCLING_AVAILABLE, _DOCLING_CONVERTER
    if _DOCLING_AVAILABLE is None:
        try:
            from docling.document_converter import DocumentConverter
            from docling.datamodel.pipeline_options import PdfPipelineOptions
            opts = PdfPipelineOptions(do_ocr=True, do_table_structure=True)
            _DOCLING_CONVERTER = DocumentConverter(format_options={"pdf": opts})
            _DOCLING_AVAILABLE = True
        except ImportError:
            _DOCLING_AVAILABLE = False
    return _DOCLING_AVAILABLE

async def parse_with_docling(file_path: Path, engine: str) -> dict:
    if not _ensure_docling():
        raise RuntimeError("docling not installed; pip install -r requirements-real.txt")
    result = _DOCLING_CONVERTER.convert(str(file_path))
    return {
        "engine": engine,
        "version": "docling-2.x",
        "pages": len(result.document.pages),
        "markdown": result.document.export_to_markdown(),
        "tables": [t.export_to_dict() for t in result.document.tables],
        "rawJson": result.document.export_to_dict(),
    }
```

### 2.3 LlamaParse fallback cloud — `services/doc-intel-py/llamaparse_client.py`
```python
from __future__ import annotations
import os, httpx
from pathlib import Path

LLAMA_API_KEY = os.getenv("LLAMA_PARSE_API_KEY")
LLAMA_BASE = "https://api.cloud.llamaindex.ai/api/parsing"

async def parse_with_llamaparse(file_path: Path) -> dict:
    if not LLAMA_API_KEY:
        raise RuntimeError("LLAMA_PARSE_API_KEY non configurata; fallback cloud disabilitato")
    async with httpx.AsyncClient(timeout=120.0) as client:
        with open(file_path, "rb") as f:
            up = await client.post(
                f"{LLAMA_BASE}/upload",
                headers={"Authorization": f"Bearer {LLAMA_API_KEY}"},
                files={"file": (file_path.name, f, "application/pdf")},
            )
        up.raise_for_status()
        job_id = up.json()["id"]
        # poll fino a 90s
        for _ in range(45):
            r = await client.get(
                f"{LLAMA_BASE}/job/{job_id}",
                headers={"Authorization": f"Bearer {LLAMA_API_KEY}"},
            )
            if r.json().get("status") == "SUCCESS":
                break
        out = await client.get(
            f"{LLAMA_BASE}/job/{job_id}/result/markdown",
            headers={"Authorization": f"Bearer {LLAMA_API_KEY}"},
        )
        return {"engine": "llama-parse", "markdown": out.json().get("markdown")}
```

### 2.4 NestJS sidecar wrapper — `apps/api/src/modules/doc-intelligence/services/docling-client.service.ts`
```typescript
import { Injectable, Logger, Optional } from '@nestjs/common';
import axios, { AxiosInstance } from 'axios';

@Injectable()
export class DoclingClientService {
  private readonly logger = new Logger(DoclingClientService.name);
  private readonly http: AxiosInstance;
  private failures = 0;
  private circuitOpenedAt: number | null = null;
  private readonly CIRCUIT_THRESHOLD = 3;
  private readonly CIRCUIT_OPEN_MS = 30_000;

  constructor() {
    this.http = axios.create({
      baseURL: process.env.DOC_INTEL_SIDECAR_URL ?? 'http://127.0.0.1:8891',
      timeout: 120_000,
    });
  }

  private isCircuitOpen(): boolean {
    if (this.circuitOpenedAt === null) return false;
    if (Date.now() - this.circuitOpenedAt > this.CIRCUIT_OPEN_MS) {
      this.circuitOpenedAt = null; // half-open
      this.failures = 0;
      return false;
    }
    return true;
  }

  async parseAsync(payload: { documentVaultId: string; tenantSlug: string; engine?: string }): Promise<{ jobId: string }> {
    if (this.isCircuitOpen()) {
      throw new Error('docling-sidecar circuit open; retry after 30s');
    }
    try {
      const { data } = await this.http.post('/parse-async

…[truncato — apri il file MD per testo completo]