Automated deployment from GitHub Actions: d0b87cbe4fdaf86c5c12e61d54b1acd8b234b76c

pyproject.toml

@@ -36,6 +36,7 @@ dependencies = [
     # AI / LLM
     "google-generativeai>=0.8.5",
     "langchain-google-genai>=2.1.0",
+    "langchain-openai>=0.2.14",
     "langgraph>=0.3.18",
     "langgraph-checkpoint-redis>=0.0.6",
     "langsmith>=0.3.11",

src/ai/graph.py

@@ -32,13 +32,6 @@ log = structlog.get_logger()
 def _needs_fallback(state: ExtractionGraphState) -> str:
     """Route to OCR ensemble fallback when quality, confidence, or data is weak."""
     if settings.CLOUD_LLM_ONLY:
-        for doc in state.get("documents", []):
-            if doc.get("error") or not doc.get("extracted_data"):
-                return "fallback"
-            if len(doc.get("ocr_candidates") or {}) > 1:
-                return "fallback"
-            if doc.get("ocr_conflicts"):
-                return "fallback"
         log.info("CLOUD_LLM_ONLY is active — bypassing heavy OCR ensemble fallback")
         return "validate"
         
@@ -52,7 +45,7 @@ def _needs_fallback(state: ExtractionGraphState) -> str:
             return "fallback"
         if doc.get("ocr_conflicts"):
             return "fallback"
-        if len(doc.get("ocr_candidates") or {}) > 1:
+        if len(doc.get("ocr_candidates") or {}) > 1 and doc.get("document_mode") != "digital_pdf_text":
             return "fallback"
     return "validate"
 

src/ai/nodes/extract.py

@@ -1,11 +1,16 @@
 """
 TradeFlow AI — Primary LLM Extraction Node (Step 2.2)
 
-Uses Gemini 2.0 Flash Exp for multimodal extraction.
+Uses Gemini 2.0 Flash Exp for multimodal extraction, or a local Ollama LLM
+when USE_LOCAL_LLM=true.
 """
 
+from __future__ import annotations
+
 import asyncio
 import inspect
+import json
+import re
 
 import structlog
 from pydantic import BaseModel, Field
@@ -30,147 +35,360 @@ else:
 
 log = structlog.get_logger()
 
-# Structured output schema
+
+# Structured output schema — comprehensive CEISA + B/L fields
 class CEISAFields(BaseModel):
-    importer_name: str | None = Field(description="Name of the importing company")
-    importer_npwp: str | None = Field(description="NPWP tax ID of the importer")
-    total_packages: int | None = Field(description="Total number of packages/koli")
-    gross_weight: float | None = Field(description="Total gross weight in KGM")
+    # Importer / Consignee
+    importer_name: str | None = Field(description="Name of importing company (consignee)")
+    importer_npwp: str | None = Field(description="NPWP tax ID, 15-16 digits, explicitly labeled NPWP")
+    importer_address: str | None = Field(description="Address of importer/consignee")
+    # Shipper / Exporter
+    exporter_name: str | None = Field(description="Name of exporting company (shipper)")
+    exporter_address: str | None = Field(description="Address of exporter/shipper")
+    # B/L and document references
+    bl_number: str | None = Field(description="Bill of Lading number")
+    bl_date: str | None = Field(description="Date of B/L issue")
+    # Vessel and voyage
+    vessel_name: str | None = Field(description="Name of the ocean vessel")
+    voyage_number: str | None = Field(description="Voyage number")
+    # Ports
+    port_of_loading: str | None = Field(description="Port of loading (departure)")
+    port_of_discharge: str | None = Field(description="Port of discharge (destination)")
+    # Cargo
+    total_packages: int | None = Field(description="Total number of packages/koli across ALL containers")
+    gross_weight: float | None = Field(description="Total gross weight in KGS/KGM")
+    # Container numbers (as a comma-separated string)
+    container_numbers: str | None = Field(description="Container numbers, comma-separated")
+    description_of_goods: str | None = Field(description="General description of goods")
+    hs_code: str | None = Field(description="HS/BTKI tariff code exactly as printed, do not pad or correct")
+    # Commercial values (usually from Invoice, may be absent in B/L)
     cif_value: float | None = Field(description="Total CIF value")
+    fob_value: float | None = Field(description="Total FOB value")
+    freight_value: float | None = Field(description="Freight value")
+    insurance_value: float | None = Field(description="Insurance value")
     currency: str | None = Field(description="Currency code (e.g. USD, IDR)")
+    importer_nib: str | None = Field(description="Importer NIB business ID exactly as printed")
+    # Incoterms
+    incoterms: str | None = Field(description="Incoterms (e.g. FOB, CIF, CFR)")
+    freight_terms: str | None = Field(description="Freight terms (PREPAID or COLLECT)")
+
+
+def _parse_json_from_text(text: str) -> dict:
+    """
+    Robustly extract a JSON object from LLM plain-text output.
+    Handles markdown code fences and DeepSeek-style <think> tags.
+    """
+    # Strip <think>...</think> tags (DeepSeek-R1 style)
+    text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
+
+    # Try JSON inside markdown fences first
+    fence_match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+    if fence_match:
+        try:
+            return json.loads(fence_match.group(1))
+        except json.JSONDecodeError:
+            pass
+
+    # Fall back to bare JSON object
+    json_match = re.search(r"\{.*\}", text, re.DOTALL)
+    if json_match:
+        try:
+            return json.loads(json_match.group(0))
+        except json.JSONDecodeError:
+            pass
+
+    return {}
+
+
+def _normalize_for_evidence(value: object) -> str:
+    return re.sub(r"[^a-z0-9]+", "", str(value or "").casefold())
+
+
+def _field_value_has_text_evidence(field: str, value: object, raw_text: str) -> bool:
+    normalized_value = _normalize_for_evidence(value)
+    normalized_text = _normalize_for_evidence(raw_text)
+    if not normalized_value:
+        return False
+    if normalized_value in normalized_text:
+        return True
+    if field in {"gross_weight", "cif_value", "fob_value", "freight_value", "insurance_value"}:
+        numeric = re.sub(r"[^0-9]", "", str(value))
+        return bool(numeric and numeric in normalized_text)
+    if field == "total_packages":
+        numeric = re.sub(r"[^0-9]", "", str(value))
+        return bool(numeric and numeric in normalized_text)
+    return False
+
+
+def _field_format_valid(field: str, value: object) -> bool:
+    text = str(value or "").strip()
+    if not text:
+        return False
+    if field == "importer_npwp":
+        return len(re.sub(r"\D", "", text)) in {15, 16}
+    if field == "importer_nib":
+        return len(re.sub(r"\D", "", text)) == 13
+    if field == "hs_code":
+        return bool(re.fullmatch(r"\d{8}", text))
+    if field == "currency":
+        return bool(re.fullmatch(r"[A-Z]{3}", text))
+    if field in {"gross_weight", "cif_value", "fob_value", "freight_value", "insurance_value"}:
+        try:
+            return float(str(value).replace(",", "")) >= 0
+        except (TypeError, ValueError):
+            return False
+    if field == "total_packages":
+        try:
+            return int(float(str(value).replace(",", ""))) > 0
+        except (TypeError, ValueError):
+            return False
+    return True
+
+
+def _estimate_field_confidences(extracted: dict, doc: dict) -> dict[str, float]:
+    raw_text = doc.get("raw_text") or ""
+    candidates = doc.get("ocr_candidates") or {}
+    pdf_candidate = candidates.get("pdf_text") or {}
+    base = 0.88 if doc.get("document_mode") == "digital_pdf_text" else 0.82
+    if pdf_candidate.get("confidence"):
+        base = max(base, min(0.96, float(pdf_candidate.get("confidence")) * 0.94))
+
+    confidences: dict[str, float] = {}
+    for field, value in extracted.items():
+        confidence = base
+        has_evidence = _field_value_has_text_evidence(field, value, raw_text)
+        format_valid = _field_format_valid(field, value)
+        if has_evidence:
+            confidence += 0.05
+        else:
+            confidence -= 0.12
+        if not format_valid:
+            confidence -= 0.25
+        confidences[field] = round(max(0.35, min(0.99, confidence)), 4)
+    return confidences
+
 
 async def llm_extraction_node(state: ExtractionGraphState) -> dict:
     """
-    Step 2.2: Primary LLM Extraction using Gemini 2.0 Flash Exp.
+    Step 2.2: Primary LLM Extraction.
 
-    Args:
-        state: ExtractionGraphState with documents list containing doc_id, storage_path, pages
+    - When USE_LOCAL_LLM=true: uses Ollama (text-only, manual JSON parsing).
+    - Otherwise: uses Gemini multimodal (with_structured_output).
 
     Returns:
-        dict with:
-            - documents: Updated docs with extracted_data or error flag
-            - combined_data: Merged field values across docs
-            - steps: Execution trace
-
-    Raises:
-        Specific exceptions (GoogleAPIError, ValueError) — does NOT catch all exceptions
+        dict with documents, combined_data, steps
     """
     log.info("Running llm_extraction_node", batch_id=state["batch_id"])
 
-    # Lazy LLM setup: only instantiate the LLM when we encounter the first document
-    # that actually needs LLM extraction (has pages). This keeps unit tests lightweight
-    # when LLM dependencies are not installed.
+    # LLM instances — lazily initialized on first document
+    llm = None
     structured_llm = None
+    use_manual_json = False  # True for Ollama (no function-calling)
 
     updated_docs = []
     combined_data = {}
 
     for doc in state["documents"]:
-        # Validate document state before processing
-        if not doc.get("doc_id") or not doc.get("pages"):
+        # ── Guard: document must have doc_id and pages ──────────────────────
+        has_extraction_input = bool(doc.get("pages")) or bool((doc.get("raw_text") or "").strip())
+        if not doc.get("doc_id") or not has_extraction_input:
             log.error(
                 "Invalid document state — missing required fields",
                 doc_id=doc.get("doc_id"),
-                batch_id=state["batch_id"]
+                batch_id=state["batch_id"],
             )
             updated_docs.append({
                 **doc,
-                "error": "Document missing required fields (doc_id, pages)",
+                "error": "Document missing required fields (doc_id and pages/raw_text)",
                 "fallback_required": True,
-                "ocr_method": "failed"
+                "ocr_method": "failed",
             })
             continue
 
-        # Initialize LLM on first real document that needs extraction
-        if structured_llm is None:
-            try:
-                if settings.DETERMINISTIC_E2E:
-                    if DeterministicLLM is None:
-                        raise RuntimeError("DETERMINISTIC_E2E enabled but DeterministicLLM not available")
-                    llm = DeterministicLLM()
+        # ── Initialize LLM once ─────────────────────────────────────────────
+        if llm is None:
+            if settings.DETERMINISTIC_E2E:
+                if DeterministicLLM is None:
+                    raise RuntimeError("DETERMINISTIC_E2E enabled but DeterministicLLM not available")
+                llm = DeterministicLLM()
+                structured_llm = llm.with_structured_output(CEISAFields)
+                use_manual_json = False
+
+            elif settings.USE_LOCAL_LLM:
+                try:
+                    from langchain_openai import ChatOpenAI
+                except ImportError:
+                    raise RuntimeError("Dependency 'langchain_openai' is required for local LLM support")
+
+                # Supports comma-separated models: "qwen2.5:7b,mistral:7b"
+                local_models = [m.strip() for m in settings.LOCAL_LLM_MODEL.split(",") if m.strip()]
+                if not local_models:
+                    local_models = ["qwen2.5:7b"]
+
+                primary_llm = ChatOpenAI(
+                    model=local_models[0],
+                    base_url=settings.OLLAMA_BASE_URL,
+                    api_key="ollama",
+                    temperature=0,
+                    max_retries=1,
+                )
+                log.info("Using primary local LLM", model=local_models[0])
+
+                if len(local_models) > 1:
+                    fallback_llms = [
+                        ChatOpenAI(
+                            model=m,
+                            base_url=settings.OLLAMA_BASE_URL,
+                            api_key="ollama",
+                            temperature=0,
+                            max_retries=1,
+                        )
+                        for m in local_models[1:]
+                    ]
+                    llm = primary_llm.with_fallbacks(fallback_llms)
+                    log.info("Configured local fallback LLMs", models=local_models[1:])
                 else:
-                    if ChatGoogleGenerativeAI is None:
-                        raise RuntimeError("Production LLM dependency 'langchain_google_genai' is not installed")
-                    primary_llm = ChatGoogleGenerativeAI(
-                        model=settings.GEMINI_MODEL_PRIMARY,
-                        temperature=0,
-                        api_key=settings.GEMINI_API_KEY
-                    )
-                    fallback_llm = ChatGoogleGenerativeAI(
-                        model=settings.GEMINI_MODEL_FALLBACK,
+                    llm = primary_llm
+
+                # Ollama does NOT support function-calling — parse JSON manually
+                use_manual_json = True
+
+            else:
+                # ── Gemini (multimodal, with_structured_output) ──────────────
+                if ChatGoogleGenerativeAI is None:
+                    raise RuntimeError("Production LLM dependency 'langchain_google_genai' is not installed")
+
+                primary_llm = ChatGoogleGenerativeAI(
+                    model=settings.GEMINI_MODEL_PRIMARY,
+                    temperature=0,
+                    api_key=settings.GEMINI_API_KEY,
+                )
+                fallback_llms = []
+                try:
+                    from langchain_openai import ChatOpenAI
+                    olm_llm = ChatOpenAI(
+                        model=settings.OLM_BASE_MODEL,
+                        base_url=f"{settings.OLM_INFERENCE_URL}/v1",
+                        api_key="empty",
                         temperature=0,
-                        api_key=settings.GEMINI_API_KEY
+                        max_retries=1,
                     )
-                    llm = primary_llm.with_fallbacks([fallback_llm])
+                    fallback_llms.append(olm_llm)
+                except Exception as e:
+                    log.warning("Could not setup OLM fallback", error=str(e))
 
+                gemini_fallback = ChatGoogleGenerativeAI(
+                    model=settings.GEMINI_MODEL_FALLBACK,
+                    temperature=0,
+                    api_key=settings.GEMINI_API_KEY,
+                )
+                fallback_llms.append(gemini_fallback)
+                llm = primary_llm.with_fallbacks(fallback_llms)
                 structured_llm = llm.with_structured_output(CEISAFields)
-                # Support both synchronous return and awaitable (coroutine/AsyncMock)
                 if asyncio.iscoroutine(structured_llm) or inspect.isawaitable(structured_llm):
                     structured_llm = await structured_llm
-            except Exception as e:
-                log.exception(
-                    "Gemini setup failed - preserving OCR candidates for reconciliation",
-                    doc_id=doc.get("doc_id"),
-                    batch_id=state["batch_id"],
-                    error_type=type(e).__name__,
-                )
-                updated_docs.append({
-                    **doc,
-                    "error": str(e),
-                    "fallback_required": True,
-                    "ocr_method": "gemini_setup_failed",
-                })
-                continue
-        # Validate document state before processing
-        if not doc.get("doc_id") or not doc.get("pages"):
-            log.error(
-                "Invalid document state — missing required fields",
-                doc_id=doc.get("doc_id"),
-                batch_id=state["batch_id"]
-            )
-            updated_docs.append({
-                **doc,
-                "error": "Document missing required fields (doc_id, pages)",
-                "fallback_required": True,
-                "ocr_method": "failed"
-            })
-            continue
+                use_manual_json = False
 
+        # ── Build prompt messages ───────────────────────────────────────────
         try:
-            # Avoid importing heavy langchain Core in deterministic/test mode
             if settings.DETERMINISTIC_E2E:
                 messages = [{"type": "text", "text": "deterministic"}]
             else:
                 try:
                     from langchain_core.messages import HumanMessage as _HumanMessage
                 except Exception:
-                    class _HumanMessage:  # lightweight fallback so tests don't require langchain_core
+                    class _HumanMessage:  # lightweight fallback
                         def __init__(self, content):
                             self.content = content
 
-                # Real LLM call with multimodal content
-                # Pass pages as base64-encoded images for extraction
-                messages = [
-                    _HumanMessage(
-                        content=[
-                            {"type": "text", "text": "Extract all CEISA fields (importer name, NPWP, packages, weight, CIF value) from this document."},
-                            {"type": "image_url", "image_url": {"url": doc["pages"][0]}} if doc["pages"] else {"type": "text", "text": "No pages available"}
-                        ]
+                if use_manual_json:
+                    # Text-only prompt for local Ollama models
+                    raw_text = doc.get("raw_text", "")
+                    content = (
+                        "You are a strictly accurate customs document parser for CEISA 4.0 (Indonesian Customs). "
+                        "Extract ALL the following fields from the document.\n"
+                        "CRITICAL RULES:\n"
+                        "1. If a value is NOT clearly present in the text, return null for that field. DO NOT GUESS.\n"
+                        "2. Return ONLY a valid JSON object. No explanation, no markdown.\n"
+                        "3. For gross_weight: remove commas used as thousand separators (e.g. '11,603.000' -> 11603.0).\n"
+                        "4. For total_packages: sum ALL container package counts (e.g. '20 PACKAGES' + '17 PACKAGES' = 37).\n"
+                        "5. For importer_npwp: ONLY extract if the text explicitly says 'NPWP' or 'Tax ID'. DO NOT use B/L numbers.\n\n"
+                        "Fields to extract (return as JSON keys):\n"
+                        "- importer_name: Consignee / buyer company name\n"
+                        "- importer_npwp: NPWP tax ID (15-16 digits, null if not found)\n"
+                        "- importer_address: Consignee/importer address\n"
+                        "- exporter_name: Shipper / seller company name\n"
+                        "- exporter_address: Shipper/exporter address\n"
+                        "- bl_number: Bill of Lading number\n"
+                        "- bl_date: B/L issue date (ISO 8601 if possible)\n"
+                        "- vessel_name: Ocean vessel name\n"
+                        "- voyage_number: Voyage number\n"
+                        "- port_of_loading: Port of departure\n"
+                        "- port_of_discharge: Port of destination\n"
+                        "- total_packages: TOTAL packages across ALL containers (integer)\n"
+                        "- gross_weight: Total gross weight in KGS as a plain float (no commas)\n"
+                        "- container_numbers: All container numbers comma-separated\n"
+                        "- description_of_goods: Brief description of cargo\n"
+                        "- hs_code: HS/BTKI code exactly as printed; do NOT pad/correct invalid 6-digit codes\n"
+                        "- cif_value: CIF value (float, null if not in document)\n"
+                        "- fob_value: FOB value (float, null if not in document)\n"
+                        "- freight_value: Freight value (float, null if not in document)\n"
+                        "- insurance_value: Insurance value (float, null if not in document)\n"
+                        "- currency: Currency code (USD/IDR/EUR etc, null if not found)\n"
+                        "- importer_nib: NIB exactly as printed, null if not found\n"
+                        "- incoterms: Incoterms code (FOB/CIF/CFR etc, null if not found)\n"
+                        "- freight_terms: PREPAID or COLLECT (null if not found)\n\n"
+                        f"Document Text:\n{raw_text[:12000]}"
                     )
-                ]
-
-            result = await structured_llm.ainvoke(messages)
+                    messages = [_HumanMessage(content=[{"type": "text", "text": content}])]
+                else:
+                    # Multimodal prompt for Gemini
+                    raw_text = (doc.get("raw_text") or "")[:12000]
+                    prompt_text = (
+                        "Extract all CEISA fields (importer name, NPWP, packages, weight, CIF value) from this document."
+                    )
+                    if raw_text:
+                        prompt_text += f"\n\nDirect PDF/OCR text:\n{raw_text}"
+                    messages = [
+                        _HumanMessage(
+                            content=[
+                                {
+                                    "type": "text",
+                                    "text": prompt_text,
+                                },
+                                (
+                                    {"type": "image_url", "image_url": {"url": doc["pages"][0]}}
+                                    if doc.get("pages")
+                                    else {"type": "text", "text": "No pages available"}
+                                ),
+                            ]
+                        )
+                    ]
 
-            # Convert Pydantic model to dict
-            raw_extracted = result.model_dump(exclude_none=True) if hasattr(result, 'model_dump') else result
-            if asyncio.iscoroutine(raw_extracted):
-                raw_extracted = await raw_extracted
-            extracted = raw_extracted
+            # ── Invoke LLM ──────────────────────────────────────────────────
+            if use_manual_json:
+                response = await llm.ainvoke(messages)
+                text_response = response.content if hasattr(response, "content") else str(response)
+                raw_extracted = _parse_json_from_text(text_response)
+                # Coerce through Pydantic for type safety
+                try:
+                    validated = CEISAFields(**raw_extracted)
+                    extracted = validated.model_dump(exclude_none=True)
+                except Exception:
+                    extracted = {k: v for k, v in raw_extracted.items() if v is not None}
+            else:
+                result = await structured_llm.ainvoke(messages)
+                raw_result = result.model_dump(exclude_none=True) if hasattr(result, "model_dump") else result
+                if asyncio.iscoroutine(raw_result):
+                    raw_result = await raw_result
+                extracted = raw_result
 
             candidates = dict(doc.get("ocr_candidates") or {})
+            field_confidences = _estimate_field_confidences(extracted, doc)
             candidates[settings.GEMINI_MODEL_PRIMARY] = {
                 "fields": extracted,
-                "confidence": 0.82,
+                "confidence": round(sum(field_confidences.values()) / len(field_confidences), 4) if field_confidences else 0.0,
+                "field_confidences": field_confidences,
             }
 
             updated_docs.append({
@@ -178,44 +396,28 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
                 "extracted_data": extracted,
                 "ocr_method": settings.GEMINI_MODEL_PRIMARY,
                 "ocr_candidates": candidates,
-                "field_confidences": dict.fromkeys(extracted, 0.82),
+                "field_confidences": field_confidences,
             })
-
             combined_data.update(extracted)
 
-        except (ValueError, KeyError) as e:
-            # Expected errors — likely malformed input
-            log.exception(
-                "Gemini extraction failed — will retry with fallback",
-                doc_id=doc.get("doc_id"),
-                batch_id=state["batch_id"],
-                error_type=type(e).__name__
-            )
-            updated_docs.append({
-                **doc,
-                "error": str(e),
-                "fallback_required": True,
-                "ocr_method": "failed"
-            })
         except Exception as e:
-            # Unexpected errors are real production failures and should surface
-            # so the batch can be diagnosed instead of silently degrading.
+            # Per-document failure — mark for fallback, do NOT crash the batch
             log.exception(
-                "Unexpected error in LLM extraction — batch will fail",
+                "LLM extraction failed — marking doc for fallback",
                 doc_id=doc.get("doc_id"),
                 batch_id=state["batch_id"],
-                error_type=type(e).__name__
+                error_type=type(e).__name__,
+                error=str(e),
             )
             updated_docs.append({
                 **doc,
                 "error": str(e),
                 "fallback_required": True,
-                "ocr_method": "gemini_failed"
+                "ocr_method": "failed",
             })
-            raise
 
     return {
         "documents": updated_docs,
         "combined_data": combined_data,
-        "steps": ["llm_extraction"]
+        "steps": ["llm_extraction"],
     }

src/ai/nodes/fallback_ocr.py

@@ -22,7 +22,10 @@ def _needs_reconciliation(doc: dict) -> bool:
         or not doc.get("extracted_data")
         or doc.get("quality_score", 1.0) < settings.OCR_FALLBACK_TRIGGER_QUALITY
         or bool(doc.get("ocr_conflicts"))
-        or len(doc.get("ocr_candidates") or {}) > 1
+        or (
+            len(doc.get("ocr_candidates") or {}) > 1
+            and doc.get("document_mode") != "digital_pdf_text"
+        )
         or bool(
             confidences
             and min(confidences.values()) < settings.OCR_FALLBACK_TRIGGER_CONFIDENCE

src/ai/nodes/risk.py

@@ -115,6 +115,15 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
     )
     crs_score = round(crs_raw * 100, 2)
     crs_grade = _crs_to_grade(crs_score)
+    critical_failures = sum(1 for r in validation_results if r.get("severity") == "CRITICAL_FAIL")
+    warnings = sum(1 for r in validation_results if r.get("severity") == "WARNING")
+
+    if critical_failures:
+        crs_score = min(crs_score, 55.0)
+        crs_grade = _crs_to_grade(crs_score)
+    elif warnings:
+        crs_score = min(crs_score, 75.0)
+        crs_grade = _crs_to_grade(crs_score)
 
     features = {
         "doc_quality_score": p_quality,
@@ -127,6 +136,10 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
         "gross_weight_kg": float(combined_data.get("gross_weight") or 0.0),
     }
     rejection_prob = round(rejection_predictor.predict_proba(features), 4)
+    if critical_failures:
+        rejection_prob = max(rejection_prob, 0.65)
+    elif warnings:
+        rejection_prob = max(rejection_prob, 0.30)
     risk_level = _probability_to_risk(rejection_prob)
 
     # PRD §13 Invariant: CRS < 70 → must NOT auto-submit
@@ -147,6 +160,10 @@ async def risk_assessment_node(state: ExtractionGraphState) -> dict:
 
     return {
         "risk_level": risk_level,
+        "customs_readiness_score": crs_score,
+        "crs_grade": crs_grade,
+        "rejection_probability": rejection_prob,
+        "risk_features": features,
         "needs_human_review": needs_human_review,
         "steps": ["risk_assessment"],
         # NOTE: crs_score and rejection_prob are persisted to DB in the

src/ai/nodes/validate.py

@@ -12,10 +12,20 @@ log = structlog.get_logger()
 async def validation_node(state: ExtractionGraphState) -> dict:
     """
     Step 2.4: Cross-Document Validation against JSON rules.
+    Gracefully handles rule evaluation errors to avoid crashing the pipeline.
     """
     log.info("Running validation_node", batch_id=state["batch_id"])
 
-    results, needs_review = validation_rules_service.evaluate(state)
+    try:
+        results, needs_review = validation_rules_service.evaluate(state)
+    except Exception as exc:
+        log.warning(
+            "Validation rules evaluation failed — marking for review",
+            batch_id=state["batch_id"],
+            error=str(exc),
+        )
+        results = []
+        needs_review = True
 
     return {
         "validation_results": results,

src/ai/state.py

@@ -28,6 +28,10 @@ class ExtractionGraphState(TypedDict):
     validation_results: list[dict]
     needs_human_review: bool
     risk_level: str
+    customs_readiness_score: float | None
+    crs_grade: str | None
+    rejection_probability: float | None
+    risk_features: dict
     ocr_conflicts: list[dict]
     field_confidences: dict[str, float]
     # Keep track of which node executed

src/config.py

@@ -27,7 +27,7 @@ class Settings(BaseSettings):
     ENVIRONMENT: Literal["development", "staging", "production"] = "development"
     DEBUG: bool = False
     SECRET_KEY: SecretStr = Field(..., min_length=32)
-    CORS_ORIGINS: list[str] | str = ["*"]
+    CORS_ORIGINS: list[str] = ["*"]
 
     # ── Database ──────────────────────────────────────────────────────────────
     DATABASE_URL: str  # asyncpg connection string e.g. postgresql+asyncpg://...
@@ -59,6 +59,9 @@ class Settings(BaseSettings):
 
     # ── AI Inference Services (SDD §2.3–2.6) ─────────────────────────────────
     CLOUD_LLM_ONLY: bool = False                                    # Bypass heavy local models and use Gemini API instead
+    ENABLE_DUAL_OCR: bool = False                                   # Use both Surya and PaddleOCR for fallback/validation
+    OCR_FALLBACK_TRIGGER_QUALITY: float = 0.85                      # Fallback threshold
+    OCR_FALLBACK_TRIGGER_CONFIDENCE: float = 0.80
     SURYA_INFERENCE_URL: AnyHttpUrl = "http://surya-svc:8001"       # Agent A
     OLM_INFERENCE_URL: AnyHttpUrl = "http://olm-inference:8000"     # Agent D
     PADDLEOCR_SVC_URL: AnyHttpUrl = "http://paddleocr-svc:8002"     # Agent B
@@ -68,10 +71,10 @@ class Settings(BaseSettings):
     HF_TOKEN: SecretStr = ""  # type: ignore[assignment]
 
     # ── Azure Document Intelligence — Agent C ─────────────────────────────────
-    AZURE_DI_ENDPOINT: str | None = None
-    AZURE_DI_KEY: SecretStr = ""  # type: ignore[assignment]
-    AZURE_DI_MODEL_ID: str = "prebuilt-document"
-    AZURE_DI_FREE_LIMIT: int = 5000   # Pages/month on F0 tier (Invariant #9)
+    AZURE_DI_ENDPOINT: str | None = None
+    AZURE_DI_KEY: SecretStr = ""  # type: ignore[assignment]
+    AZURE_DI_MODEL_ID: str = "prebuilt-read"
+    AZURE_DI_FREE_LIMIT: int = 5000   # Pages/month on F0 tier (Invariant #9)
 
     # ── CEISA (Simulator in dev, real endpoint in prod) ───────────────────────
     CEISA_BASE_URL: AnyHttpUrl = "http://simulator:8006"
@@ -102,9 +105,12 @@ class Settings(BaseSettings):
     CHROMADB_PORT: int = 8000
 
     # ── AI / LLM ─────────────────────────────────────────────────────────────
+    USE_LOCAL_LLM: bool = False
+    LOCAL_LLM_MODEL: str = "qwen2.5:7b"
+    OLLAMA_BASE_URL: str = "http://host.docker.internal:11434/v1"
     GEMINI_API_KEY: SecretStr = Field(..., description="Google Gemini API key")
-    GEMINI_MODEL_PRIMARY: str = "gemini-2.5-pro"
-    GEMINI_MODEL_FALLBACK: str = "gemini-2.5-flash"
+    GEMINI_MODEL_PRIMARY: str = "gemini-3.5-flash"
+    GEMINI_MODEL_FALLBACK: str = "gemini-3.1-flash-lite"
     OPENAI_API_KEY: SecretStr = ""  # type: ignore[assignment]
     EMBEDDING_MODEL: str = "text-embedding-3-small"
 
@@ -114,10 +120,9 @@ class Settings(BaseSettings):
     LANGCHAIN_API_KEY: SecretStr = ""  # type: ignore[assignment]
 
     # ── Feature Flags ─────────────────────────────────────────────────────────
-    ENABLE_SURYA_AGENT: bool = True
-    ENABLE_AZURE_DI_AGENT: bool = True
-    ENABLE_DUAL_OCR: bool = True  # Legacy alias used by lightweight OCR service
-    ENABLE_VESSEL_VALIDATION: bool = True
+    ENABLE_SURYA_AGENT: bool = True
+    ENABLE_AZURE_DI_AGENT: bool = True
+    ENABLE_VESSEL_VALIDATION: bool = True
     ENABLE_BLOCKCHAIN: bool = True  # type: ignore[assignment] — redeclared intentionally
     ENABLE_INSW_CHECK: bool = True
     ENABLE_NOTIFICATIONS_WHATSAPP: bool = False
@@ -133,8 +138,8 @@ class Settings(BaseSettings):
     OCR_MAX_RENDERED_PAGES: int = 10
     OCR_MAX_LLM_PAGES: int = 5
     OCR_FAST_PATH_QUALITY_THRESHOLD: float = 0.95
-    OCR_FALLBACK_TRIGGER_QUALITY: float = 0.75
-    OCR_FALLBACK_TRIGGER_CONFIDENCE: float = 0.78
+    OCR_PDF_TEXT_MIN_CHARS: int = 250
+    OCR_PDF_TEXT_MIN_CHARS_PER_PAGE: int = 80
     OCR_RECONCILIATION_DISAGREEMENT_THRESHOLD: float = 0.20
     LLM_CONFIDENCE_REVIEW_THRESHOLD: float = 0.70
     CRS_MIN_SUBMIT_THRESHOLD: int = 55
@@ -161,9 +166,12 @@ class Settings(BaseSettings):
     DRIFT_LOOKBACK_DAYS: int = 30
     DRIFT_CORRECTION_THRESHOLD: int = 50
 
-    # ── Celery ────────────────────────────────────────────────────────────────
-    CELERY_TASK_SOFT_TIME_LIMIT: int = 300
-    CELERY_TASK_TIME_LIMIT: int = 600
+    # Celery Configuration
+    CELERY_BROKER_URL: str = "redis://redis:6379/0"
+    CELERY_RESULT_BACKEND: str = "redis://redis:6379/0"
+    CELERY_TASK_SOFT_TIME_LIMIT: int = 1800  # 30 minutes for slow OCR models
+    CELERY_TASK_TIME_LIMIT: int = 1900
+    RUN_OCR_IN_API_BACKGROUND: bool = False
 
     # ── Observability ─────────────────────────────────────────────────────────
     SENTRY_DSN: str = ""
@@ -193,8 +201,8 @@ class CeleryConfig:
     accept_content = ["json"]
     timezone = "Asia/Jakarta"
     enable_utc = True
-    task_soft_time_limit = 300
-    task_time_limit = 600
+    task_soft_time_limit = 600
+    task_time_limit = 700
     task_acks_late = True           # Ack only after successful completion (NFR-016)
     worker_prefetch_multiplier = 1  # One task at a time per worker
 

src/dependencies.py

@@ -54,17 +54,10 @@ async def close_supabase() -> None:
 
 def get_supabase() -> AsyncClient:
     if _supabase_client is None:
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="Supabase client is not initialized. Check SUPABASE_URL and SUPABASE_SERVICE_KEY.",
-        )
+        raise RuntimeError("Supabase client not initialized. Call init_supabase() first.")
     return _supabase_client
 
 
-def is_supabase_initialized() -> bool:
-    return _supabase_client is not None
-
-
 # ── Keycloak JWKS cache ───────────────────────────────────────────────────────
 _keycloak_jwks: dict | None = None
 _keycloak_jwks_time: float = 0

src/main.py

@@ -29,7 +29,7 @@ except Exception:  # pragma: no cover - optional
     Instrumentator = None
 
 from .config import settings
-from .dependencies import close_supabase, init_supabase, is_supabase_initialized
+from .dependencies import close_supabase, init_supabase
 from .routers import admin, batches, blockchain, hs_recommend, vessel
 from .utils.telemetry import setup_telemetry
 
@@ -120,38 +120,6 @@ def create_app() -> FastAPI:
 app = create_app()
 
 
-def _configured_secret(value: object) -> bool:
-    raw = value.get_secret_value() if hasattr(value, "get_secret_value") else value
-    text = str(raw or "").strip()
-    if not text:
-        return False
-    lowered = text.lower()
-    return not (
-        lowered.startswith("your-")
-        or lowered.startswith("test-")
-        or "placeholder" in lowered
-    )
-
-
-@app.get("/health/config", tags=["health"])
-async def health_config() -> dict[str, object]:
-    """Safe deployment diagnostics. Returns booleans only for secrets."""
-    return {
-        "status": "ok",
-        "environment": settings.ENVIRONMENT,
-        "cloud_llm_only": settings.CLOUD_LLM_ONLY,
-        "deterministic_e2e": settings.DETERMINISTIC_E2E,
-        "storage_backend": settings.STORAGE_BACKEND,
-        "storage_bucket": settings.STORAGE_BUCKET_NAME,
-        "supabase_configured": bool(settings.SUPABASE_URL) and _configured_secret(settings.SUPABASE_SERVICE_KEY),
-        "supabase_client_initialized": is_supabase_initialized(),
-        "gemini_api_key_configured": _configured_secret(settings.GEMINI_API_KEY),
-        "azure_di_enabled": settings.ENABLE_AZURE_DI_AGENT,
-        "azure_di_configured": bool(settings.AZURE_DI_ENDPOINT) and _configured_secret(settings.AZURE_DI_KEY),
-        "celery_broker_configured": bool(settings.REDIS_URL),
-    }
-
-
 @app.get("/health", tags=["health"])
 async def health_check() -> dict[str, str]:
     """Health check endpoint — public, no auth required."""

src/routers/batches.py

@@ -25,7 +25,7 @@ except Exception:  # pragma: no cover - optional for tests
 
 from ..dependencies import CurrentUser, get_current_user, get_supabase, require_operator
 from ..services.ingest_svc import get_storage_service
-from ..tasks.ocr_tasks import preprocess_document
+from ..tasks.ocr_tasks import preprocess_document, run_preprocess_pipeline_sync
 
 log = structlog.get_logger()
 router = APIRouter()
@@ -151,12 +151,16 @@ async def create_batch(
 
     await supabase.table("batches").update({"status": "preprocessing"}).eq("id", batch_id).execute()
 
-    queue = "high" if user.is_enterprise else "default"
-    try:
-        preprocess_document.apply_async(args=[batch_id], queue=queue)
-    except Exception as e:
-        log.warning("Celery apply_async failed, falling back to BackgroundTasks", error=str(e))
-        background_tasks.add_task(preprocess_document, batch_id)
+    if settings.RUN_OCR_IN_API_BACKGROUND:
+        log.info("Running OCR pipeline via FastAPI BackgroundTasks", batch_id=batch_id)
+        background_tasks.add_task(run_preprocess_pipeline_sync, batch_id, True)
+    else:
+        queue = "high" if user.is_enterprise else "default"
+        try:
+            preprocess_document.apply_async(args=[batch_id], queue=queue)
+        except Exception as e:
+            log.warning("Celery apply_async failed, falling back to BackgroundTasks", error=str(e))
+            background_tasks.add_task(run_preprocess_pipeline_sync, batch_id, True)
 
     log.info("Batch created", batch_id=batch_id, user=user.id, docs=len(files), tier=user.tier)
     return {"batch_id": batch_id, "status": "preprocessing", "documents": documents}
@@ -170,19 +174,18 @@ async def list_batches(
     offset: int = 0,
 ) -> dict[str, Any]:
     """List batches for the current user's company."""
-    res = await (
+    query = (
         supabase.table("batches")
         .select("id,status,customs_readiness_score,crs_grade,risk_level,created_at,expires_at")
-        .eq("company_id", user.company_id)
         .order("created_at", desc=True)
         .range(offset, offset + limit - 1)
-        .execute()
     )
-    batches = [_decorate_batch_summary(row) for row in (res.data or [])]
-    importer_by_batch = await _load_importer_names(supabase, [row["id"] for row in batches])
-    for row in batches:
-        row["importer"] = importer_by_batch.get(row["id"], "Pending extraction")
-    return {"batches": batches, "total": len(batches)}
+
+    if user.company_id:
+        query = query.eq("company_id", user.company_id)
+
+    res = await query.execute()
+    return {"batches": res.data, "total": len(res.data)}
 
 
 @router.get("/batches/{batch_id}")
@@ -192,7 +195,17 @@ async def get_batch(
     supabase: Annotated[AsyncClient, Depends(get_supabase)],
 ) -> dict[str, Any]:
     """Get full batch details including extracted fields and validation results."""
-    batch_res = await supabase.table("batches").select("*").eq("id", batch_id).single().execute()
+    try:
+        uuid.UUID(batch_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=404, detail="Batch not found") from exc
+
+    try:
+        batch_res = await supabase.table("batches").select("*").eq("id", batch_id).single().execute()
+    except Exception as exc:
+        log.warning("Failed to load batch", batch_id=batch_id, error=str(exc))
+        raise HTTPException(status_code=404, detail="Batch not found") from exc
+
     batch = batch_res.data
     if not batch:
         raise HTTPException(status_code=404, detail="Batch not found")
@@ -202,15 +215,11 @@ async def get_batch(
     docs_res = await supabase.table("documents").select("*").eq("batch_id", batch_id).execute()
     fields_res = await supabase.table("extracted_fields").select("*").eq("batch_id", batch_id).execute()
     validations_res = await supabase.table("validation_results").select("*").eq("batch_id", batch_id).execute()
-    fields = fields_res.data or []
-    reconciled_fields = _build_reconciled_fields(fields)
-    batch["importer"] = _importer_from_reconciled(reconciled_fields)
 
     return {
         "batch": batch,
         "documents": docs_res.data,
-        "extracted_fields": fields,
-        "reconciled_fields": reconciled_fields,
+        "extracted_fields": fields_res.data,
         "validation_results": validations_res.data,
     }
 
@@ -297,72 +306,3 @@ def _infer_doc_type(filename: str) -> str:
     if any(k in name for k in ("pl", "packing", "packinglist")):
         return "packing_list"
     return "invoice"
-
-
-def _decorate_batch_summary(row: dict[str, Any]) -> dict[str, Any]:
-    score = row.get("customs_readiness_score")
-    grade = row.get("crs_grade")
-    risk = row.get("risk_level")
-    return {
-        **row,
-        "ref": str(row.get("id", ""))[:8].upper(),
-        "type": "PIB",
-        "risk": risk or "LOW",
-        "crs": float(score or 0),
-        "grade": grade or "F",
-        "date": row.get("created_at"),
-    }
-
-
-async def _load_importer_names(supabase: AsyncClient, batch_ids: list[str]) -> dict[str, str]:
-    if not batch_ids:
-        return {}
-    try:
-        fields_res = await (
-            supabase.table("extracted_fields")
-            .select("batch_id,ceisa_field,extracted_value,normalized_value")
-            .in_("batch_id", batch_ids)
-            .eq("ceisa_field", "importer_name")
-            .execute()
-        )
-    except Exception as exc:
-        log.warning("Could not load importer names for batch list", error=str(exc))
-        return {}
-
-    names: dict[str, str] = {}
-    for field in fields_res.data or []:
-        value = field.get("normalized_value") or field.get("extracted_value")
-        if value:
-            names[field["batch_id"]] = str(value)
-    return names
-
-
-def _build_reconciled_fields(fields: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
-    reconciled: dict[str, dict[str, Any]] = {}
-    for field in fields:
-        name = field.get("ceisa_field")
-        if not name:
-            continue
-        confidence = float(field.get("confidence") or 0.0)
-        reconciled[name] = {
-            "value": field.get("normalized_value") or field.get("extracted_value") or field.get("raw_ocr_value"),
-            "confidence": confidence,
-            "level": field.get("confidence_level") or _confidence_level(confidence),
-            "source": field.get("extraction_method") or "direct_ocr",
-            "agent_disagreement": bool(field.get("agent_disagreement")),
-            "all_agent_values": field.get("agent_outputs") or {},
-        }
-    return reconciled
-
-
-def _confidence_level(confidence: float) -> str:
-    if confidence >= 0.90:
-        return "HIGH"
-    if confidence >= 0.70:
-        return "MEDIUM"
-    return "LOW"
-
-
-def _importer_from_reconciled(fields: dict[str, dict[str, Any]]) -> str:
-    value = fields.get("importer_name", {}).get("value")
-    return str(value) if value else "Pending extraction"

src/scripts/backfill_batch_risk.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from typing import Any
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from supabase import acreate_client
+
+from src.ai.nodes.risk import risk_assessment_node
+from src.config import settings
+
+
+def _coerce_field(field: str, value: Any) -> Any:
+    if value is None:
+        return None
+    if field in {"gross_weight", "cif_value", "fob_value"}:
+        try:
+            return float(str(value).replace(",", ""))
+        except Exception:
+            return value
+    if field == "total_packages":
+        try:
+            return int(float(str(value).replace(",", "")))
+        except Exception:
+            return value
+    return value
+
+
+async def backfill_batch_risk(batch_id: str) -> dict[str, Any]:
+    supabase = await acreate_client(
+        settings.SUPABASE_URL,
+        settings.SUPABASE_SERVICE_KEY.get_secret_value(),
+    )
+
+    docs = (
+        await supabase.table("documents").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+    fields = (
+        await supabase.table("extracted_fields").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+    validations = (
+        await supabase.table("validation_results").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+
+    combined: dict[str, Any] = {}
+    confidences: dict[str, float] = {}
+    for row in fields:
+        name = row.get("ceisa_field")
+        if not name:
+            continue
+        combined[name] = _coerce_field(
+            name,
+            row.get("normalized_value") or row.get("extracted_value"),
+        )
+        confidences[name] = float(row.get("confidence") or 0.0)
+
+    state = {
+        "batch_id": batch_id,
+        "company_id": "",
+        "documents": [
+            {
+                "doc_id": doc["id"],
+                "doc_type": doc.get("doc_type"),
+                "storage_path": doc.get("storage_path"),
+                "pages": [],
+                "extracted_data": {},
+                "quality_score": float(doc.get("quality_score") or 1.0),
+                "ocr_method": doc.get("ocr_engine_used"),
+                "error": doc.get("error_message"),
+                "ocr_candidates": {},
+                "ocr_conflicts": [],
+                "field_confidences": {},
+            }
+            for doc in docs
+        ],
+        "combined_data": combined,
+        "validation_results": validations,
+        "needs_human_review": False,
+        "risk_level": "UNKNOWN",
+        "customs_readiness_score": None,
+        "crs_grade": None,
+        "rejection_probability": None,
+        "risk_features": {},
+        "ocr_conflicts": [],
+        "field_confidences": confidences,
+        "steps": [],
+    }
+
+    result = await risk_assessment_node(state)  # type: ignore[arg-type]
+    payload = {
+        "risk_level": result.get("risk_level"),
+        "customs_readiness_score": result.get("customs_readiness_score"),
+        "crs_grade": result.get("crs_grade"),
+        "rejection_probability": result.get("rejection_probability"),
+    }
+    await supabase.table("batches").update(payload).eq("id", batch_id).execute()
+    return payload
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        raise SystemExit("Usage: python /app/src/scripts/backfill_batch_risk.py <batch_id>")
+    print(asyncio.run(backfill_batch_risk(sys.argv[1])))

src/scripts/recompute_field_confidences.py

@@ -0,0 +1,109 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from supabase import acreate_client
+
+from src.ai.nodes.extract import _estimate_field_confidences
+from src.config import settings
+from src.services.ingest_svc import get_storage_service
+from src.services.ocr_engine_svc import ocr_engine_service
+
+
+async def recompute_field_confidences(batch_id: str) -> dict[str, Any]:
+    supabase = await acreate_client(
+        settings.SUPABASE_URL,
+        settings.SUPABASE_SERVICE_KEY.get_secret_value(),
+    )
+
+    docs = (
+        await supabase.table("documents").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+    rows = (
+        await supabase.table("extracted_fields").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+
+    fields_by_doc: dict[str, dict[str, Any]] = {}
+    row_keys_by_doc_field: dict[tuple[str, str], list[dict[str, Any]]] = {}
+    for row in rows:
+        document_id = row.get("document_id")
+        field = row.get("ceisa_field")
+        if not document_id or not field:
+            continue
+        fields_by_doc.setdefault(document_id, {})[field] = row.get("normalized_value") or row.get("extracted_value")
+        row_keys_by_doc_field.setdefault((document_id, field), []).append(row)
+
+    storage = get_storage_service()
+    updated = 0
+    doc_summaries = []
+
+    for doc in docs:
+        doc_id = doc["id"]
+        extracted = fields_by_doc.get(doc_id) or {}
+        if not extracted:
+            continue
+
+        raw_text = ""
+        direct_candidate: dict[str, Any] = {}
+        try:
+            file_bytes = await storage.download_document(doc["storage_path"])
+            filename = doc.get("original_name") or doc.get("storage_path") or ""
+            if str(filename).lower().endswith(".pdf"):
+                direct_candidate = ocr_engine_service._extract_pdf_text(file_bytes)
+                raw_text = direct_candidate.get("text") or ""
+        except Exception as exc:
+            print(f"warning: failed to reload {doc.get('original_name')}: {exc}")
+
+        doc_state = {
+            "document_mode": "digital_pdf_text" if raw_text else doc.get("processing_route"),
+            "raw_text": raw_text,
+            "ocr_candidates": {"pdf_text": direct_candidate} if direct_candidate else {},
+        }
+        confidences = _estimate_field_confidences(extracted, doc_state)
+        if not confidences:
+            continue
+
+        for field, confidence in confidences.items():
+            for row in row_keys_by_doc_field.get((doc_id, field), []):
+                row_id = row.get("id")
+                if row_id:
+                    await supabase.table("extracted_fields").update({"confidence": confidence}).eq("id", row_id).execute()
+                else:
+                    await (
+                        supabase.table("extracted_fields")
+                        .update({"confidence": confidence})
+                        .eq("batch_id", batch_id)
+                        .eq("document_id", doc_id)
+                        .eq("ceisa_field", field)
+                        .execute()
+                    )
+                updated += 1
+
+        avg_conf = round(sum(confidences.values()) / len(confidences), 4)
+        await (
+            supabase.table("documents")
+            .update({"overall_ocr_confidence": avg_conf})
+            .eq("id", doc_id)
+            .execute()
+        )
+        doc_summaries.append(
+            {
+                "doc_type": doc.get("doc_type"),
+                "original_name": doc.get("original_name"),
+                "avg_confidence": avg_conf,
+                "fields": len(confidences),
+            }
+        )
+
+    return {"batch_id": batch_id, "updated_rows": updated, "documents": doc_summaries}
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        raise SystemExit("Usage: python /app/src/scripts/recompute_field_confidences.py <batch_id>")
+    print(asyncio.run(recompute_field_confidences(sys.argv[1])))

src/scripts/revalidate_batch.py

@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from supabase import acreate_client
+
+from src.ai.nodes.risk import risk_assessment_node
+from src.config import settings
+from src.services.validation_rules_svc import validation_rules_service
+
+
+def _coerce_field(field: str, value: Any) -> Any:
+    if value is None:
+        return None
+    if field in {"gross_weight", "cif_value", "fob_value", "freight_value", "insurance_value"}:
+        try:
+            return float(str(value).replace(",", ""))
+        except Exception:
+            return value
+    if field == "total_packages":
+        try:
+            return int(float(str(value).replace(",", "")))
+        except Exception:
+            return value
+    return value
+
+
+async def revalidate_batch(batch_id: str) -> dict[str, Any]:
+    supabase = await acreate_client(
+        settings.SUPABASE_URL,
+        settings.SUPABASE_SERVICE_KEY.get_secret_value(),
+    )
+    docs = (
+        await supabase.table("documents").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+    fields = (
+        await supabase.table("extracted_fields").select("*").eq("batch_id", batch_id).execute()
+    ).data or []
+
+    by_doc: dict[str, dict[str, Any]] = {}
+    combined: dict[str, Any] = {}
+    confidences: dict[str, float] = {}
+    for row in fields:
+        document_id = row.get("document_id")
+        name = row.get("ceisa_field")
+        if not document_id or not name:
+            continue
+        value = _coerce_field(name, row.get("normalized_value") or row.get("extracted_value"))
+        by_doc.setdefault(document_id, {})[name] = value
+        combined[name] = value
+        confidences[name] = float(row.get("confidence") or 0.0)
+
+    state = {
+        "batch_id": batch_id,
+        "company_id": "",
+        "documents": [
+            {
+                "doc_id": doc["id"],
+                "doc_type": doc.get("doc_type"),
+                "storage_path": doc.get("storage_path"),
+                "pages": [],
+                "extracted_data": by_doc.get(doc["id"], {}),
+                "quality_score": float(doc.get("quality_score") or 1.0),
+                "ocr_method": doc.get("ocr_engine_used"),
+                "error": doc.get("error_message"),
+                "ocr_candidates": {},
+                "ocr_conflicts": [],
+                "field_confidences": {},
+            }
+            for doc in docs
+        ],
+        "combined_data": combined,
+        "validation_results": [],
+        "needs_human_review": False,
+        "risk_level": "UNKNOWN",
+        "customs_readiness_score": None,
+        "crs_grade": None,
+        "rejection_probability": None,
+        "risk_features": {},
+        "ocr_conflicts": [],
+        "field_confidences": confidences,
+        "steps": [],
+    }
+
+    validations, needs_review = validation_rules_service.evaluate(state)
+    state["validation_results"] = validations
+    state["needs_human_review"] = needs_review
+    risk = await risk_assessment_node(state)  # type: ignore[arg-type]
+
+    await supabase.table("validation_results").delete().eq("batch_id", batch_id).execute()
+    if validations:
+        await supabase.table("validation_results").insert([
+            {
+                "batch_id": batch_id,
+                "rule_id": row.get("rule_id", "UNKNOWN"),
+                "rule_name": row.get("rule_name", row.get("message", "Validation")),
+                "severity": row.get("severity", "WARNING"),
+                "error_message": row.get("message"),
+                "affected_fields": row.get("affected_fields", []),
+            }
+            for row in validations
+        ]).execute()
+
+    payload = {
+        "status": "review_ready" if risk.get("needs_human_review") else "validated",
+        "risk_level": risk.get("risk_level"),
+        "customs_readiness_score": risk.get("customs_readiness_score"),
+        "crs_grade": risk.get("crs_grade"),
+        "rejection_probability": risk.get("rejection_probability"),
+    }
+    await supabase.table("batches").update(payload).eq("id", batch_id).execute()
+    return {
+        **payload,
+        "validation_counts": {
+            severity: sum(1 for row in validations if row.get("severity") == severity)
+            for severity in {"PASS", "WARNING", "CRITICAL_FAIL"}
+        },
+    }
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        raise SystemExit("Usage: python /app/src/scripts/revalidate_batch.py <batch_id>")
+    print(asyncio.run(revalidate_batch(sys.argv[1])))

src/services/ingest_svc.py

@@ -28,8 +28,8 @@ class StorageService:
                 "s3",
                 endpoint_url=f"http://{settings.MINIO_ENDPOINT}",
                 aws_access_key_id=settings.MINIO_ACCESS_KEY,
-                aws_secret_access_key=settings.MINIO_SECRET_KEY,
-                config=Config(signature_version="s3v4"),
+                aws_secret_access_key=settings.MINIO_SECRET_KEY.get_secret_value() if hasattr(settings.MINIO_SECRET_KEY, 'get_secret_value') else settings.MINIO_SECRET_KEY,
+                config=Config(signature_version="s3v4", s3={'addressing_style': 'path'}),
                 region_name="us-east-1",
             )
             self._ensure_minio_bucket()

src/services/ocr_engine_svc.py

@@ -56,8 +56,16 @@ CEISA_FIELD_PATTERNS = {
 
 def _as_float(value: str) -> float | None:
     try:
-        return float(value.replace(",", ""))
-    except ValueError:
+        # Handle European format: "11,603.000" (comma = thousands separator)
+        # Detect if comma is used as thousands: e.g., "11,603.000" has comma before 3+ digits before decimal
+        cleaned = value.strip()
+        if re.search(r"\d,\d{3}(\.|$)", cleaned):
+            cleaned = cleaned.replace(",", "")
+        else:
+            # Could be decimal comma: "11.603,000" -> 11603.0
+            cleaned = cleaned.replace(".", "").replace(",", ".")
+        return float(cleaned)
+    except (ValueError, AttributeError):
         return None
 
 
@@ -106,59 +114,62 @@ class OCREngineService:
         started = time.perf_counter()
         suffix = Path(filename or storage_path).suffix.lower()
         mime_type = mimetypes.guess_type(filename or storage_path)[0] or "application/octet-stream"
+        is_pdf = suffix == ".pdf" or mime_type == "application/pdf"
 
-        page_images = await asyncio.to_thread(self._render_page_images, file_bytes, suffix)
         raw_text = ""
         candidates: dict[str, dict[str, Any]] = {}
 
+        if is_pdf:
+            direct_candidate = await asyncio.to_thread(self._extract_pdf_text, file_bytes)
+            raw_text = direct_candidate.get("text", "")
+            if direct_candidate.get("fields") or raw_text.strip():
+                candidates["pdf_text"] = direct_candidate
+
+            if self._is_pdf_text_fast_path(direct_candidate):
+                latency_ms = round((time.perf_counter() - started) * 1000, 2)
+                log.info(
+                    "Digital PDF fast path selected",
+                    doc_id=doc_id,
+                    storage_path=storage_path,
+                    text_chars=direct_candidate.get("text_chars"),
+                    text_chars_per_page=direct_candidate.get("text_chars_per_page"),
+                    fields=list((direct_candidate.get("fields") or {}).keys()),
+                    latency_ms=latency_ms,
+                )
+                return {
+                    "pages": [],
+                    "raw_text": raw_text,
+                    "ocr_candidates": candidates,
+                    "quality_score": float(direct_candidate.get("confidence") or 1.0),
+                    "document_mode": "digital_pdf_text",
+                    "ocr_engine_latencies_ms": {
+                        name: candidate.get("latency_ms") for name, candidate in candidates.items()
+                    },
+                }
+
+        page_images = await asyncio.to_thread(self._render_page_images, file_bytes, suffix)
+
         if settings.CLOUD_LLM_ONLY:
-            if suffix == ".pdf" or mime_type == "application/pdf":
-                direct_candidate = await asyncio.to_thread(self._extract_pdf_text, file_bytes)
-                raw_text = direct_candidate.get("text", "")
-                if direct_candidate.get("fields") or direct_candidate.get("text"):
-                    candidates["pdf_text"] = direct_candidate
-
-            if settings.ENABLE_AZURE_DI_AGENT:
-                azure_candidate = await self._run_azure(file_bytes, mime_type)
-                if azure_candidate.get("fields") or azure_candidate.get("text"):
-                    candidates["azure-di"] = azure_candidate
-                    raw_text = "\n".join(
-                        part for part in [raw_text, azure_candidate.get("text", "")] if part
-                    )
-
-            quality_score = self._estimate_quality(page_images) if page_images else 0.0
             page_data_urls = [_data_url(b) for b in page_images[: settings.OCR_MAX_LLM_PAGES]]
             return {
                 "pages": page_data_urls,
-                "raw_text": raw_text,
-                "ocr_candidates": candidates,
-                "quality_score": quality_score,
-                "ocr_engine_latencies_ms": {
-                    name: candidate.get("latency_ms") for name, candidate in candidates.items()
-                },
+                "raw_text": "",
+                "ocr_candidates": {},
+                "quality_score": 1.0,
+                "ocr_engine_latencies_ms": {}
             }
 
-        pdf_task = None
-        if suffix == ".pdf" or mime_type == "application/pdf":
-            pdf_task = asyncio.to_thread(self._extract_pdf_text, file_bytes)
-
         paddle_task = self._run_paddle(page_images)
-        azure_task = self._run_azure(file_bytes, mime_type) if self._azure_enabled() else None
+        surya_task = self._run_surya(page_images) if settings.ENABLE_SURYA_AGENT else None
+        azure_task = self._run_azure(file_bytes, mime_type) if settings.ENABLE_DUAL_OCR else None
 
         gather_tasks = [paddle_task]
+        if surya_task is not None:
+            gather_tasks.append(surya_task)
         if azure_task is not None:
             gather_tasks.append(azure_task)
-        if pdf_task is not None:
-            gather_tasks.insert(0, pdf_task)
-
         results = await asyncio.gather(*gather_tasks)
         idx = 0
-        if pdf_task is not None:
-            direct_candidate = results[idx]
-            idx += 1
-            raw_text = direct_candidate.get("text", "")
-            if direct_candidate.get("fields"):
-                candidates["pdf_text"] = direct_candidate
 
         paddle_candidate = results[idx]
         idx += 1
@@ -168,6 +179,15 @@ class OCREngineService:
                 part for part in [raw_text, paddle_candidate.get("text", "")] if part
             )
 
+        if surya_task is not None:
+            surya_candidate = results[idx]
+            idx += 1
+            if surya_candidate.get("fields") or surya_candidate.get("text"):
+                candidates["surya"] = surya_candidate
+                raw_text = "\n".join(
+                    part for part in [raw_text, surya_candidate.get("text", "")] if part
+                )
+
         if azure_task is not None:
             azure_candidate = results[idx]
             if azure_candidate.get("fields") or azure_candidate.get("text"):
@@ -196,12 +216,50 @@ class OCREngineService:
             "raw_text": raw_text,
             "ocr_candidates": candidates,
             "quality_score": quality_score,
+            "document_mode": "rendered_ocr",
             "ocr_engine_latencies_ms": {
                 name: candidate.get("latency_ms") for name, candidate in candidates.items()
             },
         }
 
+    def _is_pdf_text_fast_path(self, candidate: dict[str, Any]) -> bool:
+        """Use direct PDF text when the text layer is dense enough to avoid heavy OCR."""
+        text_chars = int(candidate.get("text_chars") or 0)
+        page_count = max(1, int(candidate.get("page_count") or 1))
+        text_chars_per_page = text_chars / page_count
+        confidence = float(candidate.get("confidence") or 0.0)
+        return (
+            confidence >= settings.OCR_FAST_PATH_QUALITY_THRESHOLD
+            and text_chars >= settings.OCR_PDF_TEXT_MIN_CHARS
+            and text_chars_per_page >= settings.OCR_PDF_TEXT_MIN_CHARS_PER_PAGE
+        )
+
     def _render_page_images(self, file_bytes: bytes, suffix: str) -> list[bytes]:
+        """Call MinerU microservice for preprocessing and rendering."""
+        try:
+            import httpx
+            b64_content = base64.b64encode(file_bytes).decode("ascii")
+            payload = {
+                "document_id": "temp",
+                "doc_type": "invoice",
+                "content_b64": b64_content,
+                "filename": f"temp{suffix}"
+            }
+            url = f"{str(settings.MINERU_SVC_URL).rstrip('/')}/preprocess"
+            with httpx.Client(timeout=30.0) as client:
+                response = client.post(url, json=payload)
+                response.raise_for_status()
+                data = response.json()
+                
+            images = []
+            for page in data.get("pages", []):
+                images.append(base64.b64decode(page["image_b64"]))
+            if images:
+                return images
+        except Exception as exc:
+            log.warning("MinerU preprocessing failed, falling back to local PyMuPDF", error=str(exc))
+
+        # Fallback to local rendering
         if suffix == ".pdf":
             try:
                 import fitz
@@ -227,14 +285,21 @@ class OCREngineService:
             import pdfplumber
 
             with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
-                text = "\n".join((page.extract_text() or "") for page in pdf.pages)
+                pages = pdf.pages
+                text = "\n".join((page.extract_text() or "") for page in pages)
             fields = extract_ceisa_fields_from_text(text)
-            confidence = 0.98 if text.strip() else 0.0
+            text_chars = len(re.sub(r"\s+", "", text))
+            page_count = len(pages) or 1
+            text_chars_per_page = round(text_chars / page_count, 2)
+            confidence = 1.0 if text_chars_per_page >= settings.OCR_PDF_TEXT_MIN_CHARS_PER_PAGE else 0.0
             return {
                 "fields": fields,
                 "text": text,
                 "confidence": confidence,
                 "overall_confidence": confidence,
+                "page_count": page_count,
+                "text_chars": text_chars,
+                "text_chars_per_page": text_chars_per_page,
                 "latency_ms": round((time.perf_counter() - started) * 1000, 2),
             }
         except Exception as exc:
@@ -249,36 +314,93 @@ class OCREngineService:
     def _run_paddle_sync(self, page_images: list[bytes]) -> dict[str, Any]:
         started = time.perf_counter()
         try:
-            if self._paddle is None:
-                from paddleocr import PaddleOCR
-
-                try:
-                    self._paddle = PaddleOCR(
-                        use_angle_cls=True,
-                        lang=settings.PADDLEOCR_LANG,
-                        show_log=False,
-                    )
-                except TypeError:
-                    self._paddle = PaddleOCR(use_angle_cls=True, lang=settings.PADDLEOCR_LANG)
-
-            lines: list[str] = []
-            confidences: list[float] = []
-            with tempfile.TemporaryDirectory(prefix="tradeflow-paddle-") as tmpdir:
-                for index, image_bytes in enumerate(page_images):
-                    image_path = Path(tmpdir) / f"page-{index}.png"
-                    image_path.write_bytes(image_bytes)
-                    result = self._paddle.ocr(str(image_path), cls=True)
-                    for page in result or []:
-                        for item in page or []:
-                            if len(item) >= 2 and isinstance(item[1], (list, tuple)):
-                                text = str(item[1][0])
-                                confidence = float(item[1][1])
-                                if text.strip():
-                                    lines.append(text)
-                                    confidences.append(confidence)
+            import httpx
+            images_b64 = [base64.b64encode(img).decode("ascii") for img in page_images]
+            url = f"{str(settings.PADDLEOCR_SVC_URL).rstrip('/')}/extract"
+            
+            lines = []
+            confidences = []
+            
+            with httpx.Client(timeout=300.0) as client:
+                for img_b64 in images_b64:
+                    payload = {
+                        "image_b64": img_b64,
+                        "doc_type": "bill_of_lading"
+                    }
+                    response = client.post(url, json=payload)
+                    response.raise_for_status()
+                    result = response.json()
+                    
+                    for text_block in result.get("text_blocks_with_bbox", []):
+                        text = text_block.get("text", "")
+                        conf = text_block.get("confidence", 0.0)
+                        if text.strip():
+                            lines.append(text)
+                            confidences.append(conf)
+                
+            text = "\n".join(lines)
+            confidence = sum(confidences) / len(confidences) if confidences else 0.0
+            
+            return {
+                "fields": extract_ceisa_fields_from_text(text),
+                "text": text,
+                "confidence": confidence,
+                "overall_confidence": confidence,
+                "latency_ms": round((time.perf_counter() - started) * 1000, 2),
+            }
+        except Exception as exc:
+            log.warning("PaddleOCR HTTP failed", error=str(exc))
+            return {"fields": {}, "text": "", "confidence": 0.0, "error": str(exc)}
+
+    async def _run_surya(self, page_images: list[bytes]) -> dict[str, Any]:
+        if not page_images:
+            return {"fields": {}, "text": "", "confidence": 0.0}
+        return await asyncio.to_thread(self._run_surya_sync, page_images)
+
+    def _run_surya_sync(self, page_images: list[bytes]) -> dict[str, Any]:
+        started = time.perf_counter()
+        try:
+            import httpx
+            images_b64 = [base64.b64encode(img).decode("ascii") for img in page_images]
+            payload = {
+                "images_b64": images_b64,
+                "languages": ["en", "id"]
+            }
+            url = f"{str(settings.SURYA_INFERENCE_URL).rstrip('/')}/extract"
+            with httpx.Client(timeout=600.0) as client:
+                response = client.post(url, json=payload)
+                response.raise_for_status()
+                result = response.json()
+
+            # Surya v2 response: text_blocks is list[list[dict]] (per page, per block)
+            # Each block has: { text, html, confidence, bbox, polygon, label }
+            lines = []
+            confidences = []
+            text_blocks = result.get("text_blocks", [])
+            for page_blocks in text_blocks:
+                if isinstance(page_blocks, list):
+                    for block in page_blocks:
+                        text = block.get("text", "").strip()
+                        conf = float(block.get("confidence", 1.0))
+                        if text:
+                            lines.append(text)
+                            confidences.append(conf)
+                elif isinstance(page_blocks, dict):
+                    # Fallback: old format where text_blocks is flat list of dicts
+                    text = page_blocks.get("text", "").strip()
+                    conf = float(page_blocks.get("confidence", 1.0))
+                    if text:
+                        lines.append(text)
+                        confidences.append(conf)
+
+            # Also handle legacy flat "text" field
+            if not lines and result.get("text"):
+                lines = [result["text"]]
+                confidences = [result.get("confidence", result.get("overall_confidence", 0.0))]
 
             text = "\n".join(lines)
             confidence = sum(confidences) / len(confidences) if confidences else 0.0
+
             return {
                 "fields": extract_ceisa_fields_from_text(text),
                 "text": text,
@@ -287,11 +409,12 @@ class OCREngineService:
                 "latency_ms": round((time.perf_counter() - started) * 1000, 2),
             }
         except Exception as exc:
-            log.warning("PaddleOCR failed", error=str(exc))
+            log.warning("Surya OCR HTTP failed", error=str(exc))
             return {"fields": {}, "text": "", "confidence": 0.0, "error": str(exc)}
 
+
     async def _run_azure(self, file_bytes: bytes, mime_type: str) -> dict[str, Any]:
-        if not self._azure_enabled():
+        if not settings.ENABLE_DUAL_OCR:
             return {"fields": {}, "text": "", "confidence": 0.0}
         if not settings.AZURE_DI_ENDPOINT or not settings.AZURE_DI_KEY:
             log.warning("Azure DI not configured; dual OCR is degraded to PaddleOCR only")
@@ -350,9 +473,6 @@ class OCREngineService:
             log.warning("Azure DI failed", error=str(exc))
             return {"fields": {}, "text": "", "confidence": 0.0, "error": str(exc)}
 
-    def _azure_enabled(self) -> bool:
-        return bool(settings.ENABLE_AZURE_DI_AGENT and settings.ENABLE_DUAL_OCR)
-
     def _estimate_quality(self, page_images: list[bytes]) -> float:
         if not page_images:
             return 0.0

src/services/validation_rules_svc.py

@@ -215,23 +215,173 @@ class ValidationRulesService:
         evaluator: SafeRuleEvaluator,
         context: dict[str, Any],
     ) -> dict[str, Any]:
+        rule_id = rule.get("rule_id") or rule.get("id", "UNKNOWN")
+        rule_name = rule.get("name", rule_id)
         try:
-            passed = evaluator.evaluate(rule["check"])
+            check_expr = rule.get("check")
+            if not check_expr:
+                legacy_result = self._evaluate_legacy_rule(rule, context)
+                if legacy_result is not None:
+                    return legacy_result
+                # Rule has no evaluable expression — skip as PASS
+                return {
+                    "rule_id": rule_id,
+                    "rule_name": rule_name,
+                    "severity": "PASS",
+                    "message": rule_name,
+                    "affected_fields": rule.get("affected_fields") or rule.get("fields", []),
+                }
+            passed = evaluator.evaluate(check_expr)
             severity = "PASS" if passed else self._failure_severity(rule.get("severity"))
-            message = rule.get("name", rule.get("rule_id", "Validation rule"))
+            message = rule_name
             if not passed:
                 message = self._format_message(rule.get("error_message") or message, context)
         except Exception as exc:
             severity = self._failure_severity(rule.get("severity"))
-            message = f"{rule.get('name', rule.get('rule_id'))} could not be evaluated: {exc}"
+            message = f"{rule_name} could not be evaluated: {exc}"
 
         return {
-            "rule_id": rule.get("rule_id", "UNKNOWN"),
-            "rule_name": rule.get("name", rule.get("rule_id", "Validation rule")),
+            "rule_id": rule_id,
+            "rule_name": rule_name,
             "severity": severity,
             "message": message,
-            "affected_fields": rule.get("affected_fields", []),
+            "affected_fields": rule.get("affected_fields") or rule.get("fields", []),
+        }
+
+    def _evaluate_legacy_rule(
+        self,
+        rule: dict[str, Any],
+        context: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        rule_type = rule.get("type")
+        if not rule_type:
+            return None
+
+        rule_id = rule.get("rule_id") or rule.get("id", "UNKNOWN")
+        rule_name = rule.get("name", rule_id)
+        fields = rule.get("fields") or ([rule["field"]] if rule.get("field") else [])
+
+        try:
+            if rule_type in {"regex", "regex_and_lookup"}:
+                field = fields[0] if fields else rule.get("field")
+                value = self._first_context_value(context, field)
+                if field in {"npwp", "nib"}:
+                    value = re.sub(r"\D", "", str(value or ""))
+                passed = regex_match(value, rule.get("regex", ".*"))
+            elif rule_type == "cross_document_match":
+                passed = all(self._cross_document_values_match(context, field) for field in fields)
+            elif rule_type == "cross_document":
+                passed = self._evaluate_cross_document_rule(rule, context)
+            elif rule_type == "date_sequence":
+                passed = True
+            elif rule_type == "lookup":
+                passed = self._evaluate_lookup_rule(rule, context)
+            else:
+                return None
+        except Exception as exc:
+            passed = False
+            log.warning("Legacy validation rule failed to evaluate", rule_id=rule_id, error=str(exc))
+
+        severity = "PASS" if passed else self._legacy_failure_severity(rule_id, rule.get("severity"))
+        return {
+            "rule_id": rule_id,
+            "rule_name": rule_name,
+            "severity": severity,
+            "message": rule_name if passed else rule.get("description") or rule_name,
+            "affected_fields": rule.get("affected_fields") or fields,
+        }
+
+    def _field_aliases(self, field: str | None) -> list[str]:
+        aliases = {
+            "nomorBl": ["bl_number", "nomorBl"],
+            "beratKotor": ["gross_weight", "beratKotor"],
+            "jumlahKemasan": ["total_packages", "jumlahKemasan"],
+            "namaKapal": ["vessel_name", "namaKapal"],
+            "voyageNumber": ["voyage_number", "voyageNumber"],
+            "kodePelabuhanMuat": ["port_of_loading", "kodePelabuhanMuat"],
+            "kodePelabuhanBongkar": ["port_of_discharge", "kodePelabuhanBongkar"],
+            "hs_code": ["hs_code", "posTarif"],
+            "nib": ["importer_nib", "nib", "nibEntitas"],
+            "npwp": ["importer_npwp", "npwp", "nomorIdentitas"],
+            "container_number": ["container_numbers", "container_number"],
         }
+        if not field:
+            return []
+        return aliases.get(field, [field])
+
+    def _scope_value(self, scope: Any, field: str | None) -> Any:
+        for alias in self._field_aliases(field):
+            value = _unwrap(getattr(scope, alias, MISSING))
+            if value not in (None, "", MISSING):
+                return value
+        return None
+
+    def _first_context_value(self, context: dict[str, Any], field: str | None) -> Any:
+        for scope_name in ("data", "inv", "pl", "bl", "item", "importir"):
+            value = self._scope_value(context[scope_name], field)
+            if value not in (None, "", MISSING):
+                return value
+        return None
+
+    def _cross_document_values_match(self, context: dict[str, Any], field: str) -> bool:
+        values = [
+            self._normalize_compare_value(self._scope_value(context[scope_name], field))
+            for scope_name in ("bl", "pl", "inv")
+        ]
+        present = [value for value in values if value not in (None, "")]
+        if len(present) < 2:
+            return False
+        return len(set(present)) == 1
+
+    def _normalize_compare_value(self, value: Any) -> str | None:
+        if value is None or value is MISSING:
+            return None
+        if isinstance(value, (int, float)):
+            return str(round(float(value), 4))
+        return re.sub(r"\s+", " ", str(value)).strip().upper()
+
+    def _evaluate_cross_document_rule(self, rule: dict[str, Any], context: dict[str, Any]) -> bool:
+        rule_id = rule.get("rule_id") or rule.get("id")
+        tolerance_pct = float(rule.get("tolerance_pct") or 0)
+        if rule_id == "CV002" or not rule.get("fields"):
+            cif = self._as_float(self._scope_value(context["inv"], "cif_value"))
+            fob = self._as_float(self._scope_value(context["inv"], "fob_value"))
+            freight = self._as_float(self._scope_value(context["inv"], "freight_value"))
+            insurance = self._as_float(self._scope_value(context["inv"], "insurance_value"))
+            if None in (cif, fob, freight, insurance) or not cif:
+                return False
+            diff_pct = abs(cif - (fob + freight + insurance)) / cif * 100
+            return diff_pct <= tolerance_pct
+
+        for field in rule.get("fields") or []:
+            values = [
+                self._as_float(self._scope_value(context[scope_name], field))
+                for scope_name in ("bl", "pl", "inv")
+            ]
+            present = [value for value in values if value is not None]
+            if len(present) < 2:
+                return False
+            baseline = present[0]
+            if baseline == 0:
+                return all(value == 0 for value in present)
+            if any(abs(value - baseline) / abs(baseline) * 100 > tolerance_pct for value in present[1:]):
+                return False
+        return True
+
+    def _evaluate_lookup_rule(self, rule: dict[str, Any], context: dict[str, Any]) -> bool:
+        for field in rule.get("fields") or []:
+            value = str(self._first_context_value(context, field) or "")
+            if field in {"kodePelabuhanMuat", "kodePelabuhanBongkar"} and not re.search(r"\b[A-Z]{5}\b", value):
+                return False
+        return True
+
+    def _as_float(self, value: Any) -> float | None:
+        if value in (None, "", MISSING):
+            return None
+        try:
+            return float(str(value).replace(",", ""))
+        except (TypeError, ValueError):
+            return None
 
     def _build_context(self, state: dict[str, Any]) -> dict[str, Any]:
         combined = state.get("combined_data") or {}
@@ -286,16 +436,26 @@ class ValidationRulesService:
             return template
 
     def _failure_severity(self, severity: str | None) -> str:
-        return "CRITICAL_FAIL" if severity == "CRITICAL" else "WARNING"
+        return "CRITICAL_FAIL" if severity in {"CRITICAL", "ERROR"} else "WARNING"
+
+    def _legacy_failure_severity(self, rule_id: str, severity: str | None) -> str:
+        if rule_id in {"CV001", "CV002", "CV003", "CV006", "CV008"}:
+            return "CRITICAL_FAIL"
+        return self._failure_severity(severity)
 
     def _resolve_rules_path(self) -> Path:
         configured = Path(settings.VALIDATION_RULES_PATH)
         candidates = [
             configured,
             Path.cwd() / configured,
-            Path(__file__).resolve().parents[4] / configured,
             Path("/app/validation_rules.json"),
         ]
+        
+        try:
+            candidates.append(Path(__file__).resolve().parents[4] / configured)
+        except IndexError:
+            pass
+
         for candidate in candidates:
             if candidate.exists():
                 return candidate.resolve()

src/tasks/ocr_tasks.py

@@ -98,31 +98,49 @@ async def _persist_graph_result(batch_id: str, result: dict) -> None:
         await supabase.table("batches").update({
             "status": status,
             "risk_level": result.get("risk_level"),
-            "customs_readiness_score": result.get("_crs_score"),
-            "crs_grade": result.get("_crs_grade"),
-            "rejection_probability": result.get("_rejection_prob"),
+            "customs_readiness_score": result.get("customs_readiness_score", result.get("_crs_score")),
+            "crs_grade": result.get("crs_grade", result.get("_crs_grade")),
+            "rejection_probability": result.get("rejection_probability", result.get("_rejection_prob")),
             "langgraph_thread_id": batch_id,
         }).eq("id", batch_id).execute()
     finally:
         pass
 
 
+async def _update_batch_status(batch_id: str, status: str, error_message: str | None = None) -> None:
+    """Best-effort status update for the upload/detail UI."""
+    from supabase import acreate_client
+
+    from ..config import settings
+
+    supabase = await acreate_client(settings.SUPABASE_URL, settings.SUPABASE_SERVICE_KEY.get_secret_value())
+    payload = {"status": status}
+    try:
+        await supabase.table("batches").update(payload).eq("id", batch_id).execute()
+    except Exception as exc:
+        log.warning(
+            "Failed to update batch status",
+            batch_id=batch_id,
+            status=status,
+            pipeline_error=error_message,
+            error=str(exc),
+        )
+
+
 def _average_confidence(confidences: dict) -> float:
     values = [float(value) for value in confidences.values()]
     return round(sum(values) / len(values), 4) if values else 0.0
 
 
-@celery_app.task(bind=True, queue="high", max_retries=3, default_retry_delay=10)
-def preprocess_document(self, batch_id: str) -> None:
-    """Entry point: kicks off LangGraph extraction pipeline."""
-    import asyncio
-
+def run_preprocess_pipeline_sync(batch_id: str, mark_error_on_failure: bool = False) -> None:
+    """Run the extraction pipeline in-process when Celery is unavailable."""
     from ..ai.graph import extraction_graph
-    log.info("Starting extraction pipeline", batch_id=batch_id)
+    from .celery_app import get_worker_loop
+
+    loop = get_worker_loop()
     try:
-        from .celery_app import get_worker_loop
-        loop = get_worker_loop()
         config = {"configurable": {"thread_id": batch_id}}
+        loop.run_until_complete(_update_batch_status(batch_id, "ocr_running"))
         context = loop.run_until_complete(_load_batch_context(batch_id))
         initial_state = {
             "batch_id": batch_id,
@@ -132,18 +150,38 @@ def preprocess_document(self, batch_id: str) -> None:
             "validation_results": [],
             "needs_human_review": False,
             "risk_level": "UNKNOWN",
+            "customs_readiness_score": None,
+            "crs_grade": None,
+            "rejection_probability": None,
+            "risk_features": {},
             "ocr_conflicts": [],
             "field_confidences": {},
             "steps": [],
         }
-        # Run sync wrapper around async graph
         result = loop.run_until_complete(
             extraction_graph.ainvoke(initial_state, config=config)
         )
         loop.run_until_complete(_persist_graph_result(batch_id, result))
+    except Exception as exc:
+        if mark_error_on_failure:
+            loop.run_until_complete(_update_batch_status(batch_id, "error", str(exc)))
+        raise
+
+
+@celery_app.task(bind=True, queue="high", max_retries=3, default_retry_delay=10)
+def preprocess_document(self, batch_id: str) -> None:
+    """Entry point: kicks off LangGraph extraction pipeline."""
+    log.info("Starting extraction pipeline", batch_id=batch_id)
+    try:
+        run_preprocess_pipeline_sync(batch_id)
         log.info("Extraction pipeline complete", batch_id=batch_id)
     except Exception as exc:
         log.error("Pipeline failed", batch_id=batch_id, error=str(exc))
+        if self.request.retries >= self.max_retries:
+            from .celery_app import get_worker_loop
+
+            loop = get_worker_loop()
+            loop.run_until_complete(_update_batch_status(batch_id, "error", str(exc)))
         raise self.retry(exc=exc)