Deploy TradeFlow API to HF

Dockerfile

@@ -0,0 +1,24 @@
+FROM python:3.13-slim-bookworm
+
+WORKDIR /app
+
+# Fix Debian GPG key rotation, install uv, and system dependencies
+RUN pip install uv && \
+    apt-get update -o Acquire::AllowInsecureRepositories=true 2>/dev/null; \
+    apt-get install -y --allow-unauthenticated debian-archive-keyring 2>/dev/null; \
+    apt-get update && apt-get install -y --no-install-recommends \
+    libgl1 \
+    libglib2.0-0 \
+    libmagic1 \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install dependencies using uv (lockfile ensures deterministic resolution)
+COPY pyproject.toml uv.lock ./
+RUN uv pip install --system -r pyproject.toml
+
+# Copy source code
+COPY . .
+
+# Run FastAPI via Uvicorn
+CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "7860", "--proxy-headers", "--forwarded-allow-ips", "*"]

LOCAL_SERVICES.md

@@ -0,0 +1,67 @@
+Local services required for full integration testing
+
+Overview
+- Unit tests run isolated with mocks; no services needed.
+- For end-to-end / integration you should run the services below.
+
+Core services (recommended minimal set)
+- redis: Used by Celery broker and LangGraph `RedisSaver` checkpointer.
+- minio: S3-compatible object storage when `STORAGE_BACKEND=minio`.
+- supabase (postgres + storage + realtime + rest + kong): Optional but required if using Supabase storage or DB-backed features.
+- keycloak: OIDC provider used for authentication (tests mock `get_current_user`).
+- chromadb: Vector DB used by HS-code embeddings (optional unless running embedding flows).
+
+Starting the minimal set with docker-compose (from repo root):
+
+```powershell
+# From repository root
+docker-compose up -d redis minio chromadb supabase-db supabase-storage supabase-realtime supabase-rest supabase-kong keycloak
+```
+
+Notes & env vars
+- The repo `docker-compose.yml` already configures sensible defaults. Override with environment variables in `.env` at `apps/api/.env` or repo root.
+- Typical overrides:
+  - `POSTGRES_PASSWORD` (postgres/supabase)
+  - `MINIO_ROOT_USER` / `MINIO_ROOT_PASSWORD`
+  - `SUPABASE_JWT_SECRET`
+  - `KEYCLOAK_ADMIN` / `KEYCLOAK_ADMIN_PASSWORD`
+
+Quick checks
+- Redis available at `redis://localhost:6379` (or `redis://redis:6379` inside Docker network).
+- Minio console: http://localhost:9001, API: http://localhost:9000
+- Supabase storage API: http://localhost:5000 (configured in compose)
+- Keycloak admin console: http://localhost:8080
+
+Integration test tips
+- Many tests mock Supabase and Keycloak; only enable real services when running integration/e2e tests.
+- If you want a minimal integration run, start `redis` and `minio` first and set `STORAGE_BACKEND=minio` in your `.env` before running the API.
+
+Running integration tests
+- Start the minimal services with the helper script from the repo root:
+
+```powershell
+.\scripts\integration-up.ps1
+```
+
+- Run only integration-marked tests:
+
+```powershell
+cd apps/api
+.venv\Scripts\python.exe -m pytest -m integration -q
+```
+
+- A dedicated GitHub Actions workflow is available at `.github/workflows/integration.yml` for manual or main-branch integration runs.
+
+- Run unit tests (fast, uses mocks):
+
+```powershell
+cd apps/api
+.venv\Scripts\python.exe -m pytest tests/ -q
+```
+
+CI recommendations
+- Run unit tests on every push; run integration tests in a separate CI job that brings up services (docker-compose or Testcontainers) and runs `pytest -m integration`.
+- Use `pytest.ini` to declare markers (already added at `apps/api/pytest.ini`).
+
+Next steps
+- I can run a minimal `docker-compose up` for you and then run integration tests, or add a `docker-compose.integration.yml` with a trimmed service list. Which would you prefer?

pyproject.toml

@@ -0,0 +1,142 @@
+[project]
+name = "tradeflow-api"
+version = "1.0.0"
+description = "TradeFlow AI — FastAPI Backend"
+requires-python = ">=3.13,<3.14"  # paddlepaddle does not yet have cp314 wheels
+dependencies = [
+    # Web framework
+    "fastapi>=0.115.6",
+    "uvicorn[standard]>=0.34.0",
+    "python-multipart>=0.0.20",
+    "slowapi>=0.1.9",  # Rate limiting
+
+    # Config & validation
+    "pydantic>=2.10.6",
+    "pydantic-settings>=2.7.1",
+
+    # Auth — Keycloak JWT validation
+    "python-jose[cryptography]>=3.3.0",
+    "httpx>=0.28.1",
+
+    # Database
+    "asyncpg>=0.30.0",
+    "sqlalchemy[asyncio]>=2.0.40",
+    "supabase>=2.13.0",
+
+    # Storage
+    "boto3>=1.37.8",         # MinIO (S3-compatible)
+    "minio>=7.2.13",
+    "python-magic>=0.4.27",  # File type validation
+
+    # Task queue
+    "celery>=5.5.0",
+    "kombu>=5.5.2",
+    "redis>=5.2.1",
+
+    # AI / LLM
+    "google-generativeai>=0.8.5",
+    "langchain-google-genai>=2.1.0",
+    "langgraph>=0.3.18",
+    "langgraph-checkpoint-redis>=0.0.6",
+    "langsmith>=0.3.11",
+    "openai>=1.58.1",         # text-embedding-3-small
+
+    # OCR & document processing
+    "paddleocr>=2.9.1,<3.0",
+    "paddlepaddle>=2.6.2,<3.4",  # cp313 wheels only; 3.3.1 is latest tested
+    "pymupdf>=1.25.3",
+    "pdfplumber>=0.11.4",
+    "openpyxl>=3.1.0",
+    "opencv-python-headless>=4.11.0.86",
+    "pillow>=11.1.0",
+    "lingua-language-detector>=2.0.2",
+
+    # ML
+    "xgboost>=2.1.4",
+    "scikit-learn>=1.6.1",
+    "joblib>=1.4.2",
+    "numpy>=2.2.3",
+    "pandas>=2.2.3",
+
+    # Vector store
+    "chromadb>=0.6.3",
+
+    # Azure DI (fallback OCR)
+    "azure-ai-documentintelligence>=1.0.0",
+    "azure-core>=1.32.0",
+
+    # Blockchain
+    "web3>=7.8.0",
+    "eth-account>=0.13.4",
+
+    # Notifications
+    "resend>=2.7.0",
+
+    # Observability
+    "sentry-sdk[fastapi]>=2.20.0",
+    "opentelemetry-sdk>=1.30.0",
+    "opentelemetry-instrumentation-fastapi>=0.51b0",
+    "opentelemetry-instrumentation-celery>=0.51b0",
+    "opentelemetry-instrumentation-httpx>=0.51b0",
+    "opentelemetry-instrumentation-redis>=0.51b0",
+    "prometheus-fastapi-instrumentator>=7.0.2",
+
+    # Utilities
+    "circuitbreaker>=2.0.0",
+    "cryptography>=44.0.0",
+    "python-dateutil>=2.9.0",
+    "pytz>=2025.1",
+    "tenacity>=9.0.0",
+    "structlog>=24.4.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.3.4",
+    "pytest-asyncio>=0.25.3",
+    "pytest-cov>=6.0.0",
+    "httpx>=0.28.1",
+    "ruff>=0.9.4",
+    "mypy>=1.14.1",
+    "factory-boy>=3.3.1",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.ruff]
+target-version = "py313"
+line-length = 120
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "W", "UP", "S", "B", "C4", "SIM"]
+ignore = [
+    "E501",   # allow lines longer than line-length
+    "S101",   # allow assert in tests
+    "N802",   # allow uppercase function names (pydantic property KEYCLOAK_JWKS_URL)
+    "N803",   # allow uppercase argument names (ML convention: X)
+    "N806",   # allow uppercase variable names (ML convention: X_train, X_valid)
+    "B008",   # allow function calls in defaults (FastAPI Depends/File/Form pattern)
+    "S108",   # allow /tmp paths (controlled usage in predictor_svc)
+    "S110",   # allow try-except-pass (fire-and-forget status updates)
+    "SIM105", # allow explicit try-except-pass over contextlib.suppress
+    "SIM108", # allow if-else blocks over ternary (readability)
+]
+
+[tool.ruff.lint.per-file-ignores]
+"src/tasks/*.py" = ["B904"]  # Celery self.retry() cannot use raise-from
+"src/auth/keycloak.py" = ["B904"]
+"src/dependencies.py" = ["B904"]
+
+[tool.mypy]
+python_version = "3.13"
+strict = true
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src"]

