fix: sync cloud deployment snapshot with verified backend path

src/ai/graph.py

@@ -32,6 +32,13 @@ 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"
         

src/ai/nodes/extract.py

@@ -83,29 +83,44 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
 
         # Initialize LLM on first real document that needs extraction
         if structured_llm is None:
-            if settings.DETERMINISTIC_E2E:
-                if DeterministicLLM is None:
-                    raise RuntimeError("DETERMINISTIC_E2E enabled but DeterministicLLM not available")
-                llm = DeterministicLLM()
-            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,
-                    temperature=0,
-                    api_key=settings.GEMINI_API_KEY
+            try:
+                if settings.DETERMINISTIC_E2E:
+                    if DeterministicLLM is None:
+                        raise RuntimeError("DETERMINISTIC_E2E enabled but DeterministicLLM not available")
+                    llm = DeterministicLLM()
+                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,
+                        temperature=0,
+                        api_key=settings.GEMINI_API_KEY
+                    )
+                    llm = primary_llm.with_fallbacks([fallback_llm])
+
+                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__,
                 )
-                llm = primary_llm.with_fallbacks([fallback_llm])
-
-            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
+                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(
@@ -183,13 +198,20 @@ async def llm_extraction_node(state: ExtractionGraphState) -> dict:
                 "ocr_method": "failed"
             })
         except Exception as e:
-            # Unexpected errors — log and re-raise to fail the batch
-            log.critical(
+            # Unexpected errors are real production failures and should surface
+            # so the batch can be diagnosed instead of silently degrading.
+            log.exception(
                 "Unexpected error in LLM extraction — batch will fail",
                 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_failed"
+            })
             raise
 
     return {

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] = ["*"]
+    CORS_ORIGINS: list[str] | str = ["*"]
 
     # ── Database ──────────────────────────────────────────────────────────────
     DATABASE_URL: str  # asyncpg connection string e.g. postgresql+asyncpg://...
@@ -68,9 +68,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_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-document"
+    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"
@@ -113,9 +114,10 @@ class Settings(BaseSettings):
     LANGCHAIN_API_KEY: SecretStr = ""  # type: ignore[assignment]
 
     # ── Feature Flags ─────────────────────────────────────────────────────────
-    ENABLE_SURYA_AGENT: bool = True
-    ENABLE_AZURE_DI_AGENT: bool = True
-    ENABLE_VESSEL_VALIDATION: bool = True
+    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_BLOCKCHAIN: bool = True  # type: ignore[assignment] — redeclared intentionally
     ENABLE_INSW_CHECK: bool = True
     ENABLE_NOTIFICATIONS_WHATSAPP: bool = False
@@ -128,10 +130,12 @@ class Settings(BaseSettings):
     DETERMINISTIC_E2E: bool = False  # Used in extract.py to swap LLM for deterministic mock
 
     # ── Thresholds ────────────────────────────────────────────────────────────
-    OCR_MAX_RENDERED_PAGES: int = 10
-    OCR_MAX_LLM_PAGES: int = 5
-    OCR_FAST_PATH_QUALITY_THRESHOLD: float = 0.95
-    OCR_RECONCILIATION_DISAGREEMENT_THRESHOLD: float = 0.20
+    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_RECONCILIATION_DISAGREEMENT_THRESHOLD: float = 0.20
     LLM_CONFIDENCE_REVIEW_THRESHOLD: float = 0.70
     CRS_MIN_SUBMIT_THRESHOLD: int = 55
     HS_CONFIDENCE_RAG_THRESHOLD: float = 0.75

src/dependencies.py

@@ -54,10 +54,17 @@ async def close_supabase() -> None:
 
 def get_supabase() -> AsyncClient:
     if _supabase_client is None:
-        raise RuntimeError("Supabase client not initialized. Call init_supabase() first.")
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Supabase client is not initialized. Check SUPABASE_URL and SUPABASE_SERVICE_KEY.",
+        )
     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
+from .dependencies import close_supabase, init_supabase, is_supabase_initialized
 from .routers import admin, batches, blockchain, hs_recommend, vessel
 from .utils.telemetry import setup_telemetry
 
@@ -120,6 +120,38 @@ 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

@@ -178,7 +178,11 @@ async def list_batches(
         .range(offset, offset + limit - 1)
         .execute()
     )
-    return {"batches": res.data, "total": len(res.data)}
+    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)}
 
 
 @router.get("/batches/{batch_id}")
@@ -198,11 +202,15 @@ 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_res.data,
+        "extracted_fields": fields,
+        "reconciled_fields": reconciled_fields,
         "validation_results": validations_res.data,
     }
 
@@ -289,3 +297,72 @@ 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/services/ocr_engine_svc.py

@@ -112,13 +112,30 @@ class OCREngineService:
         candidates: dict[str, dict[str, Any]] = {}
 
         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": "",
-                "ocr_candidates": {},
-                "quality_score": 1.0,
-                "ocr_engine_latencies_ms": {}
+                "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()
+                },
             }
 
         pdf_task = None
@@ -126,7 +143,7 @@ class OCREngineService:
             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 settings.ENABLE_DUAL_OCR else None
+        azure_task = self._run_azure(file_bytes, mime_type) if self._azure_enabled() else None
 
         gather_tasks = [paddle_task]
         if azure_task is not None:
@@ -274,7 +291,7 @@ class OCREngineService:
             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 settings.ENABLE_DUAL_OCR:
+        if not self._azure_enabled():
             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")
@@ -333,6 +350,9 @@ 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

start.sh

@@ -6,6 +6,7 @@ redis-server --daemonize yes
 sleep 2
 
 # Start Celery worker in the background
+export C_FORCE_ROOT=true
 celery -A src.tasks.ocr_tasks worker --loglevel=info &
 
 # Start FastAPI