pytest.ini

@@ -0,0 +1,6 @@
+[pytest]
+minversion = 7.0
+addopts = -ra
+testpaths = tests
+markers =
+    integration: mark test as integration which requires external services

src/__init__.py

@@ -0,0 +1 @@
+"""TradeFlow AI — API package init."""

src/ai/graph.py

@@ -0,0 +1,131 @@
+"""
+TradeFlow AI — LangGraph Extraction Graph (Step 2 Assembly)
+
+PRD §10 — Full LangGraph pipeline:
+
+  preprocess → llm_extraction → [fallback if needed] → validate
+    → risk_assessment → [interrupt if review needed] → DONE
+
+The graph is compiled with a Redis checkpointer for persistence
+and resumability across server restarts.
+"""
+
+from __future__ import annotations
+
+import redis
+import structlog
+from langgraph.checkpoint.redis import RedisSaver
+from langgraph.graph import END, StateGraph
+
+from ..config import settings
+from .nodes.extract import llm_extraction_node
+from .nodes.fallback_ocr import fallback_ocr_node
+from .nodes.human_review import human_review_node
+from .nodes.preprocess import preprocess_documents_node
+from .nodes.risk import risk_assessment_node
+from .nodes.validate import validation_node
+from .state import ExtractionGraphState
+
+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:
+        log.info("CLOUD_LLM_ONLY is active — bypassing heavy OCR ensemble fallback")
+        return "validate"
+        
+    for doc in state.get("documents", []):
+        if doc.get("error") or not doc.get("extracted_data"):
+            return "fallback"
+        if doc.get("quality_score", 1.0) < settings.OCR_FALLBACK_TRIGGER_QUALITY:
+            return "fallback"
+        confidences = doc.get("field_confidences") or {}
+        if confidences and min(confidences.values()) < settings.OCR_FALLBACK_TRIGGER_CONFIDENCE:
+            return "fallback"
+        if doc.get("ocr_conflicts"):
+            return "fallback"
+        if len(doc.get("ocr_candidates") or {}) > 1:
+            return "fallback"
+    return "validate"
+
+
+def _needs_review(state: ExtractionGraphState) -> str:
+    """Conditional edge: route to human review if flagged."""
+    if state.get("needs_human_review", False):
+        return "human_review"
+    return END
+
+
+def build_extraction_graph() -> StateGraph:
+    """Build and compile the LangGraph extraction pipeline."""
+    workflow = StateGraph(ExtractionGraphState)
+
+    # ── Add nodes ────────────────────────────────────────────────
+    workflow.add_node("preprocess",      preprocess_documents_node)
+    workflow.add_node("llm_extraction",  llm_extraction_node)
+    workflow.add_node("fallback_ocr",    fallback_ocr_node)
+    workflow.add_node("validate",        validation_node)
+    workflow.add_node("risk_assessment", risk_assessment_node)
+    workflow.add_node("human_review",    human_review_node)
+
+    # ── Entry point ───────────────────────────────────────────────
+    workflow.set_entry_point("preprocess")
+
+    # ── Edges ─────────────────────────────────────────────────────
+    workflow.add_edge("preprocess", "llm_extraction")
+
+    # After extraction: check if fallback needed
+    workflow.add_conditional_edges(
+        "llm_extraction",
+        _needs_fallback,
+        {
+            "fallback": "fallback_ocr",
+            "validate": "validate",
+        },
+    )
+
+    # Fallback always proceeds to validate
+    workflow.add_edge("fallback_ocr", "validate")
+
+    # After validation: compute risk
+    workflow.add_edge("validate", "risk_assessment")
+
+    # After risk: check if human review needed
+    workflow.add_conditional_edges(
+        "risk_assessment",
+        _needs_review,
+        {
+            "human_review": "human_review",
+            END: END,
+        },
+    )
+
+    # After human review: graph ends (operator approved)
+    workflow.add_edge("human_review", END)
+
+    return workflow
+
+
+def get_compiled_graph():
+    """
+    Returns the compiled graph with Redis checkpointer.
+    The checkpointer enables:
+      - State persistence across Celery task restarts
+      - interrupt() resumability for human review
+      - LangSmith tracing integration
+    """
+    workflow = build_extraction_graph()
+
+    # Redis checkpointer — stores full graph state per thread_id (batch_id)
+    redis_conn = redis.Redis.from_url(settings.REDIS_URL)
+    checkpointer = RedisSaver(redis_client=redis_conn)
+
+    graph = workflow.compile(checkpointer=checkpointer, interrupt_before=["human_review"])
+
+    log.info("LangGraph extraction graph compiled", nodes=list(workflow.nodes.keys()))
+    return graph
+
+
+# ── Singleton (imported by Celery tasks) ──────────────────────────────────────
+extraction_graph = get_compiled_graph()

src/ai/mock_llm.py

@@ -0,0 +1,62 @@
+"""
+Deterministic LLM stub for deterministic E2E and tests.
+Returns predictable CEISAFields outputs so the LangGraph flow is repeatable.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class DeterministicStructuredLLM:
+    def __init__(self, model_schema: type[BaseModel]):
+        self._schema = model_schema
+
+    async def ainvoke(self, messages: Any):
+        # Return a deterministic instance matching the pydantic output schema
+        # Use simple fixed safe defaults; tests relying on presence of fields
+        # can assert these exact values for determinism.
+        data = {}
+        # Pydantic v2 uses `model_fields`; v1 uses `__fields__` with different metadata
+        schema_fields = getattr(self._schema, 'model_fields', None) or getattr(self._schema, '__fields__', {})
+        for k, meta in schema_fields.items():
+            # Determine annotation/type across pydantic versions
+            if isinstance(meta, dict):
+                ftype = meta.get('annotation')
+            elif hasattr(meta, 'annotation'):
+                ftype = meta.annotation
+            elif hasattr(meta, 'outer_type_'):
+                ftype = meta.outer_type_
+            else:
+                ftype = None
+
+            # Provide reasonable deterministic defaults by common types
+            if ftype is str or getattr(ftype, '__name__', '') == 'str':
+                data[k] = f"det-{k}"
+            elif ftype is int or getattr(ftype, '__name__', '') == 'int':
+                data[k] = 1
+            elif ftype is float or getattr(ftype, '__name__', '') == 'float':
+                data[k] = 1.0
+            else:
+                data[k] = None
+
+        # Create a pydantic model instance if possible
+        try:
+            return self._schema.model_validate(data) if hasattr(self._schema, 'model_validate') else self._schema(**data)
+        except Exception:
+            # Last resort: return raw dict
+            return data
+
+
+class DeterministicLLM:
+    def __init__(self, *_, **__):
+        pass
+
+    def with_structured_output(self, schema: type[BaseModel]):
+        return DeterministicStructuredLLM(schema)
+
+
+# synchronous convenience factory
+def create_deterministic_llm(*args, **kwargs):
+    return DeterministicLLM()

src/ai/nodes/extract.py

@@ -0,0 +1,199 @@
+"""
+TradeFlow AI — Primary LLM Extraction Node (Step 2.2)
+
+Uses Gemini 2.0 Flash Exp for multimodal extraction.
+"""
+
+import asyncio
+import inspect
+
+import structlog
+from pydantic import BaseModel, Field
+
+# Optional production LLM — may be absent in lightweight test environments
+try:
+    from langchain_google_genai import ChatGoogleGenerativeAI
+except Exception:  # pragma: no cover - optional dependency
+    ChatGoogleGenerativeAI = None
+
+from ...config import settings
+from ..state import ExtractionGraphState
+
+# Deterministic stub for tests/E2E
+if settings.DETERMINISTIC_E2E:
+    try:
+        from ..mock_llm import DeterministicLLM as DeterministicLLM  # type: ignore
+    except Exception:
+        DeterministicLLM = None
+else:
+    DeterministicLLM = None
+
+log = structlog.get_logger()
+
+# Structured output schema
+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")
+    cif_value: float | None = Field(description="Total CIF value")
+    currency: str | None = Field(description="Currency code (e.g. USD, IDR)")
+
+async def llm_extraction_node(state: ExtractionGraphState) -> dict:
+    """
+    Step 2.2: Primary LLM Extraction using Gemini 2.0 Flash Exp.
+
+    Args:
+        state: ExtractionGraphState with documents list containing doc_id, storage_path, pages
+
+    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
+    """
+    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.
+    structured_llm = None
+
+    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"):
+            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
+
+        # 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
+                )
+                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
+        # 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
+
+        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
+                        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"}
+                        ]
+                    )
+                ]
+
+            result = await structured_llm.ainvoke(messages)
+
+            # 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
+
+            candidates = dict(doc.get("ocr_candidates") or {})
+            candidates[settings.GEMINI_MODEL_PRIMARY] = {
+                "fields": extracted,
+                "confidence": 0.82,
+            }
+
+            updated_docs.append({
+                **doc,
+                "extracted_data": extracted,
+                "ocr_method": settings.GEMINI_MODEL_PRIMARY,
+                "ocr_candidates": candidates,
+                "field_confidences": dict.fromkeys(extracted, 0.82),
+            })
+
+            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 — log and re-raise to fail the batch
+            log.critical(
+                "Unexpected error in LLM extraction — batch will fail",
+                doc_id=doc.get("doc_id"),
+                batch_id=state["batch_id"],
+                error_type=type(e).__name__
+            )
+            raise
+
+    return {
+        "documents": updated_docs,
+        "combined_data": combined_data,
+        "steps": ["llm_extraction"]
+    }

src/ai/nodes/fallback_ocr.py

@@ -0,0 +1,116 @@
+"""
+TradeFlow AI — Fallback OCR Node (Step 2.3)
+
+Used when Gemini extraction fails or confidence is too low.
+"""
+
+import re
+
+import structlog
+
+from ...config import settings
+from ...services.ocr_conflict_svc import reconcile_ocr_candidates
+from ..state import ExtractionGraphState
+
+log = structlog.get_logger()
+
+
+def _needs_reconciliation(doc: dict) -> bool:
+    confidences = doc.get("field_confidences") or {}
+    return (
+        bool(doc.get("error"))
+        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 bool(
+            confidences
+            and min(confidences.values()) < settings.OCR_FALLBACK_TRIGGER_CONFIDENCE
+        )
+    )
+
+
+def _rule_based_candidates(doc: dict) -> dict:
+    """Extract obvious CEISA fields from embedded text without external OCR."""
+    text = "\n".join(str(page) for page in doc.get("pages", []) if isinstance(page, str))
+    text = "\n".join([doc.get("raw_text", ""), doc.get("text_layer", ""), text])
+
+    fields = {}
+    npwp = re.search(r"\b(?:NPWP|Tax\s*ID)\D*([0-9.\- ]{10,24})", text, re.IGNORECASE)
+    packages = re.search(r"\b(?:total\s+packages|packages|koli)\D*(\d{1,7})", text, re.IGNORECASE)
+    gross = re.search(r"\b(?:gross\s+weight|gross)\D*([0-9,.]+)", text, re.IGNORECASE)
+    cif = re.search(
+        r"\b(?:CIF|total\s+amount|invoice\s+value)\D*([A-Z]{3})?\s*([0-9,.]+)",
+        text,
+        re.IGNORECASE,
+    )
+
+    if npwp:
+        fields["importer_npwp"] = npwp.group(1)
+    if packages:
+        fields["total_packages"] = int(packages.group(1))
+    if gross:
+        fields["gross_weight"] = float(gross.group(1).replace(",", ""))
+    if cif:
+        if cif.group(1):
+            fields["currency"] = cif.group(1).upper()
+        fields["cif_value"] = float(cif.group(2).replace(",", ""))
+
+    return {"fields": fields, "confidence": 0.68}
+
+
+async def fallback_ocr_node(state: ExtractionGraphState) -> dict:
+    """
+    Step 2.3: Fallback OCR using Azure Document Intelligence or PaddleOCR.
+    """
+    log.info("Running fallback_ocr_node", batch_id=state["batch_id"])
+
+    updated_docs = []
+    all_conflicts = list(state.get("ocr_conflicts", []))
+    combined_data = dict(state.get("combined_data", {}))
+    needs_review = state.get("needs_human_review", False)
+    field_confidences = dict(state.get("field_confidences", {}))
+
+    for doc in state["documents"]:
+        if _needs_reconciliation(doc):
+            log.info("Reconciling fallback OCR candidates for doc", doc_id=doc["doc_id"])
+            candidates = dict(doc.get("ocr_candidates") or {})
+
+            if doc.get("extracted_data"):
+                candidates[doc.get("ocr_method") or "gemini"] = {
+                    "fields": doc["extracted_data"],
+                    "confidence": 0.82,
+            }
+            if settings.ENABLE_DUAL_OCR and "azure-di" not in candidates:
+                log.warning(
+                    "Azure DI candidate missing; preserving degraded OCR evidence",
+                    doc_id=doc["doc_id"],
+                )
+            rule_candidate = _rule_based_candidates(doc)
+            if rule_candidate["fields"]:
+                candidates["rule_based"] = rule_candidate
+
+            reconciled = reconcile_ocr_candidates(candidates)
+            doc["ocr_method"] = "ensemble-reconciled"
+            doc["extracted_data"] = reconciled["fields"]
+            doc["field_confidences"] = reconciled["field_confidences"]
+            doc["ocr_conflicts"] = reconciled["conflicts"]
+            doc["error"] = None if reconciled["fields"] else "No OCR engine produced usable fields"
+
+            combined_data.update(reconciled["fields"])
+            field_confidences.update(reconciled["field_confidences"])
+            all_conflicts.extend(
+                {**conflict, "doc_id": doc["doc_id"]} for conflict in reconciled["conflicts"]
+            )
+            needs_review = needs_review or reconciled["needs_human_review"] or bool(doc["error"])
+
+        updated_docs.append(doc)
+
+    return {
+        "documents": updated_docs,
+        "combined_data": combined_data,
+        "field_confidences": field_confidences,
+        "ocr_conflicts": all_conflicts,
+        "needs_human_review": needs_review,
+        "steps": ["fallback_ocr"]
+    }

src/ai/nodes/human_review.py

@@ -0,0 +1,59 @@
+"""
+TradeFlow AI — Human Review Interrupt Node (Step 2.6)
+
+PRD §0.2 Invariant #5: LangGraph's interrupt() is the ONLY mechanism
+to pause execution for human review. No polling loops.
+"""
+
+from __future__ import annotations
+
+import structlog
+from langgraph.types import interrupt
+
+from ..state import ExtractionGraphState
+
+log = structlog.get_logger()
+
+
+async def human_review_node(state: ExtractionGraphState) -> dict:
+    """
+    Pause the graph and surface extracted fields for operator review.
+
+    The graph will resume when an operator POSTs to
+    POST /api/v1/batches/{batch_id}/review with their corrections.
+
+    `interrupt()` saves full graph state to Redis checkpoint store.
+    """
+    log.info(
+        "Interrupting for human review",
+        batch_id=state["batch_id"],
+        risk_level=state.get("risk_level"),
+    )
+
+    # Surfaces the current extraction output to the operator
+    review_payload = {
+        "batch_id": state["batch_id"],
+        "combined_data": state.get("combined_data", {}),
+        "validation_results": state.get("validation_results", []),
+        "risk_level": state.get("risk_level", "UNKNOWN"),
+        "message": "Dokumen ini memerlukan tinjauan manual sebelum dapat diajukan.",
+    }
+
+    # PRD §10: interrupt() — execution suspends here.
+    # The operator's corrections are returned as the `interrupt` return value.
+    corrections: dict = interrupt(review_payload)
+
+    # Merge operator corrections into combined_data
+    corrected_data = {**state.get("combined_data", {}), **corrections}
+
+    log.info(
+        "Human review complete — resuming graph",
+        batch_id=state["batch_id"],
+        corrections_count=len(corrections),
+    )
+
+    return {
+        "combined_data": corrected_data,
+        "needs_human_review": False,
+        "steps": ["human_review"],
+    }

src/ai/nodes/preprocess.py

@@ -0,0 +1,68 @@
+"""
+TradeFlow AI — Preprocessing Node (Step 2.1)
+"""
+
+import structlog
+
+from ...services.ingest_svc import get_storage_service
+from ...services.ocr_engine_svc import ocr_engine_service
+from ..state import ExtractionGraphState
+
+log = structlog.get_logger()
+
+async def preprocess_documents_node(state: ExtractionGraphState) -> dict:
+    """
+    Step 2.1: Document Preprocessing Node
+    - Checks document quality
+    - Converts PDFs to images if necessary
+    - Sets quality score
+    """
+    log.info("Running preprocess_documents_node", batch_id=state["batch_id"])
+
+    updated_docs = []
+    for doc in state["documents"]:
+        storage_path = doc.get("storage_path")
+        if not storage_path:
+            updated_docs.append({
+                **doc,
+                "quality_score": 0.0,
+                "pages": [],
+                "ocr_candidates": {},
+                "error": "Document missing storage_path",
+            })
+            continue
+
+        try:
+            file_bytes = await get_storage_service().download_document(storage_path)
+            prepared = await ocr_engine_service.prepare_document(
+                doc_id=doc["doc_id"],
+                storage_path=storage_path,
+                filename=doc.get("original_name") or storage_path,
+                file_bytes=file_bytes,
+            )
+        except Exception as exc:
+            log.exception(
+                "Document preprocessing/OCR failed",
+                batch_id=state["batch_id"],
+                doc_id=doc.get("doc_id"),
+                error=str(exc),
+            )
+            updated_docs.append({
+                **doc,
+                "quality_score": 0.0,
+                "pages": [],
+                "ocr_candidates": {},
+                "error": str(exc),
+            })
+            continue
+
+        updated_docs.append({
+            **doc,
+            **prepared,
+            "ocr_method": "+".join(prepared["ocr_candidates"].keys()) or None,
+        })
+
+    return {
+        "documents": updated_docs,
+        "steps": ["preprocess"]
+    }

src/ai/nodes/risk.py

@@ -0,0 +1,159 @@
+"""
+TradeFlow AI — Risk Assessment Node (Step 2.5)
+
+Runs the XGBoost rejection predictor and computes the
+Customs Readiness Score (CRS) for a batch.
+
+PRD §13 — CRS = weighted average across 5 pillars:
+  (1) Document Quality 20%
+  (2) Data Completeness 25%
+  (3) Cross-document Consistency 30%
+  (4) Historical Performance 15%
+  (5) HS Code Confidence 10%
+"""
+
+from __future__ import annotations
+
+import structlog
+
+from ...services.predictor_svc import rejection_predictor
+from ..state import ExtractionGraphState
+
+log = structlog.get_logger()
+
+# Pillar weights per PRD §13
+PILLAR_WEIGHTS = {
+    "doc_quality": 0.20,
+    "completeness": 0.25,
+    "consistency": 0.30,
+    "historical": 0.15,
+    "hs_confidence": 0.10,
+}
+
+REQUIRED_CEISA_FIELDS = [
+    "importer_name", "importer_npwp", "total_packages",
+    "gross_weight", "cif_value", "currency",
+]
+
+
+def _compute_completeness(combined_data: dict) -> float:
+    filled = sum(1 for f in REQUIRED_CEISA_FIELDS if combined_data.get(f))
+    return filled / len(REQUIRED_CEISA_FIELDS)
+
+
+def _compute_consistency(validation_results: list[dict]) -> float:
+    if not validation_results:
+        return 1.0
+    passed = sum(1 for r in validation_results if r.get("severity") == "PASS")
+    return passed / len(validation_results)
+
+
+def _compute_doc_quality(documents: list[dict]) -> float:
+    scores = [d.get("quality_score", 0.8) for d in documents]
+    return sum(scores) / len(scores) if scores else 0.0
+
+
+def _crs_to_grade(score: float) -> str:
+    if score >= 90:
+        return "A"
+    if score >= 80:
+        return "B"
+    if score >= 70:
+        return "C"
+    if score >= 60:
+        return "D"
+    return "F"
+
+
+def _score_to_risk(score: float) -> str:
+    if score >= 80:
+        return "LOW"
+    if score >= 65:
+        return "MEDIUM"
+    if score >= 50:
+        return "HIGH"
+    return "CRITICAL"
+
+
+def _probability_to_risk(probability: float) -> str:
+    if probability < 0.15:
+        return "LOW"
+    if probability < 0.35:
+        return "MEDIUM"
+    if probability < 0.60:
+        return "HIGH"
+    return "CRITICAL"
+
+
+async def risk_assessment_node(state: ExtractionGraphState) -> dict:
+    """
+    Compute CRS (0-100) and rejection probability (0-1).
+
+    XGBoost inference uses the shared predictor service, with heuristic
+    fallback when no trained model is available yet.
+    """
+    log.info("Running risk_assessment_node", batch_id=state["batch_id"])
+
+    combined_data = state.get("combined_data", {})
+    validation_results = state.get("validation_results", [])
+    documents = state.get("documents", [])
+
+    # ── Pillar scores ──────────────────────────────────────────────
+    p_quality     = _compute_doc_quality(documents)
+    p_completeness = _compute_completeness(combined_data)
+    p_consistency  = _compute_consistency(validation_results)
+    p_historical   = 0.80  # Stub — fetched from company submission history
+    p_hs_conf      = 0.85  # Stub — from HS recommender confidence
+
+    # ── Weighted CRS ───────────────────────────────────────────────
+    crs_raw = (
+        p_quality      * PILLAR_WEIGHTS["doc_quality"]
+        + p_completeness * PILLAR_WEIGHTS["completeness"]
+        + p_consistency  * PILLAR_WEIGHTS["consistency"]
+        + p_historical   * PILLAR_WEIGHTS["historical"]
+        + p_hs_conf      * PILLAR_WEIGHTS["hs_confidence"]
+    )
+    crs_score = round(crs_raw * 100, 2)
+    crs_grade = _crs_to_grade(crs_score)
+
+    features = {
+        "doc_quality_score": p_quality,
+        "completeness_score": p_completeness,
+        "consistency_score": p_consistency,
+        "historical_rate": p_historical,
+        "hs_confidence": p_hs_conf,
+        "cif_value_usd": float(combined_data.get("cif_value") or 0.0),
+        "package_count": float(combined_data.get("total_packages") or 0.0),
+        "gross_weight_kg": float(combined_data.get("gross_weight") or 0.0),
+    }
+    rejection_prob = round(rejection_predictor.predict_proba(features), 4)
+    risk_level = _probability_to_risk(rejection_prob)
+
+    # PRD §13 Invariant: CRS < 70 → must NOT auto-submit
+    needs_human_review = (
+        state.get("needs_human_review", False)
+        or crs_score < 70.0
+        or rejection_prob >= 0.35
+    )
+
+    log.info(
+        "CRS computed",
+        batch_id=state["batch_id"],
+        crs=crs_score,
+        grade=crs_grade,
+        risk=risk_level,
+        rejection_prob=rejection_prob,
+    )
+
+    return {
+        "risk_level": risk_level,
+        "needs_human_review": needs_human_review,
+        "steps": ["risk_assessment"],
+        # NOTE: crs_score and rejection_prob are persisted to DB in the
+        # caller task (ocr_tasks.assess_risk), not stored in graph state
+        # to keep the state lean per PRD §0.2 Invariant #5.
+        "_crs_score": crs_score,
+        "_crs_grade": crs_grade,
+        "_rejection_prob": rejection_prob,
+        "_risk_features": features,
+    }

src/ai/nodes/validate.py

@@ -0,0 +1,24 @@
+"""
+TradeFlow AI — Cross-Document Validation Node (Step 2.4)
+"""
+
+import structlog
+
+from ...services.validation_rules_svc import validation_rules_service
+from ..state import ExtractionGraphState
+
+log = structlog.get_logger()
+
+async def validation_node(state: ExtractionGraphState) -> dict:
+    """
+    Step 2.4: Cross-Document Validation against JSON rules.
+    """
+    log.info("Running validation_node", batch_id=state["batch_id"])
+
+    results, needs_review = validation_rules_service.evaluate(state)
+
+    return {
+        "validation_results": results,
+        "needs_human_review": needs_review,
+        "steps": ["validation"]
+    }

src/ai/state.py

@@ -0,0 +1,34 @@
+"""
+TradeFlow AI — LangGraph State Definition
+"""
+
+from operator import add
+from typing import Annotated, TypedDict
+
+
+class DocumentState(TypedDict):
+    doc_id: str
+    doc_type: str
+    storage_path: str
+    pages: list[str]  # Base64 encoded images or temporary paths
+    extracted_data: dict | None
+    quality_score: float
+    ocr_method: str | None
+    error: str | None
+    ocr_candidates: dict
+    ocr_conflicts: list[dict]
+    field_confidences: dict[str, float]
+
+class ExtractionGraphState(TypedDict):
+    """The state of the document extraction LangGraph."""
+    batch_id: str
+    company_id: str
+    documents: list[DocumentState]
+    combined_data: dict
+    validation_results: list[dict]
+    needs_human_review: bool
+    risk_level: str
+    ocr_conflicts: list[dict]
+    field_confidences: dict[str, float]
+    # Keep track of which node executed
+    steps: Annotated[list[str], add]

src/auth/__init__.py

@@ -0,0 +1,26 @@
+"""Auth package init."""
+from .dependencies import (
+    CurrentUser,
+    RequireAdmin,
+    RequireOperator,
+    RequireSME,
+    RequireSupervisor,
+    get_current_user,
+    get_current_user_id,
+    require_roles,
+)
+from .keycloak import extract_roles, extract_user_id, verify_keycloak_token
+
+__all__ = [
+    "verify_keycloak_token",
+    "extract_roles",
+    "extract_user_id",
+    "get_current_user",
+    "get_current_user_id",
+    "require_roles",
+    "CurrentUser",
+    "RequireOperator",
+    "RequireAdmin",
+    "RequireSME",
+    "RequireSupervisor",
+]

src/auth/dependencies.py

@@ -0,0 +1,90 @@
+"""
+TradeFlow AI — FastAPI Auth Dependencies (T-009)
+
+Provides reusable dependency functions for role-based access control.
+All protected endpoints must use one of these dependencies.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated, Any
+
+from fastapi import Depends, HTTPException, Security, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from .keycloak import extract_roles, extract_user_id, verify_keycloak_token
+
+bearer_scheme = HTTPBearer(auto_error=True)
+
+
+async def get_current_token_payload(
+    credentials: Annotated[HTTPAuthorizationCredentials, Security(bearer_scheme)],
+) -> dict[str, Any]:
+    """Verify the Bearer JWT and return its decoded payload."""
+    return await verify_keycloak_token(credentials.credentials)
+
+
+async def get_current_user_id(
+    payload: Annotated[dict[str, Any], Depends(get_current_token_payload)],
+) -> str:
+    """Returns the authenticated user's Keycloak sub (UUID)."""
+    return extract_user_id(payload)
+
+
+async def get_current_roles(
+    payload: Annotated[dict[str, Any], Depends(get_current_token_payload)],
+) -> list[str]:
+    """Returns the list of Keycloak realm roles for the current user."""
+    return extract_roles(payload)
+
+
+def require_roles(*allowed_roles: str):
+    """
+    Dependency factory that enforces role-based access.
+
+    Usage:
+        @router.post("/submit")
+        async def submit(
+            _: None = Depends(require_roles("operator", "admin"))
+        ):
+    """
+
+    async def _check_roles(
+        roles: Annotated[list[str], Depends(get_current_roles)],
+    ) -> None:
+        if not any(role in roles for role in allowed_roles):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Required roles: {list(allowed_roles)}",
+            )
+
+    return Depends(_check_roles)
+
+
+# Convenience singletons for common role checks
+RequireOperator = require_roles("operator", "admin")
+RequireAdmin = require_roles("admin")
+RequireSME = require_roles("sme", "operator", "admin")
+RequireSupervisor = require_roles("supervisor", "admin")
+
+
+class CurrentUser:
+    """Dependency class bundling user_id + roles in one inject."""
+
+    def __init__(self, user_id: str, roles: list[str]) -> None:
+        self.user_id = user_id
+        self.roles = roles
+
+    def has_role(self, *roles: str) -> bool:
+        return any(r in self.roles for r in roles)
+
+    def is_admin(self) -> bool:
+        return "admin" in self.roles
+
+
+async def get_current_user(
+    user_id: Annotated[str, Depends(get_current_user_id)],
+    roles: Annotated[list[str], Depends(get_current_roles)],
+) -> CurrentUser:
+    """Returns a CurrentUser object with id and roles."""
+    return CurrentUser(user_id=user_id, roles=roles)

src/auth/keycloak.py

@@ -0,0 +1,128 @@
+"""
+TradeFlow AI — Keycloak 26 JWT Authentication (T-008)
+
+PRD Invariant #4: Keycloak 26 is the SOLE auth provider. No Supabase Auth.
+Validates JWTs via JWKS endpoint with a 5-minute cache.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import httpx
+from fastapi import HTTPException, status
+from jose import JWTError, jwk, jwt
+
+from ..config import settings
+
+# JWKS TTL — 5 minutes (SDD §4.1)
+_JWKS_TTL_SECONDS = 300
+
+
+class JWKSCache:
+    """Thread-safe JWKS cache with 5-minute TTL."""
+
+    def __init__(self) -> None:
+        self._keys: dict[str, Any] = {}
+        self._fetched_at: float = 0.0
+
+    def _is_stale(self) -> bool:
+        return time.monotonic() - self._fetched_at > _JWKS_TTL_SECONDS
+
+    async def get_keys(self) -> dict[str, Any]:
+        if not self._keys or self._is_stale():
+            await self._refresh()
+        return self._keys
+
+    async def _refresh(self) -> None:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.get(settings.KEYCLOAK_JWKS_URL)
+            resp.raise_for_status()
+            jwks_data = resp.json()
+
+        # Build kid → key mapping
+        self._keys = {}
+        for key_data in jwks_data.get("keys", []):
+            kid = key_data.get("kid")
+            if kid:
+                self._keys[kid] = jwk.construct(key_data)
+
+        self._fetched_at = time.monotonic()
+
+
+_jwks_cache = JWKSCache()
+
+
+async def verify_keycloak_token(token: str) -> dict[str, Any]:
+    """
+    Verify a Keycloak JWT token.
+
+    Returns the decoded payload (claims) on success.
+    Raises HTTP 401 on any failure.
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        # Step 1: Decode header to get kid without signature verification
+        unverified_header = jwt.get_unverified_header(token)
+    except JWTError:
+        raise credentials_exception
+
+    kid = unverified_header.get("kid")
+    if not kid:
+        raise credentials_exception
+
+    # Step 2: Get the signing key from JWKS cache
+    try:
+        keys = await _jwks_cache.get_keys()
+    except Exception:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Auth service temporarily unavailable",
+        )
+
+    signing_key = keys.get(kid)
+    if not signing_key:
+        # Key not found — JWKS may have rotated, force refresh
+        await _jwks_cache._refresh()
+        keys = await _jwks_cache.get_keys()
+        signing_key = keys.get(kid)
+        if not signing_key:
+            raise credentials_exception
+
+    # Step 3: Verify signature + claims
+    try:
+        payload = jwt.decode(
+            token,
+            signing_key,
+            algorithms=["RS256"],
+            audience=settings.KEYCLOAK_CLIENT_ID,
+            issuer=settings.KEYCLOAK_ISSUER,
+            options={"verify_exp": True},
+        )
+    except JWTError as e:
+        raise credentials_exception from e
+
+    return payload
+
+
+def extract_roles(payload: dict[str, Any]) -> list[str]:
+    """Extract realm-level roles from a decoded Keycloak token."""
+    realm_access = payload.get("realm_access", {})
+    return realm_access.get("roles", [])
+
+
+def extract_user_id(payload: dict[str, Any]) -> str:
+    """Extract the user UUID (Keycloak sub claim)."""
+    sub = payload.get("sub")
+    if not sub:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token missing subject claim",
+        )
+    return sub

src/config.py

@@ -0,0 +1,192 @@
+"""
+TradeFlow AI — Application Configuration (T-004)
+
+SDD §2.1 + PRD §22 Invariant: No bare os.getenv().
+ALL environment variables are validated here via pydantic-settings.
+This is the SINGLE source of truth for configuration.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from typing import Literal
+
+from pydantic import AnyHttpUrl, Field, SecretStr, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    # ── Application ───────────────────────────────────────────────────────────
+    ENVIRONMENT: Literal["development", "staging", "production"] = "development"
+    DEBUG: bool = False
+    SECRET_KEY: SecretStr = Field(..., min_length=32)
+    CORS_ORIGINS: list[str] = ["*"]
+
+    # ── Database ──────────────────────────────────────────────────────────────
+    DATABASE_URL: str  # asyncpg connection string e.g. postgresql+asyncpg://...
+
+    # ── Supabase ──────────────────────────────────────────────────────────────
+    SUPABASE_URL: str
+    SUPABASE_ANON_KEY: str
+    SUPABASE_SERVICE_KEY: SecretStr
+    SUPABASE_JWT_SECRET: SecretStr
+
+    # ── Keycloak 26 — SOLE auth provider (Invariant #4) ───────────────────────
+    KEYCLOAK_SERVER_URL: str
+    KEYCLOAK_REALM: str = "tradeflow"
+    KEYCLOAK_CLIENT_ID: str = "tradeflow-api"
+    KEYCLOAK_CLIENT_SECRET: SecretStr
+    KEYCLOAK_ISSUER: str
+
+    @property
+    def KEYCLOAK_JWKS_URL(self) -> str:
+        base = self.KEYCLOAK_SERVER_URL.rstrip("/")
+        return f"{base}/realms/{self.KEYCLOAK_REALM}/protocol/openid-connect/certs"
+
+    # ── Redis 8 Standalone ────────────────────────────────────────────────────
+    REDIS_URL: str = "redis://localhost:6379/0"
+    REDIS_CELERY_DB: int = 0
+    REDIS_CACHE_DB: int = 1
+
+    # ── AI Inference Services (SDD §2.3–2.6) ─────────────────────────────────
+    CLOUD_LLM_ONLY: bool = False                                    # Bypass heavy local models and use Gemini API instead
+    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
+    MINERU_SVC_URL: AnyHttpUrl = "http://mineru-svc:8003"           # Preprocessor
+    OLM_BASE_MODEL: str = "allenai/olmOCR-2-7B-1025"
+    OLM_LORA_ADAPTER: str = "muhammadghiffari/olm-ocr-cipl-v1"
+    HF_TOKEN: SecretStr = ""  # type: ignore[assignment]
+
+    # ── Azure Document Intelligence — Agent C ─────────────────────────────────
+    AZURE_DI_ENDPOINT: AnyHttpUrl = ""  # type: ignore[assignment]
+    AZURE_DI_KEY: SecretStr = ""  # type: ignore[assignment]
+    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"
+    CEISA_CLIENT_ID: str = ""
+    CEISA_CLIENT_SECRET: SecretStr = ""  # type: ignore[assignment]
+    CEISA_REQUEST_TIMEOUT_SECONDS: int = 30
+    CEISA_POLL_INTERVAL_SECONDS: int = 30
+    CEISA_AES_KEY: SecretStr = ""  # type: ignore[assignment]  # AES-256-GCM key for payload encryption (base64)
+
+    # ── Blockchain ────────────────────────────────────────────────────────────
+    ENABLE_BLOCKCHAIN: bool = True
+    OPERATOR_WALLET_PRIVATE_KEY: SecretStr = ""  # type: ignore[assignment]  # Never log!
+    POLYGON_RPC_URL: str = "https://rpc-amoy.polygon.technology"
+    CONTRACT_ADDRESS: str = ""
+    PINATA_JWT: SecretStr = ""  # type: ignore[assignment]
+    POLYGON_MAX_FEE_GWEI: int = 80
+    POLYGON_MAX_PRIORITY_FEE_GWEI: int = 3
+    POLYGON_ANCHOR_GAS_LIMIT: int = 250_000
+
+    # ── Notifications ─────────────────────────────────────────────────────────
+    RESEND_API_KEY: SecretStr = ""  # type: ignore[assignment]
+    WHATSAPP_TOKEN: SecretStr = ""  # type: ignore[assignment]
+    WHATSAPP_PHONE_NUMBER_ID: str = ""
+    NOTIFICATION_EMAIL_FROM: str = "noreply@tradeflow.ai"
+
+    # ── ChromaDB ──────────────────────────────────────────────────────────────
+    CHROMADB_HOST: str = "localhost"
+    CHROMADB_PORT: int = 8000
+
+    # ── AI / LLM ─────────────────────────────────────────────────────────────
+    GEMINI_API_KEY: SecretStr = Field(..., description="Google Gemini API key")
+    GEMINI_MODEL_PRIMARY: str = "gemini-3.1-pro"
+    GEMINI_MODEL_FALLBACK: str = "gemini-3.5-flash"
+    OPENAI_API_KEY: SecretStr = ""  # type: ignore[assignment]
+    EMBEDDING_MODEL: str = "text-embedding-3-small"
+
+    # LangSmith tracing
+    LANGCHAIN_TRACING_V2: bool = True
+    LANGCHAIN_PROJECT: str = "tradeflow-ai"
+    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_BLOCKCHAIN: bool = True  # type: ignore[assignment] — redeclared intentionally
+    ENABLE_INSW_CHECK: bool = True
+    ENABLE_NOTIFICATIONS_WHATSAPP: bool = False
+    ENABLE_ADAPTIVE_LEARNING: bool = True
+    ENABLE_AI_COPILOT: bool = True
+    ENABLE_HS_RAG: bool = True
+    ENABLE_REJECTION_PREDICTION: bool = True
+    ENABLE_STATUS_POLLING: bool = True
+    ENABLE_MARITIME_DATA_FEATURES: bool = True
+    DETERMINISTIC_E2E: bool = False  # Used in extract.py to swap LLM for deterministic mock
+
+    # ── Thresholds ────────────────────────────────────────────────────────────
+    OCR_FAST_PATH_QUALITY_THRESHOLD: float = 0.95
+    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
+    XGB_MIN_SAMPLES_FOR_MODEL: int = 500
+    MAX_RESUBMIT_ATTEMPTS: int = 5
+    REJECTION_RISK_BLOCK_THRESHOLD: float = 0.70
+
+    # ── Validation rules ──────────────────────────────────────────────────────
+    VALIDATION_RULES_PATH: str = "packages/db/validation_rules.json"
+    CARRIER_PROFILES_PATH: str = "packages/db/carrier_profiles.json"
+    XGBOOST_MODEL_PATH: str = "models/rejection_predictor.json"
+
+    # ── Adaptive learning / drift ─────────────────────────────────────────────
+    RETRAIN_MIN_NEW_SAMPLES: int = 100
+    RETRAIN_MIN_TOTAL_SAMPLES: int = 500
+    DRIFT_LOOKBACK_DAYS: int = 30
+    DRIFT_CORRECTION_THRESHOLD: int = 50
+
+    # ── Celery ────────────────────────────────────────────────────────────────
+    CELERY_TASK_SOFT_TIME_LIMIT: int = 300
+    CELERY_TASK_TIME_LIMIT: int = 600
+
+    # ── Observability ─────────────────────────────────────────────────────────
+    SENTRY_DSN: str = ""
+    POSTHOG_API_KEY: str = ""
+    OTEL_ENABLED: bool = False
+    OTEL_EXPORTER_OTLP_ENDPOINT: str = ""
+
+    @field_validator("CORS_ORIGINS", mode="before")
+    @classmethod
+    def parse_cors_origins(cls, v: str | list[str]) -> list[str]:
+        if isinstance(v, str):
+            return [origin.strip() for origin in v.split(",")]
+        return v
+
+    @field_validator("DEBUG", mode="before")
+    @classmethod
+    def parse_debug(cls, v: bool | str) -> bool:
+        if isinstance(v, str) and v.lower() in {"release", "prod", "production"}:
+            return False
+        return bool(v)
+
+
+class CeleryConfig:
+    """Celery configuration — separate class for Celery's config_from_object."""
+    task_serializer = "json"
+    result_serializer = "json"
+    accept_content = ["json"]
+    timezone = "Asia/Jakarta"
+    enable_utc = True
+    task_soft_time_limit = 300
+    task_time_limit = 600
+    task_acks_late = True           # Ack only after successful completion (NFR-016)
+    worker_prefetch_multiplier = 1  # One task at a time per worker
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()  # type: ignore[call-arg]
+
+
+settings = get_settings()

src/dependencies.py

@@ -0,0 +1,231 @@
+"""
+TradeFlow AI — FastAPI Dependencies
+
+Auth (Keycloak JWT), DB session, tier/role guards.
+PRD §4 Decision 2: Keycloak 26 is the ONLY auth provider.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Annotated
+
+import httpx
+import structlog
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jose import JWTError, jwt
+from jose.exceptions import ExpiredSignatureError
+
+try:
+    from supabase import AsyncClient, acreate_client
+except Exception:  # pragma: no cover - optional in lightweight test environments
+    AsyncClient = None
+    acreate_client = None
+
+from .config import settings
+
+log = structlog.get_logger()
+
+# ── Supabase client (singleton) ───────────────────────────────────────────────
+_supabase_client: AsyncClient | None = None
+
+
+async def init_supabase() -> None:
+    global _supabase_client
+    if acreate_client is None:
+        log.info("Supabase client not available in this environment; skipping initialization")
+        _supabase_client = None
+        return
+
+    _supabase_client = await acreate_client(
+        settings.SUPABASE_URL,
+        settings.SUPABASE_SERVICE_KEY.get_secret_value(),  # Service key for server-side ops
+    )
+    log.info("Supabase client initialized")
+
+
+async def close_supabase() -> None:
+    global _supabase_client
+    if _supabase_client:
+        await _supabase_client.aclose()
+        _supabase_client = None
+
+
+def get_supabase() -> AsyncClient:
+    if _supabase_client is None:
+        raise RuntimeError("Supabase client not initialized. Call init_supabase() first.")
+    return _supabase_client
+
+
+# ── Keycloak JWKS cache ───────────────────────────────────────────────────────
+_keycloak_jwks: dict | None = None
+_keycloak_jwks_time: float = 0
+KEYCLOAK_JWKS_TTL = 3600  # Refresh every hour
+
+
+def get_keycloak_jwks() -> dict:
+    """Fetch Keycloak JWKS with TTL-based caching (refresh every hour)."""
+    global _keycloak_jwks, _keycloak_jwks_time
+    now = time.time()
+
+    if not _keycloak_jwks or (now - _keycloak_jwks_time) > KEYCLOAK_JWKS_TTL:
+        with httpx.Client() as client:
+            response = client.get(settings.KEYCLOAK_JWKS_URL)
+            response.raise_for_status()
+            _keycloak_jwks = response.json()
+            _keycloak_jwks_time = now
+            log.info("Refreshed Keycloak JWKS cache")
+
+    return _keycloak_jwks
+
+
+# ── JWT Bearer scheme ─────────────────────────────────────────────────────────
+bearer_scheme = HTTPBearer(auto_error=True)
+
+
+class CurrentUser:
+    """Decoded Keycloak JWT claims, enriched with Supabase profile."""
+
+    def __init__(
+        self,
+        sub: str,
+        email: str,
+        full_name: str,
+        roles: list[str],
+        tier: str,
+        company_id: str | None,
+        raw_token: str,
+    ) -> None:
+        self.id = sub
+        self.sub = sub
+        self.email = email
+        self.full_name = full_name
+        self.roles = roles
+        self.tier = tier
+        self.company_id = company_id
+        self.raw_token = raw_token
+
+    @property
+    def is_enterprise(self) -> bool:
+        return self.tier == "enterprise"
+
+    @property
+    def is_admin(self) -> bool:
+        return "admin" in self.roles or "supervisor" in self.roles
+
+    @property
+    def role(self) -> str:
+        """Primary role (first in list)."""
+        return self.roles[0] if self.roles else "operator"
+
+
+async def get_current_user(
+    credentials: Annotated[HTTPAuthorizationCredentials, Depends(bearer_scheme)],
+    supabase: Annotated[AsyncClient, Depends(get_supabase)],
+) -> CurrentUser:
+    """
+    Validate Keycloak RS256 JWT and return enriched user.
+    PRD §4 Decision 2: verify against Keycloak JWKS endpoint.
+    """
+    token = credentials.credentials
+
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid or expired authentication token",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        jwks = get_keycloak_jwks()
+        payload = jwt.decode(
+            token,
+            jwks,
+            algorithms=["RS256"],
+            audience=settings.KEYCLOAK_CLIENT_ID,
+            issuer=settings.KEYCLOAK_ISSUER,
+        )
+    except ExpiredSignatureError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token has expired",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    except JWTError as e:
+        log.warning("JWT validation failed", error=str(e))
+        raise credentials_exception
+
+    sub: str = payload.get("sub", "")
+    if not sub:
+        raise credentials_exception
+
+    # Extract Keycloak realm roles
+    realm_access = payload.get("realm_access", {})
+    roles: list[str] = realm_access.get("roles", [])
+    # Filter to only TradeFlow roles
+    tradeflow_roles = [r for r in roles if r in ("operator", "admin", "supervisor", "importer")]
+
+    # Fetch profile from Supabase for tier + company_id
+    try:
+        result = await supabase.table("profiles").select(
+            "id, full_name, email, tier, role, company_id"
+        ).eq("id", sub).single().execute()
+        profile = result.data
+        tier = profile.get("tier", "sme")
+        company_id = profile.get("company_id")
+        full_name = profile.get("full_name", payload.get("name", ""))
+        email = profile.get("email", payload.get("email", ""))
+    except Exception:
+        # Profile not yet created — use JWT claims as fallback
+        tier = "sme"
+        company_id = sub  # Fallback to user's own ID so they can act as their own company
+        full_name = payload.get("name", "")
+        email = payload.get("email", "")
+
+    return CurrentUser(
+        sub=sub,
+        email=email,
+        full_name=full_name,
+        roles=tradeflow_roles if tradeflow_roles else ["operator"],
+        tier=tier,
+        company_id=company_id,
+        raw_token=token,
+    )
+
+
+# ── Role/tier guards ──────────────────────────────────────────────────────────
+
+async def require_enterprise(
+    user: Annotated[CurrentUser, Depends(get_current_user)],
+) -> CurrentUser:
+    """Guard: Enterprise tier only."""
+    if not user.is_enterprise:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="This feature requires an Enterprise tier subscription.",
+        )
+    return user
+
+
+async def require_admin(
+    user: Annotated[CurrentUser, Depends(get_current_user)],
+) -> CurrentUser:
+    """Guard: Admin or Supervisor role only."""
+    if not user.is_admin:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="This action requires Administrator or Supervisor role.",
+        )
+    return user
+
+
+async def require_operator(
+    user: Annotated[CurrentUser, Depends(get_current_user)],
+) -> CurrentUser:
+    """Guard: Any authenticated user with operator/admin/supervisor role."""
+    if "importer" in user.roles and len(user.roles) == 1:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Importers cannot perform operator actions.",
+        )
+    return user

src/main.py

@@ -0,0 +1,125 @@
+"""
+TradeFlow AI — FastAPI Application Entry Point
+
+PRD §1.4 — Main application setup with lifespan management,
+middleware, and router registration.
+"""
+
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+try:
+    import sentry_sdk
+except Exception:  # pragma: no cover - optional dependency for observability
+    sentry_sdk = None
+
+import structlog
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+
+try:
+    from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+except Exception:  # pragma: no cover - optional
+    FastAPIInstrumentor = None
+
+try:
+    from prometheus_fastapi_instrumentator import Instrumentator
+except Exception:  # pragma: no cover - optional
+    Instrumentator = None
+
+from .config import settings
+from .dependencies import close_supabase, init_supabase
+from .routers import admin, batches, blockchain, hs_recommend, vessel
+from .utils.telemetry import setup_telemetry
+
+log = structlog.get_logger()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
+    """Application lifespan — startup and shutdown."""
+    log.info("TradeFlow AI starting up", environment=settings.ENVIRONMENT)
+
+    # Initialize Sentry
+    if settings.SENTRY_DSN:
+        sentry_sdk.init(
+            dsn=settings.SENTRY_DSN,
+            traces_sample_rate=0.2,
+            profiles_sample_rate=0.1,
+            environment=settings.ENVIRONMENT,
+        )
+
+    # Initialize Supabase client
+    await init_supabase()
+
+    log.info("TradeFlow AI ready", version="1.0.0")
+    yield
+
+    # Shutdown
+    await close_supabase()
+    log.info("TradeFlow AI shutdown complete")
+
+
+def create_app() -> FastAPI:
+    app = FastAPI(
+        title="TradeFlow AI API",
+        description="Predictive Customs Intelligence Platform — Cikarang Dry Port",
+        version="1.0.0",
+        docs_url="/docs" if settings.ENVIRONMENT != "production" else None,
+        redoc_url="/redoc" if settings.ENVIRONMENT != "production" else None,
+        lifespan=lifespan,
+    )
+
+    # Setup OpenTelemetry & Prometheus (Phase 6)
+    setup_telemetry(app)
+
+    # ── Middleware ────────────────────────────────────────────────
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.CORS_ORIGINS,
+        allow_credentials=False,
+        allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
+        allow_headers=["Content-Type", "Authorization"],
+        expose_headers=["Content-Length"],
+        max_age=3600,
+    )
+    app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+    # Rate limiting middleware (optional)
+    try:
+        from slowapi import Limiter
+        from slowapi.util import get_remote_address
+        limiter = Limiter(key_func=get_remote_address)
+        app.state.limiter = limiter
+    except Exception:
+        app.state.limiter = None
+
+    # ── Routers ───────────────────────────────────────────────────
+    app.include_router(batches.router, prefix="/api/v1", tags=["batches"])
+    app.include_router(hs_recommend.router, prefix="/api/v1", tags=["hs-recommend"])
+    app.include_router(blockchain.router, prefix="/api/v1", tags=["blockchain"])
+    app.include_router(vessel.router)
+    app.include_router(admin.router, prefix="/api/v1/admin", tags=["admin"])
+
+    # ── Prometheus metrics (optional) ────────────────────────────
+    if Instrumentator is not None:
+        Instrumentator(
+            should_group_status_codes=True,
+            should_ignore_untemplated=True,
+        ).instrument(app).expose(app, endpoint="/metrics")
+
+    # ── OpenTelemetry (optional) ─────────────────────────────────
+    if settings.OTEL_ENABLED and FastAPIInstrumentor is not None:
+        FastAPIInstrumentor().instrument_app(app)
+
+    return app
+
+
+app = create_app()
+
+
+@app.get("/health", tags=["health"])
+async def health_check() -> dict[str, str]:
+    """Health check endpoint — public, no auth required."""
+    return {"status": "ok", "version": "1.0.0", "environment": settings.ENVIRONMENT}