Deploy VERA API (round 3 models)

README.md

@@ -1,13 +1,24 @@
 ---
-title: Vera Api
-emoji: 🚀
-colorFrom: purple
-colorTo: purple
+title: VERA API
+colorFrom: blue
+colorTo: gray
 sdk: gradio
-sdk_version: 6.27.0
-python_version: '3.12'
+sdk_version: 5.49.1
+python_version: "3.12"
 app_file: app.py
+suggested_hardware: zero-a10g
 pinned: false
+short_description: VERA earbud review analysis API (FastAPI + Celery worker)
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# VERA API (Hugging Face Space)
+
+Backend for the VERA review analyzer. `app.py` starts loopback Redis, applies database
+migrations, starts one Celery worker (which downloads the models from the private
+`HF_MODEL_REPO`), and serves the FastAPI app on port 7860.
+
+Endpoints: `GET /health`, `GET /ready`, `POST /api/reviews`, `GET /api/reviews/{id}`,
+`GET /api/reviews/{id}/status`, `GET /api/products`, `GET /api/products/{asin}`.
+
+Secrets (Settings -> Variables and secrets): `DATABASE_URL` (Supabase session pooler,
+`?sslmode=require`), `HF_MODEL_REPO`, `HF_TOKEN` (read), `SECRET_KEY`, `CORS_ORIGINS`.

app.py

@@ -0,0 +1,124 @@
+"""VERA API + worker launcher for a free Hugging Face Gradio Space on ZeroGPU hardware.
+
+Why this shape: Docker Spaces and CPU-basic Gradio Spaces require a PRO plan, while a Gradio
+Space on ZeroGPU hardware is free. ZeroGPU stops any app that never reports startup, and that
+report is normally sent by ``gradio.Blocks.launch``. This launcher instead serves the existing
+FastAPI app with uvicorn and sends the report itself (``spaces.zero.startup``). All inference
+runs on CPU with onnxruntime; the GPU is never requested, so the ZeroGPU quota is not used.
+
+One container owns: loopback Redis (Celery broker), Alembic migrations, one solo Celery worker
+(which downloads the models from ``HF_MODEL_REPO`` and runs the outbox dispatcher), and the API
+on port 7860. If Redis or the worker exits, the process exits so the Space restarts.
+
+Required Space secrets: DATABASE_URL, HF_MODEL_REPO, HF_TOKEN (read), SECRET_KEY, CORS_ORIGINS.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
+
+import spaces  # must be imported before any CUDA-related package
+
+ROOT = Path(__file__).resolve().parent
+BACKEND = ROOT / "backend"
+MODEL_DIR = Path(os.environ.get("MODEL_DIR", "/home/user/models"))
+REDIS_URL = "redis://127.0.0.1:6379/0"
+
+for name in ("DATABASE_URL", "HF_MODEL_REPO"):
+    if not os.environ.get(name):
+        sys.exit(f"Set {name} in the Space secrets")
+
+os.environ.update(
+    ENVIRONMENT="production",
+    MODEL_DIR=str(MODEL_DIR),
+    REDIS_URL=REDIS_URL,
+    CELERY_BROKER_URL=REDIS_URL,
+    CELERY_RESULT_BACKEND="",
+)
+MODEL_DIR.mkdir(parents=True, exist_ok=True)
+sys.path.insert(0, str(BACKEND))
+os.chdir(BACKEND)
+
+
+@spaces.GPU
+def _zerogpu_placeholder() -> str:
+    """ZeroGPU only reports startup when at least one GPU function exists. Never called."""
+    return "unused"
+
+
+def _log(message: str) -> None:
+    print(f"[vera-launcher] {message}", flush=True)
+
+
+def start_redis() -> subprocess.Popen[bytes]:
+    Path("/tmp/redis").mkdir(exist_ok=True)
+    proc = subprocess.Popen(
+        ["redis-server", "--bind", "127.0.0.1", "--protected-mode", "yes", "--port", "6379",
+         "--dir", "/tmp/redis", "--save", "", "--appendonly", "no"]
+    )
+    for _ in range(30):
+        ping = subprocess.run(["redis-cli", "-h", "127.0.0.1", "ping"], capture_output=True)
+        if ping.stdout.strip() == b"PONG":
+            _log("redis ready")
+            return proc
+        if proc.poll() is not None:
+            break
+        time.sleep(1)
+    sys.exit("redis-server failed to start")
+
+
+def run_migrations() -> None:
+    subprocess.run([sys.executable, "-m", "alembic", "upgrade", "head"], check=True)
+    _log("database migrations applied")
+
+
+def start_worker() -> subprocess.Popen[bytes]:
+    return subprocess.Popen(
+        [sys.executable, "-m", "celery", "-A", "app.worker.celery_app", "worker",
+         "--pool=solo", "--concurrency=1", "--loglevel=info"]
+    )
+
+
+def supervise(children: dict[str, subprocess.Popen[bytes]]) -> None:
+    """Exit the whole Space if a required child dies, so the platform restarts it."""
+    while True:
+        for name, proc in children.items():
+            if proc.poll() is not None:
+                _log(f"{name} exited with code {proc.returncode}; stopping the Space")
+                for other in children.values():
+                    if other.poll() is None:
+                        other.terminate()
+                os._exit(1)
+        time.sleep(5)
+
+
+def main() -> None:
+    import uvicorn
+
+    redis_proc = start_redis()
+    run_migrations()
+    worker = start_worker()
+    threading.Thread(
+        target=supervise, args=({"redis": redis_proc, "worker": worker},), daemon=True
+    ).start()
+
+    from app.main import app
+
+    @app.get("/", include_in_schema=False)
+    def root() -> dict[str, str]:
+        return {"service": "VERA review analysis API", "health": "/health", "ready": "/ready"}
+
+    zero = getattr(spaces, "zero", None)
+    if zero is not None and hasattr(zero, "startup"):
+        zero.startup()
+        _log("ZeroGPU startup reported")
+    uvicorn.run(app, host="0.0.0.0", port=7860, proxy_headers=False, access_log=False)
+
+
+if __name__ == "__main__":
+    main()

backend/alembic.ini

@@ -0,0 +1,37 @@
+[alembic]
+script_location = migrations
+prepend_sys_path = .
+sqlalchemy.url = postgresql://unused:unused@localhost/vera
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

backend/app/__init__.py

@@ -0,0 +1,3 @@
+"""VERA — Verified Experience & Review Analyzer (backend package)."""
+
+__version__ = "1.0.0"

backend/app/api/__init__.py

@@ -0,0 +1 @@
+"""FastAPI routing and dependencies."""

backend/app/api/deps.py

@@ -0,0 +1,18 @@
+"""Shared FastAPI dependencies."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+
+from sqlalchemy.orm import Session
+
+from app.database import SessionLocal
+
+
+def get_db() -> Generator[Session, None, None]:
+    """Yield one request-scoped SQLAlchemy session."""
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

backend/app/api/logging.py

@@ -0,0 +1,50 @@
+"""Structured API logging with request correlation and bounded, safe access fields."""
+
+from __future__ import annotations
+
+import json
+import logging
+from contextvars import ContextVar
+from datetime import UTC, datetime
+from logging.config import dictConfig
+
+request_id_context: ContextVar[str] = ContextVar("request_id", default="-")
+
+
+class JsonFormatter(logging.Formatter):
+    """Serialize log records without exception bodies, query strings, or request payloads."""
+
+    def format(self, record: logging.LogRecord) -> str:
+        payload = {
+            "timestamp": datetime.fromtimestamp(record.created, UTC).isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": record.getMessage(),
+            "request_id": request_id_context.get(),
+        }
+        for field in ("method", "route", "status_code", "duration_ms", "exception_type"):
+            if hasattr(record, field):
+                payload[field] = getattr(record, field)
+        if record.exc_info and record.exc_info[0]:
+            payload["exception_type"] = record.exc_info[0].__name__
+        return json.dumps(payload, ensure_ascii=True, default=str)
+
+
+def configure_json_logging() -> None:
+    """Configure the API's console logs; access logging is owned by the middleware."""
+    dictConfig(
+        {
+            "version": 1,
+            "disable_existing_loggers": False,
+            "formatters": {"json": {"()": JsonFormatter}},
+            "handlers": {
+                "console": {"class": "logging.StreamHandler", "formatter": "json"},
+                "null": {"class": "logging.NullHandler"},
+            },
+            "root": {"level": "INFO", "handlers": ["console"]},
+            "loggers": {
+                "uvicorn": {"handlers": ["console"], "propagate": False},
+                "uvicorn.access": {"handlers": ["null"], "propagate": False},
+            },
+        }
+    )

backend/app/api/middleware.py

@@ -0,0 +1,192 @@
+"""Bounded ASGI request handling and process-local submission throttling."""
+
+from __future__ import annotations
+
+import logging
+import math
+import re
+import time
+from collections import OrderedDict, deque
+from ipaddress import ip_address, ip_network
+from threading import Lock
+from uuid import uuid4
+
+from starlette.datastructures import Headers, MutableHeaders
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from app.api.logging import request_id_context
+from app.config import Settings
+
+logger = logging.getLogger("vera.requests")
+REQUEST_ID_PATTERN = re.compile(r"[A-Za-z0-9._-]{1,64}\Z")
+
+
+class SlidingWindowLimiter:
+    """Bound memory and requests per client; restart/multiple-process limits are documented."""
+
+    def __init__(self, limit: int, window: int, max_clients: int) -> None:
+        self.limit, self.window, self.max_clients = limit, window, max_clients
+        self.clients: OrderedDict[str, deque[float]] = OrderedDict()
+        self.lock = Lock()
+
+    def check(self, client: str, now: float | None = None) -> int:
+        """Return zero when allowed, otherwise seconds until a retry may be accepted."""
+        now = time.monotonic() if now is None else now
+        with self.lock:
+            while self.clients and next(iter(self.clients.values()))[-1] <= now - self.window:
+                self.clients.popitem(last=False)
+            bucket = self.clients.get(client)
+            if bucket is None:
+                if len(self.clients) >= self.max_clients:
+                    return self.window
+                bucket = self.clients[client] = deque()
+            while bucket and bucket[0] <= now - self.window:
+                bucket.popleft()
+            if len(bucket) >= self.limit:
+                return max(1, math.ceil(bucket[0] + self.window - now))
+            bucket.append(now)
+            self.clients.move_to_end(client)
+            return 0
+
+
+class RequestGuard:
+    """Apply correlation IDs, body bounds, security headers, and POST rate limits."""
+
+    def __init__(self, app: ASGIApp, settings: Settings) -> None:
+        self.app = app
+        self.settings = settings
+        self.limiter = SlidingWindowLimiter(
+            settings.REVIEW_RATE_LIMIT,
+            settings.REVIEW_RATE_WINDOW_SECONDS,
+            settings.RATE_LIMIT_MAX_CLIENTS,
+        )
+        self.proxies = [
+            ip_network(value.strip(), strict=False)
+            for value in settings.TRUSTED_PROXY_CIDRS.split(",")
+            if value.strip()
+        ]
+
+    def client_key(self, scope: Scope, headers: Headers) -> str:
+        """Strip only explicitly trusted proxy hops from the right of a forwarded chain."""
+        peer = (scope.get("client") or ("unknown", 0))[0]
+        try:
+            address = ip_address(peer)
+            if not any(address in network for network in self.proxies):
+                return str(address)
+            hops = headers.get("x-forwarded-for", "").split(",")
+            if len(hops) > 20:
+                return peer
+            for hop in reversed(hops):
+                address = ip_address(hop.strip())
+                if not any(address in network for network in self.proxies):
+                    return str(address)
+            return str(address)
+        except ValueError:
+            return peer
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        headers = Headers(scope=scope)
+        supplied = headers.get("x-request-id", "")
+        request_id = supplied if REQUEST_ID_PATTERN.fullmatch(supplied) else uuid4().hex
+        token = request_id_context.set(request_id)
+        scope.setdefault("state", {})["request_id"] = request_id
+        started = time.monotonic()
+        status_code = 500
+        response_started = False
+
+        async def guarded_send(message: Message) -> None:
+            nonlocal status_code, response_started
+            if message["type"] == "http.response.start":
+                response_started = True
+                status_code = message["status"]
+                outgoing = MutableHeaders(scope=message)
+                outgoing["X-Request-ID"] = request_id
+                outgoing["X-Content-Type-Options"] = "nosniff"
+                outgoing["X-Frame-Options"] = "DENY"
+                outgoing["Referrer-Policy"] = "no-referrer"
+                outgoing["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
+                outgoing["Cache-Control"] = "no-store"
+                origin = headers.get("origin")
+                if origin and (
+                    origin in self.settings.cors_origins or "*" in self.settings.cors_origins
+                ):
+                    outgoing["Access-Control-Allow-Origin"] = origin
+                    outgoing["Access-Control-Expose-Headers"] = (
+                        "X-Total-Count, X-Request-ID, Retry-After"
+                    )
+                    outgoing.add_vary_header("Origin")
+                if self.settings.is_production:
+                    outgoing["Strict-Transport-Security"] = "max-age=31536000"
+            await send(message)
+
+        async def reject(code: int, detail: str, extra: dict[str, str] | None = None) -> None:
+            await JSONResponse(
+                {"detail": detail, "request_id": request_id}, status_code=code, headers=extra
+            )(scope, receive, guarded_send)
+
+        try:
+            length = headers.get("content-length")
+            if length is not None:
+                if not length.isascii() or not length.isdigit() or len(length) > 10:
+                    await reject(400, "Invalid Content-Length header.")
+                    return
+                if int(length) > self.settings.MAX_REQUEST_BODY_BYTES:
+                    await reject(413, "Request body is too large.")
+                    return
+            if headers.get("content-encoding", "identity").lower() != "identity":
+                await reject(415, "Compressed request bodies are not supported.")
+                return
+            if scope["method"] == "POST" and scope["path"].rstrip("/") == "/api/reviews":
+                retry = self.limiter.check(self.client_key(scope, headers))
+                if retry:
+                    await reject(
+                        429,
+                        "Too many reviews submitted. Please wait before trying again.",
+                        {"Retry-After": str(retry)},
+                    )
+                    return
+            # Buffer at most the configured limit before any handler can persist side effects.
+            body = bytearray()
+            while True:
+                message = await receive()
+                if message["type"] == "http.disconnect":
+                    status_code = 499
+                    return
+                chunk = message.get("body", b"")
+                if len(body) + len(chunk) > self.settings.MAX_REQUEST_BODY_BYTES:
+                    await reject(413, "Request body is too large.")
+                    return
+                body.extend(chunk)
+                if not message.get("more_body", False):
+                    break
+            delivered = False
+
+            async def replay() -> Message:
+                nonlocal delivered
+                if not delivered:
+                    delivered = True
+                    return {"type": "http.request", "body": bytes(body), "more_body": False}
+                return await receive()
+
+            await self.app(scope, replay, guarded_send)
+        except Exception as exc:
+            logger.error("request_failed", extra={"exception_type": type(exc).__name__})
+            if response_started:
+                raise
+            await reject(500, "The service could not complete this request.")
+        finally:
+            route = getattr(scope.get("route"), "path", "unmatched")
+            logger.info(
+                "request_completed",
+                extra={
+                    "method": scope["method"],
+                    "route": route,
+                    "status_code": status_code,
+                    "duration_ms": round((time.monotonic() - started) * 1000, 2),
+                },
+            )
+            request_id_context.reset(token)

backend/app/api/routes/__init__.py

@@ -0,0 +1 @@
+"""HTTP route modules."""

backend/app/api/routes/health.py

@@ -0,0 +1,58 @@
+"""Liveness and diagnostic health endpoint."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+from sqlalchemy import text
+
+from app.config import settings
+from app.database import SessionLocal
+from app.ml.models.loader import artifact_status
+
+router = APIRouter(tags=["health"])
+
+
+def database_ready() -> bool:
+    """Check database reachability without loading models or publishing queue work."""
+    try:
+        with SessionLocal() as db:
+            db.execute(text("SELECT 1"))
+        return True
+    except Exception:
+        return False
+
+
+@router.get("/ready")
+def ready() -> JSONResponse:
+    """Readiness is database reachability, not a guarantee of worker/model readiness."""
+    available = database_ready()
+    return JSONResponse(
+        {"status": "ready" if available else "unavailable", "database": available},
+        status_code=200 if available else 503,
+        headers={} if available else {"Retry-After": "10"},
+    )
+
+
+@router.get("/health")
+def health() -> dict[str, Any]:
+    """Return 200 for liveness, with explicit dependency diagnostics.
+
+    The API itself does not load ML artifacts: workers own inference and fail
+    closed on missing artifacts. Keeping liveness independent means uptime
+    monitors can report a useful degraded state during a model rollout.
+    """
+    database_ok = database_ready()
+    try:
+        models = artifact_status(settings.MODEL_DIR)
+    except OSError:
+        models = {"accessible": False}
+    ready = database_ok and all(models.values())
+    return {
+        "status": "healthy" if ready else "degraded",
+        "database": database_ok,
+        "model_artifacts": models,
+        "environment": settings.ENVIRONMENT,
+    }

backend/app/api/routes/products.py

@@ -0,0 +1,80 @@
+"""Product ranking and detail routes (PRD §14.2, §16.4)."""
+
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from sqlalchemy import asc, desc, func, nulls_last
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db
+from app.models.product import Product
+from app.models.review import Review, ReviewStatus
+from app.schemas.product import ProductDetailResponse, ProductResponse
+from app.schemas.review import ReviewResponse
+
+router = APIRouter(prefix="/api/products", tags=["products"])
+
+SortBy = Literal["corrected_score", "star_average", "review_count", "uncertainty", "rank_position"]
+SortOrder = Literal["asc", "desc"]
+
+
+@router.get("", response_model=list[ProductResponse])
+def list_products(
+    response: Response,
+    sort_by: SortBy = "corrected_score",
+    order: SortOrder = "desc",
+    limit: int = Query(default=50, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+    db: Session = Depends(get_db),
+) -> list[Product]:
+    """Paginated ranking of scored products.
+
+    Product shells without a completed review are excluded. Ties break on ASIN
+    so pagination is stable. ``X-Total-Count`` carries the unpaginated total.
+    """
+    column = getattr(Product, sort_by)
+    direction: Any = desc(column) if order == "desc" else asc(column)
+    query = db.query(Product).filter(Product.corrected_score.is_not(None))
+    response.headers["X-Total-Count"] = str(query.count())
+    return (
+        query.order_by(nulls_last(direction), asc(Product.asin)).offset(offset).limit(limit).all()
+    )
+
+
+@router.get("/{asin}", response_model=ProductDetailResponse)
+def get_product(
+    asin: str,
+    reviews_limit: int = Query(default=20, ge=1, le=100),
+    reviews_offset: int = Query(default=0, ge=0),
+    db: Session = Depends(get_db),
+) -> ProductDetailResponse:
+    """Product aggregate plus a newest-first page of its completed reviews."""
+    product = db.get(Product, asin)
+    if product is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
+
+    base = db.query(Review).filter(
+        Review.product_asin == asin, Review.status == ReviewStatus.DONE.value
+    )
+    total = base.with_entities(func.count(Review.id)).scalar() or 0
+    reviews = (
+        base.order_by(desc(Review.created_at), desc(Review.id))
+        .offset(reviews_offset)
+        .limit(reviews_limit)
+        .all()
+    )
+    return ProductDetailResponse.model_validate(
+        {
+            **ProductResponse.model_validate(product).model_dump(),
+            "eb_shrinkage_factor": product.eb_shrinkage_factor,
+            "aspect_summary": product.aspect_summary,
+            "last_aggregated_at": product.last_aggregated_at,
+            "reviews": [ReviewResponse.model_validate(r) for r in reviews],
+            "reviews_total": total,
+            "reviews_limit": reviews_limit,
+            "reviews_offset": reviews_offset,
+            "reviews_has_more": reviews_offset + len(reviews) < total,
+        }
+    )

backend/app/api/routes/reviews.py

@@ -0,0 +1,115 @@
+"""Review submission and polling routes (PRD §14.1–14.2)."""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_db
+from app.models.job import Job, JobKind
+from app.models.product import Product
+from app.models.review import Review, ReviewStatus
+from app.schemas.review import ReviewCreate, ReviewResponse, ReviewTaskStatus
+from app.worker.outbox import send_wakeup
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/reviews", tags=["reviews"])
+
+STATUS_MESSAGES: dict[str, str] = {
+    ReviewStatus.PENDING.value: "Review is queued for processing",
+    ReviewStatus.PROCESSING.value: "Review is being processed",
+    ReviewStatus.DONE.value: "Review processing completed",
+    ReviewStatus.FAILED.value: "Review processing failed",
+}
+
+
+def _get_review_or_404(db: Session, review_id: int) -> Review:
+    review = db.get(Review, review_id)
+    if review is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Review not found")
+    return review
+
+
+def _ensure_product_shell(db: Session, asin: str, title: str | None) -> None:
+    """Create or update the product row without losing concurrent first reviews."""
+    product = db.get(Product, asin)
+    if product is None:
+        try:
+            # The savepoint lets a concurrent ASIN insert fail locally while
+            # preserving the review/job transaction that follows it.
+            with db.begin_nested():
+                db.add(Product(asin=asin, title=title))
+                db.flush()
+        except IntegrityError:
+            product = db.get(Product, asin)
+            if product is None:
+                raise
+    if product is not None and title and not product.title:
+        product.title = title
+
+
+@router.post("", response_model=ReviewTaskStatus, status_code=status.HTTP_202_ACCEPTED)
+def submit_review(payload: ReviewCreate, db: Session = Depends(get_db)) -> ReviewTaskStatus:
+    """Persist a review and its processing job atomically; return for polling.
+
+    The product shell is created first because ``reviews.product_asin`` is a
+    foreign key (the PRD's schema made every first review of an ASIN fail).
+    After commit a wake-up is sent to Redis; if Redis is down the job is still
+    durable and the outbox dispatcher delivers it later, so this never 503s.
+    """
+    asin = payload.product_asin
+    if asin:
+        _ensure_product_shell(db, asin, payload.product_title)
+
+    review = Review(
+        text=payload.text,
+        star_rating=payload.star_rating,
+        helpful_votes=payload.helpful_votes,
+        verified_purchase=payload.verified_purchase,
+        product_title=payload.product_title,
+        product_asin=asin,
+        status=ReviewStatus.PENDING.value,
+    )
+    db.add(review)
+    db.flush()  # assigns review.id
+    job = Job(kind=JobKind.REVIEW.value, review_id=review.id)
+    db.add(job)
+    db.commit()
+
+    try:
+        send_wakeup(job.id)
+        job.last_dispatched_at = datetime.now(UTC)
+        db.commit()
+    except Exception:
+        db.rollback()
+        logger.warning("Broker unavailable for job %s; outbox will retry", job.id, exc_info=True)
+
+    return ReviewTaskStatus(
+        review_id=review.id,
+        task_id=job.id,
+        status=ReviewStatus.PENDING.value,
+        message="Review queued for processing",
+    )
+
+
+@router.get("/{review_id}", response_model=ReviewResponse)
+def get_review(review_id: int, db: Session = Depends(get_db)) -> Review:
+    """Get a review and, when available, its pipeline output."""
+    return _get_review_or_404(db, review_id)
+
+
+@router.get("/{review_id}/status", response_model=ReviewTaskStatus)
+def get_review_status(review_id: int, db: Session = Depends(get_db)) -> ReviewTaskStatus:
+    """Lightweight lifecycle state for polling clients."""
+    review = _get_review_or_404(db, review_id)
+    job_id = db.query(Job.id).filter(Job.review_id == review.id).scalar()
+    return ReviewTaskStatus(
+        review_id=review.id,
+        task_id=job_id,
+        status=review.status,
+        message=STATUS_MESSAGES.get(review.status, "Unknown status"),
+    )

backend/app/config.py

@@ -0,0 +1,134 @@
+"""Application settings loaded from environment variables / the repo-root `.env`.
+
+Every setting is documented in `.env.example`. Settings are read once and
+cached via :func:`get_settings`; import the module-level ``settings`` object for
+convenience.
+"""
+
+from __future__ import annotations
+
+from functools import lru_cache
+from ipaddress import ip_network
+from pathlib import Path
+from typing import Literal
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+#: Repository root (``backend/app/config.py`` -> two parents up from ``app``).
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _normalize_database_url(url: str) -> str:
+    """Force the psycopg2 driver for Postgres URLs.
+
+    Supabase/Heroku-style ``postgres://`` URLs are rejected by SQLAlchemy 2.x,
+    so both ``postgres://`` and bare ``postgresql://`` are rewritten to
+    ``postgresql+psycopg2://``. Other schemes (e.g. ``sqlite://``) pass through.
+    """
+    for prefix in ("postgres://", "postgresql://"):
+        if url.startswith(prefix):
+            return "postgresql+psycopg2://" + url[len(prefix) :]
+    return url
+
+
+def _normalize_redis_url(url: str) -> str:
+    """Ensure TLS Redis URLs carry ``ssl_cert_reqs``.
+
+    Celery refuses ``rediss://`` broker URLs without an explicit
+    ``ssl_cert_reqs`` query parameter. Upstash presents a valid public
+    certificate, so ``required`` (full verification) is the safe default.
+    """
+    if not url.startswith("rediss://"):
+        return url
+    parts = urlsplit(url)
+    query = dict(parse_qsl(parts.query))
+    query.setdefault("ssl_cert_reqs", "required")
+    return urlunsplit(parts._replace(query=urlencode(query)))
+
+
+class Settings(BaseSettings):
+    """Typed runtime configuration."""
+
+    model_config = SettingsConfigDict(
+        env_file=REPO_ROOT / ".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+        case_sensitive=True,
+    )
+
+    # --- Database / queue ---------------------------------------------------
+    DATABASE_URL: str = Field(..., description="SQLAlchemy URL (Supabase or local Postgres).")
+    REDIS_URL: str = Field(..., description="Redis URL (Upstash rediss:// or local redis://).")
+    CELERY_BROKER_URL: str = ""
+    CELERY_RESULT_BACKEND: str = ""
+
+    # --- App ------------------------------------------------------------------
+    ENVIRONMENT: Literal["development", "test", "production"] = "development"
+    SECRET_KEY: str = ""
+    CORS_ORIGINS: str = "http://localhost:3000"
+    REVIEW_RATE_LIMIT: int = Field(default=10, ge=1, le=10000)
+    REVIEW_RATE_WINDOW_SECONDS: int = Field(default=60, ge=1, le=86400)
+    RATE_LIMIT_MAX_CLIENTS: int = Field(default=10000, ge=1, le=100000)
+    MAX_REQUEST_BODY_BYTES: int = Field(default=65536, ge=1024, le=1048576)
+    TRUSTED_PROXY_CIDRS: str = ""
+
+    @field_validator("TRUSTED_PROXY_CIDRS")
+    @classmethod
+    def _trusted_proxies(cls, value: str) -> str:
+        for network in value.split(","):
+            if network.strip():
+                ip_network(network.strip(), strict=False)
+        return value.strip()
+
+    # --- Models -----------------------------------------------------------------
+    MODEL_DIR: Path = Path("data/models")
+    HF_MODEL_REPO: str = ""
+    HF_TOKEN: str = ""
+
+    @field_validator("DATABASE_URL")
+    @classmethod
+    def _db_url(cls, value: str) -> str:
+        return _normalize_database_url(value.strip())
+
+    @field_validator("REDIS_URL", "CELERY_BROKER_URL", "CELERY_RESULT_BACKEND")
+    @classmethod
+    def _redis_url(cls, value: str) -> str:
+        return _normalize_redis_url(value.strip())
+
+    @field_validator("MODEL_DIR")
+    @classmethod
+    def _model_dir(cls, value: Path) -> Path:
+        # Relative paths are anchored at the repo root so the API (run from
+        # `backend/`) and notebooks (run from the root) resolve the same folder.
+        return value if value.is_absolute() else (REPO_ROOT / value).resolve()
+
+    # --- Derived --------------------------------------------------------------
+    @property
+    def broker_url(self) -> str:
+        """Celery broker; falls back to ``REDIS_URL``."""
+        return self.CELERY_BROKER_URL or self.REDIS_URL
+
+    @property
+    def result_backend(self) -> str:
+        """Celery result backend; falls back to ``REDIS_URL``."""
+        return self.CELERY_RESULT_BACKEND or self.REDIS_URL
+
+    @property
+    def cors_origins(self) -> list[str]:
+        """``CORS_ORIGINS`` split on commas, blanks removed."""
+        return [origin.strip() for origin in self.CORS_ORIGINS.split(",") if origin.strip()]
+
+    @property
+    def is_production(self) -> bool:
+        return self.ENVIRONMENT == "production"
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    """Return the cached settings instance."""
+    return Settings()  # values come from env
+
+
+settings = get_settings()

backend/app/database.py

@@ -0,0 +1,60 @@
+"""SQLAlchemy engine, session factory, and declarative base."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+from sqlalchemy import JSON, create_engine
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from app.config import settings
+
+#: JSON column type: native JSONB on Postgres, plain JSON elsewhere (SQLite tests/CI).
+JSONType = JSON().with_variant(JSONB(), "postgresql")
+
+
+class Base(DeclarativeBase):
+    """Declarative base for all ORM models."""
+
+
+def build_engine(url: str) -> Engine:
+    """Create an engine with settings appropriate for the URL's dialect."""
+    if url.startswith("sqlite"):
+        # In-memory SQLite needs a single shared connection across threads.
+        in_memory = ":memory:" in url or url.rstrip("/") == "sqlite:"
+        return create_engine(
+            url,
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool if in_memory else None,
+        )
+    return create_engine(
+        url,
+        pool_pre_ping=True,  # survive Supabase/pooler idle disconnects
+        pool_size=5,
+        max_overflow=5,
+        pool_recycle=1800,
+    )
+
+
+engine = build_engine(settings.DATABASE_URL)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
+
+
+@contextmanager
+def session_scope() -> Iterator[Session]:
+    """Context-managed session for non-request code (Celery tasks, scripts).
+
+    Rolls back on any exception and always closes the session.
+    """
+    db = SessionLocal()
+    try:
+        yield db
+    except Exception:
+        db.rollback()
+        raise
+    finally:
+        db.close()

backend/app/main.py

@@ -0,0 +1,33 @@
+"""FastAPI application entry point."""
+
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.api.logging import configure_json_logging
+from app.api.middleware import RequestGuard
+from app.api.routes import health, products, reviews
+from app.config import settings
+
+
+def create_app() -> FastAPI:
+    """Construct the HTTP API without loading worker-owned ML models."""
+    app = FastAPI(title="VERA API", version="1.0.0")
+    configure_json_logging()
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_origins,
+        allow_credentials=False,
+        allow_methods=["GET", "POST"],
+        allow_headers=["Content-Type", "Authorization", "X-Request-ID"],
+        expose_headers=["X-Total-Count", "X-Request-ID", "Retry-After"],
+    )
+    app.add_middleware(RequestGuard, settings=settings)
+    app.include_router(health.router)
+    app.include_router(reviews.router)
+    app.include_router(products.router)
+    return app
+
+
+app = create_app()

backend/app/ml/__init__.py

@@ -0,0 +1 @@
+"""VERA ML pipeline: seven signal agents, fusion, and calibration (PRD §8–§11)."""

backend/app/ml/agents/__init__.py

@@ -0,0 +1,10 @@
+"""The seven VERA signal agents (PRD §9).
+
+1. :class:`~app.ml.agents.preprocessor.TextPreprocessor`
+2. :class:`~app.ml.agents.experience_extractor.ExperienceExtractor`
+3. :class:`~app.ml.agents.absa.ABSAModule`
+4. :class:`~app.ml.agents.consistency.ConsistencyScorer`
+5. :class:`~app.ml.agents.quality.QualityAssessor`
+6. :class:`~app.ml.agents.domain_rating.DomainRatingModel`
+7. :class:`~app.ml.agents.fusion.FusionModel`
+"""

backend/app/ml/agents/absa.py

@@ -0,0 +1,135 @@
+"""Agent 3 — Aspect-Based Sentiment Analysis (PRD §9, Agent 3).
+
+Uses ``yangheng/deberta-v3-base-absa-v1.1`` exported to ONNX, unchanged (no
+fine-tuning). The learned :class:`~app.ml.agents.aspect_detector.AspectDetector`
+decides which of the 10 earbud aspects are mentioned (keyword matching is the
+fallback when no detector is loaded); the model then classifies sentiment for
+those aspects in ONE batched inference call.
+
+Input format: ``[CLS] review_text [SEP] aspect [SEP]`` — review FIRST, aspect
+SECOND (PRD weakness #4). Only the review is ever truncated.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Mapping
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import numpy as np
+
+from app.ml.constants import ASPECT_KEYWORDS, ASPECT_PHRASES, EARBUD_ASPECTS
+from app.ml.models.onnx_session import InferenceSessionLike, build_feed, create_session, softmax
+from app.ml.models.tokenizer_wrapper import LightTokenizer
+
+if TYPE_CHECKING:
+    from app.ml.agents.aspect_detector import AspectDetector
+
+_VALID_SENTIMENTS = frozenset({"negative", "neutral", "positive"})
+
+
+def _compile_keyword_patterns() -> dict[str, re.Pattern[str]]:
+    """One regex per aspect; keywords must start at a word boundary."""
+    return {
+        aspect: re.compile(
+            r"\b(?:" + "|".join(re.escape(kw) for kw in ASPECT_KEYWORDS[aspect]) + r")",
+            re.IGNORECASE,
+        )
+        for aspect in EARBUD_ASPECTS
+    }
+
+
+_KEYWORD_PATTERNS = _compile_keyword_patterns()
+
+
+def load_id2label(config_path: str | Path) -> dict[int, str]:
+    """Read the ABSA label mapping written by the export notebook.
+
+    The PRD hardcoded ``{0: negative, 1: neutral, 2: positive}``; reading the
+    exported ``config.id2label`` removes the risk of a silent label swap.
+    """
+    with open(config_path, encoding="utf-8") as fh:
+        raw = json.load(fh)["id2label"]
+    mapping = {int(k): str(v).strip().lower() for k, v in raw.items()}
+    if set(mapping.values()) != _VALID_SENTIMENTS:
+        raise ValueError(f"absa_config.json id2label must cover {_VALID_SENTIMENTS}, got {mapping}")
+    return mapping
+
+
+def not_mentioned() -> dict[str, Any]:
+    """Aspect result for an aspect the review does not mention."""
+    return {"detected": False, "sentiment": "not_mentioned", "confidence": 0.0}
+
+
+class ABSAModule:
+    """Agent 3: per-aspect detection + sentiment."""
+
+    def __init__(
+        self,
+        model_path: str | Path,
+        tokenizer_path: str | Path,
+        config_path: str | Path,
+        *,
+        session: InferenceSessionLike | None = None,
+        tokenizer: LightTokenizer | None = None,
+        id2label: dict[int, str] | None = None,
+        detector: AspectDetector | None = None,
+    ) -> None:
+        """
+        Args:
+            detector: Learned mention detector. Without it, keyword matching decides
+                which aspects are mentioned (test/legacy behaviour).
+        """
+        self.id2label = id2label or load_id2label(config_path)
+        self.tokenizer = tokenizer or LightTokenizer(tokenizer_path)
+        self.session = session or create_session(model_path)
+        self.detector = detector
+
+    @staticmethod
+    def keyword_detect(text: str, aspect: str) -> bool:
+        """Fast pre-filter: is the aspect plausibly mentioned?"""
+        return bool(_KEYWORD_PATTERNS[aspect].search(text))
+
+    def classify_aspects(self, text: str, aspects: list[str]) -> dict[str, dict[str, Any]]:
+        """Classify sentiment for several aspects of one review in a single batch."""
+        if not aspects:
+            return {}
+        encoded = self.tokenizer.encode_pairs([(text, ASPECT_PHRASES[a]) for a in aspects])
+        logits = np.asarray(self.session.run(None, build_feed(self.session, encoded))[0])
+        probs = softmax(logits.astype(np.float64), axis=-1)
+
+        label2id = {label: idx for idx, label in self.id2label.items()}
+        results: dict[str, dict[str, Any]] = {}
+        for row, aspect in zip(probs, aspects, strict=True):
+            pred_idx = int(np.argmax(row))
+            results[aspect] = {
+                "detected": True,
+                "sentiment": self.id2label[pred_idx],
+                "confidence": round(float(row[pred_idx]), 4),
+                "prob_neg": round(float(row[label2id["negative"]]), 4),
+                "prob_neu": round(float(row[label2id["neutral"]]), 4),
+                "prob_pos": round(float(row[label2id["positive"]]), 4),
+            }
+        return results
+
+    def mentioned(self, clean_text: str) -> dict[str, bool]:
+        """Which aspects the review mentions: learned detector if loaded, else keywords."""
+        if self.detector is not None:
+            return self.detector.detect(clean_text)
+        return {a: self.keyword_detect(clean_text, a) for a in EARBUD_ASPECTS}
+
+    def process(
+        self, clean_text: str, mentioned: Mapping[str, bool] | None = None
+    ) -> dict[str, Any]:
+        """Return ``{"aspects": {aspect_key: result}}`` for all 10 aspects, in canonical order.
+
+        Args:
+            mentioned: Precomputed mention decisions. Training passes out-of-fold
+                detector decisions here so fusion never sees in-sample detections.
+        """
+        decisions = mentioned if mentioned is not None else self.mentioned(clean_text)
+        detected = [a for a in EARBUD_ASPECTS if decisions.get(a, False)]
+        classified = self.classify_aspects(clean_text, detected)
+        return {"aspects": {a: classified.get(a, not_mentioned()) for a in EARBUD_ASPECTS}}

backend/app/ml/agents/aspect_detector.py

@@ -0,0 +1,116 @@
+"""Learned aspect detector — decides WHICH of the 10 earbud aspects a review mentions.
+
+Replaces the keyword-only pre-filter of Agent 3 (PRD §9). Keyword lists alone
+reached detection F1 0.54 (build quality) and 0.66 (connectivity) on the round-1
+test split: durability phrasings were missed and generic words ("pair of
+headphones") fired falsely.
+
+Model, per aspect: logistic regression on word 1–2-gram TF-IDF features plus
+the keyword hits of ALL aspects as extra features, with a decision threshold
+chosen by product-grouped out-of-fold F1 on training rows only
+(``training/vera_training/aspects.py``). Features are built by
+:func:`detector_features`, shared by training and serving.
+
+SECURITY: the artifact is a joblib pickle — only load files produced by the VERA
+training pipeline.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any, Final
+
+import joblib
+import numpy as np
+import scipy.sparse as sp
+import sklearn
+
+from app.ml.agents.absa import ABSAModule
+from app.ml.constants import EARBUD_ASPECTS
+
+#: Scale of the keyword indicator columns relative to TF-IDF values (≈ unit-norm rows).
+KEYWORD_FEATURE_WEIGHT: Final = 3.0
+ARTIFACT_KEYS: Final = frozenset({"vectorizer", "models", "thresholds", "aspects"})
+
+
+def keyword_matrix(texts: Sequence[str]) -> np.ndarray:
+    """``(n_texts, 10)`` float matrix of keyword hits in canonical aspect order."""
+    return np.array(
+        [[ABSAModule.keyword_detect(text, aspect) for aspect in EARBUD_ASPECTS] for text in texts],
+        dtype=np.float64,
+    )
+
+
+def detector_features(vectorizer: Any, texts: Sequence[str]) -> sp.csr_matrix:
+    """TF-IDF features concatenated with weighted keyword indicators."""
+    tfidf = vectorizer.transform(list(texts))
+    keywords = sp.csr_matrix(keyword_matrix(texts) * KEYWORD_FEATURE_WEIGHT)
+    return sp.hstack([tfidf, keywords]).tocsr()
+
+
+class AspectDetector:
+    """Per-aspect probabilities and thresholded mention decisions."""
+
+    def __init__(self, artifact_path: str | Path) -> None:
+        artifacts = joblib.load(artifact_path)
+        self._init_from(artifacts, source=str(artifact_path))
+
+    @classmethod
+    def from_components(
+        cls,
+        vectorizer: Any,
+        models: dict[str, Any],
+        thresholds: dict[str, float],
+    ) -> AspectDetector:
+        """Build from in-memory objects (training code and tests)."""
+        instance = cls.__new__(cls)
+        instance._init_from(
+            {
+                "vectorizer": vectorizer,
+                "models": models,
+                "thresholds": thresholds,
+                "aspects": list(EARBUD_ASPECTS),
+            },
+            source="memory",
+        )
+        return instance
+
+    def _init_from(self, artifacts: dict[str, Any], source: str) -> None:
+        missing = ARTIFACT_KEYS - set(artifacts)
+        if missing:
+            raise ValueError(f"{source} is missing keys {sorted(missing)}")
+        if tuple(artifacts["aspects"]) != EARBUD_ASPECTS:
+            raise ValueError(f"{source} aspect order differs from EARBUD_ASPECTS; retrain it")
+        trained_with = artifacts.get("sklearn_version")
+        if trained_with is not None and trained_with != sklearn.__version__:
+            raise ValueError(
+                f"{source} trained with scikit-learn {trained_with}, runtime has "
+                f"{sklearn.__version__}; retrain with the pinned version."
+            )
+        self.vectorizer = artifacts["vectorizer"]
+        self.models: dict[str, Any] = artifacts["models"]
+        self.thresholds: dict[str, float] = {
+            k: float(v) for k, v in artifacts["thresholds"].items()
+        }
+
+    def probabilities(self, texts: Sequence[str]) -> np.ndarray:
+        """``(n_texts, 10)`` mention probabilities in canonical aspect order."""
+        features = detector_features(self.vectorizer, texts)
+        return np.column_stack(
+            [self.models[aspect].predict_proba(features)[:, 1] for aspect in EARBUD_ASPECTS]
+        )
+
+    def decide(self, probabilities: np.ndarray) -> list[dict[str, bool]]:
+        """Apply per-aspect thresholds to a probability matrix."""
+        return [
+            {
+                aspect: bool(row[i] >= self.thresholds[aspect])
+                for i, aspect in enumerate(EARBUD_ASPECTS)
+            }
+            for row in np.atleast_2d(probabilities)
+        ]
+
+    def detect(self, text: str) -> dict[str, bool]:
+        """Mention decisions for one review."""
+        return self.decide(self.probabilities([text]))[0]

backend/app/ml/agents/consistency.py

@@ -0,0 +1,40 @@
+"""Agent 4 — Star-Text Consistency Scorer (PRD §9, Agent 4).
+
+Measures how far the clicked star rating is from the text-derived experience
+score. A large gap means the star is unreliable.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Final
+
+from app.ml.constants import SCORE_RANGE
+
+#: Gap (in stars) beyond which the star is flagged as inflated / deflated.
+INFLATION_THRESHOLD: Final = 1.0
+
+
+class ConsistencyScorer:
+    """Agent 4: star vs text agreement."""
+
+    @staticmethod
+    def process(star_rating: float, experience_score: float) -> dict[str, Any]:
+        """Compare the star with Agent 2's score.
+
+        Returns:
+            ``star_text_gap`` (positive = star higher than text), its absolute
+            value, inflation/deflation flags, ``consistency_score`` in [0, 1]
+            (1 = perfect agreement; max gap on a 1–5 scale is 4), and
+            ``star_trust_weight`` = consistency² (downweights stars harder).
+        """
+        gap = float(star_rating) - float(experience_score)
+        abs_gap = abs(gap)
+        consistency = max(0.0, 1.0 - abs_gap / SCORE_RANGE)
+        return {
+            "star_text_gap": round(gap, 3),
+            "absolute_gap": round(abs_gap, 3),
+            "star_inflated": gap > INFLATION_THRESHOLD,
+            "star_deflated": gap < -INFLATION_THRESHOLD,
+            "consistency_score": round(consistency, 3),
+            "star_trust_weight": round(consistency**2, 3),
+        }

backend/app/ml/agents/domain_rating.py

@@ -0,0 +1,67 @@
+"""Agent 6 — Domain Rating Model (PRD §9, Agent 6).
+
+TF-IDF -> Ridge regression trained on the UNLABELED earbud corpus, using its
+star ratings as a noisy proxy target. Training lives offline in
+``training/vera_training/domain.py`` (with the held-out ASIN leakage guard);
+this module only loads and applies the artifact.
+
+SECURITY: the artifact is a joblib pickle. Only load files produced by the VERA
+training notebooks from your own HF repo — unpickling untrusted files executes code.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import joblib
+import numpy as np
+import sklearn
+
+from app.ml.constants import SCORE_MAX, SCORE_MIN
+
+
+class DomainRatingModel:
+    """Agent 6: text -> implied rating from domain vocabulary."""
+
+    def __init__(self, model_path: str | Path) -> None:
+        try:
+            artifacts = joblib.load(model_path)
+        except ModuleNotFoundError as exc:  # e.g. "numpy._core" from a numpy-2 pickle
+            raise ValueError(
+                f"{model_path} was pickled with incompatible library versions ({exc}). "
+                "Retrain it in the production-pinned .venv (see notebooks/05)."
+            ) from exc
+        if not {"vectorizer", "model"} <= set(artifacts):
+            raise ValueError(f"{model_path} must contain 'vectorizer' and 'model'")
+        trained_with = artifacts.get("sklearn_version")
+        if trained_with is not None and trained_with != sklearn.__version__:
+            raise ValueError(
+                f"{model_path} trained with scikit-learn {trained_with}, runtime has "
+                f"{sklearn.__version__}; retrain with the pinned version."
+            )
+        self.vectorizer = artifacts["vectorizer"]
+        self.model = artifacts["model"]
+
+    @classmethod
+    def from_components(cls, vectorizer: Any, model: Any) -> DomainRatingModel:
+        """Build from in-memory objects (training notebooks and tests)."""
+        instance = cls.__new__(cls)
+        instance.vectorizer = vectorizer
+        instance.model = model
+        return instance
+
+    def process(self, clean_text: str) -> dict[str, Any]:
+        """Predict the implied rating.
+
+        ``domain_confidence`` is the PRD's proxy: 1 − distance to the nearest
+        integer star (Ridge gives no native uncertainty).
+        """
+        features = self.vectorizer.transform([clean_text])
+        raw_score = float(self.model.predict(features)[0])
+        domain_score = float(np.clip(raw_score, SCORE_MIN, SCORE_MAX))
+        domain_confidence = 1.0 - abs(domain_score - round(domain_score))
+        return {
+            "domain_score": round(domain_score, 3),
+            "domain_confidence": round(domain_confidence, 3),
+        }

backend/app/ml/agents/experience_extractor.py

@@ -0,0 +1,79 @@
+"""Agent 2 — Experience Extractor (PRD §9, Agent 2).
+
+Fine-tuned DeBERTa-v3-base regression head, exported to INT8 ONNX, estimates
+the experience score (1–5) from review text alone.
+
+The model input is produced by :func:`format_experience_input`, which the
+fine-tuning notebook imports too — training and serving must see identical
+text. (The PRD prepended ``"{title} [SEP]"`` at inference but trained on the
+bare text: a train/serve skew. Both now use the bare cleaned text.)
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from app.ml.constants import SCORE_MAX, SCORE_MIN
+from app.ml.models.onnx_session import InferenceSessionLike, build_feed, create_session
+from app.ml.models.tokenizer_wrapper import LightTokenizer
+
+
+def format_experience_input(clean_text: str) -> str:
+    """Canonical model input text. Shared with ``vera_training.finetune``."""
+    return clean_text.strip()
+
+
+class ExperienceExtractor:
+    """Agent 2: text -> predicted experience score."""
+
+    def __init__(
+        self,
+        model_path: str | Path,
+        tokenizer_path: str | Path,
+        *,
+        session: InferenceSessionLike | None = None,
+        tokenizer: LightTokenizer | None = None,
+    ) -> None:
+        """Load the ONNX session and tokenizer.
+
+        ``session`` / ``tokenizer`` may be injected (tests); otherwise they are
+        built from the paths.
+        """
+        # max_length comes from the exported tokenizer.json (= fine-tuning length).
+        self.tokenizer = tokenizer or LightTokenizer(tokenizer_path)
+        self.session = session or create_session(model_path)
+
+    def predict_raw(self, texts: list[str]) -> np.ndarray:
+        """Return unclipped regression outputs, shape ``(len(texts),)``."""
+        encoded = self.tokenizer.encode_batch([format_experience_input(t) for t in texts])
+        outputs = self.session.run(None, build_feed(self.session, encoded))
+        logits = np.asarray(outputs[0], dtype=np.float64)
+        return logits.reshape(len(texts), -1)[:, 0]
+
+    @staticmethod
+    def output_from_raw(raw_score: float) -> dict[str, Any]:
+        """Agent 2 output dict from one raw regression value.
+
+        Shared with training, where out-of-fold predictions come from PyTorch
+        rather than this ONNX session.
+
+        Returns:
+            ``predicted_score`` clipped to [1, 5], the ``raw_score``, and
+            ``model_uncertainty`` — distance of the raw output from the nearest
+            0.5 label step (a fixed proxy feature defined by the PRD; the
+            calibrated uncertainty comes from conformal prediction, not this).
+        """
+        raw = float(raw_score)
+        nearest_half = round(raw * 2) / 2
+        return {
+            "predicted_score": float(np.clip(raw, SCORE_MIN, SCORE_MAX)),
+            "raw_score": raw,
+            "model_uncertainty": abs(raw - nearest_half),
+        }
+
+    def predict(self, clean_text: str) -> dict[str, Any]:
+        """Predict one review (see :meth:`output_from_raw` for the fields)."""
+        return self.output_from_raw(float(self.predict_raw([clean_text])[0]))

backend/app/ml/agents/fusion.py

@@ -0,0 +1,105 @@
+"""Agent 7 — Fusion (PRD §10.2).
+
+A bootstrap ensemble of XGBoost regressors maps the 40 fusion features to the
+corrected experience score. The ensemble mean is the point estimate; the
+ensemble standard deviation feeds normalised conformal intervals.
+
+Artifact format — ``fusion_ensemble.json`` (see DECISIONS.md: JSON instead of
+the PRD's joblib pickle, because pickled XGBoost objects break across library
+versions and unpickling is a code-execution risk)::
+
+    {
+      "format_version": 1,
+      "feature_names": [...],            # must equal app.ml.features.FEATURE_NAMES
+      "imputation_means": {...},         # Agent 5 training means
+      "members": ["<xgboost JSON model>", ...],
+      "metadata": {...}                  # free-form: n_train, xgboost_version, ...
+    }
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from pathlib import Path
+from typing import Any, Final
+
+import numpy as np
+import xgboost as xgb
+
+from app.ml.agents.quality import IMPUTATION_KEYS
+from app.ml.constants import SCORE_MAX, SCORE_MIN
+from app.ml.features import FEATURE_NAMES, N_FEATURES
+
+FORMAT_VERSION: Final = 1
+
+
+def save_fusion_artifact(
+    path: str | Path,
+    members: Sequence[xgb.XGBRegressor | xgb.Booster],
+    imputation_means: dict[str, float],
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    """Serialise an ensemble to the portable JSON artifact."""
+    if not members:
+        raise ValueError("ensemble needs at least one member")
+    absent = [k for k in IMPUTATION_KEYS if k not in imputation_means]
+    if absent:
+        raise ValueError(f"imputation_means missing keys: {absent}")
+
+    serialised = []
+    for member in members:
+        booster = member.get_booster() if isinstance(member, xgb.XGBRegressor) else member
+        serialised.append(bytes(booster.save_raw(raw_format="json")).decode("utf-8"))
+
+    payload = {
+        "format_version": FORMAT_VERSION,
+        "feature_names": list(FEATURE_NAMES),
+        "imputation_means": {k: float(v) for k, v in imputation_means.items()},
+        "members": serialised,
+        "metadata": {"xgboost_version": xgb.__version__, **(metadata or {})},
+    }
+    Path(path).write_text(json.dumps(payload), encoding="utf-8")
+
+
+class FusionModel:
+    """Agent 7: loads the ensemble and predicts mean ± std."""
+
+    def __init__(self, artifact_path: str | Path) -> None:
+        payload = json.loads(Path(artifact_path).read_text(encoding="utf-8"))
+
+        if payload.get("format_version") != FORMAT_VERSION:
+            raise ValueError(f"Unsupported fusion artifact version {payload.get('format_version')}")
+        if tuple(payload["feature_names"]) != FEATURE_NAMES:
+            raise ValueError(
+                "Fusion artifact feature_names differ from app.ml.features.FEATURE_NAMES — "
+                "retrain the ensemble with the current feature contract."
+            )
+
+        self.imputation_means: dict[str, float] = payload["imputation_means"]
+        self.metadata: dict[str, Any] = payload.get("metadata", {})
+        self.members: list[xgb.Booster] = []
+        for raw in payload["members"]:
+            booster = xgb.Booster()
+            booster.load_model(bytearray(raw.encode("utf-8")))
+            if booster.num_features() != N_FEATURES:
+                raise ValueError(
+                    f"member expects {booster.num_features()} features, not {N_FEATURES}"
+                )
+            self.members.append(booster)
+        if not self.members:
+            raise ValueError("fusion artifact contains no members")
+
+    def predict_members(self, features: np.ndarray) -> np.ndarray:
+        """Raw member predictions, shape ``(n_members, n_rows)``."""
+        matrix = np.atleast_2d(np.asarray(features, dtype=np.float32))
+        return np.stack([np.asarray(m.inplace_predict(matrix)) for m in self.members])
+
+    def predict(self, features: np.ndarray) -> dict[str, float]:
+        """Predict one review: clipped ensemble mean and member std."""
+        preds = self.predict_members(features)[:, 0]
+        return {
+            "score": float(np.clip(preds.mean(), SCORE_MIN, SCORE_MAX)),
+            "raw_mean": float(preds.mean()),
+            "ensemble_std": float(preds.std()),
+        }

backend/app/ml/agents/preprocessor.py

@@ -0,0 +1,168 @@
+"""Agent 1 — Text Preprocessor (PRD §9, Agent 1).
+
+Gates downstream agents with clean text, language detection, an earbud
+contamination check, and text-quality metrics.
+
+DESIGN CONSTRAINT: zero torch / transformers imports. Loading
+SentenceTransformer here would add ~350 MB RAM and blow Render's 512 MB limit.
+Instead: ``langdetect`` (pure Python) + a tiny sklearn TF-IDF over anchor
+phrases + word-boundary keyword matching.
+"""
+
+from __future__ import annotations
+
+import html
+import re
+from typing import Any, Final
+
+from langdetect import DetectorFactory, detect
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.metrics.pairwise import cosine_similarity
+
+DetectorFactory.seed = 42  # make langdetect deterministic
+
+_URL_RE: Final = re.compile(r"(?:https?://|www\.)\S+", re.IGNORECASE)
+_BREAK_TAG_RE: Final = re.compile(r"<\s*(?:br|/p|p)\s*/?\s*>", re.IGNORECASE)
+_TAG_RE: Final = re.compile(r"<[^>]+>")
+_WS_RE: Final = re.compile(r"\s+")
+_SENTENCE_SPLIT_RE: Final = re.compile(r"[.!?]+")
+
+#: Minimum characters for langdetect to be meaningful (PRD weakness #14).
+MIN_LANGDETECT_CHARS: Final = 10
+#: TF-IDF cosine threshold for the body-text earbud check.
+EARBUD_SIMILARITY_THRESHOLD: Final = 0.25
+
+
+class TextPreprocessor:
+    """Agent 1: clean, validate, and characterise review text."""
+
+    EARBUD_KEYWORDS: Final[tuple[str, ...]] = (
+        "earbud",
+        "earbuds",
+        "earphone",
+        "earphones",
+        "in-ear",
+        "headphone",
+        "headphones",
+        "headset",
+        "tws",
+        "true wireless",
+        "noise cancel",
+        "noise cancelling",
+        "noise canceling",
+        "airpod",
+        "airpods",
+        "bluetooth ear",
+        "over-ear",
+        "on-ear",
+        "anc",
+        "wireless ear",
+    )
+
+    #: Anchor phrases for the TF-IDF cosine fallback (no model download).
+    _ANCHORS: Final[tuple[str, ...]] = (
+        "sound quality is great",
+        "bass is powerful",
+        "noise cancellation works",
+        "battery life lasts hours",
+        "fit in my ear comfortably",
+        "earbud fell out",
+        "earphone sounds",
+        "in-ear headphone",
+        "bluetooth connection drops",
+        "microphone quality for calls",
+        "charging case",
+        "pairing device",
+    )
+
+    def __init__(self) -> None:
+        # Tiny TF-IDF fitted once on the anchors (~KBs of RAM).
+        self._vec = TfidfVectorizer(ngram_range=(1, 2), max_features=2000)
+        self._vec.fit(self._ANCHORS)
+        self._anchor_matrix = self._vec.transform(self._ANCHORS)
+        # Word-boundary match: "anc" must not fire on "dance" / "balance".
+        self._keyword_re = re.compile(
+            r"\b(?:" + "|".join(re.escape(k) for k in self.EARBUD_KEYWORDS) + r")\b",
+            re.IGNORECASE,
+        )
+
+    # ------------------------------------------------------------------
+    @staticmethod
+    def clean_text(text: str) -> str:
+        """Strip URLs and HTML, unescape entities, and normalise whitespace.
+
+        Line-break tags become spaces first — Amazon reviews use ``<br />``
+        heavily and removing it outright glues adjacent words together.
+        """
+        text = html.unescape(text or "")
+        text = _URL_RE.sub(" ", text)
+        text = _BREAK_TAG_RE.sub(" ", text)
+        text = _TAG_RE.sub(" ", text)
+        return _WS_RE.sub(" ", text).strip()
+
+    @staticmethod
+    def detect_language(text: str) -> str:
+        """Return an ISO 639-1 code; fail open to ``'en'``.
+
+        langdetect raises on very short, emoji-only, or punctuation-only text.
+        Those are expected inputs, not retryable errors, so they must never
+        crash a Celery task (PRD weakness #14).
+        """
+        cleaned = text.strip()
+        if len(cleaned) < MIN_LANGDETECT_CHARS:
+            return "en"
+        try:
+            return str(detect(cleaned[:500]))
+        except Exception:
+            return "en"
+
+    def is_earbud_content(self, text: str, title: str | None) -> tuple[bool, float]:
+        """Two-stage contamination filter.
+
+        Stage 1: word-boundary keyword match on title + first 200 body chars.
+        Stage 2: max TF-IDF cosine between body text and earbud anchor phrases.
+
+        Returns:
+            ``(is_earbud, confidence)`` where confidence is 1.0 for a keyword hit.
+        """
+        combined = f"{title or ''} {text[:200]}"
+        if self._keyword_re.search(combined):
+            return True, 1.0
+
+        text_vec = self._vec.transform([text[:500]])
+        max_sim = float(cosine_similarity(text_vec, self._anchor_matrix)[0].max())
+        return max_sim > EARBUD_SIMILARITY_THRESHOLD, round(max_sim, 4)
+
+    @staticmethod
+    def quality_metrics(text: str) -> dict[str, Any]:
+        """Length and lexical-diversity metrics on cleaned text."""
+        words = text.split()
+        sentences = [s for s in (p.strip() for p in _SENTENCE_SPLIT_RE.split(text)) if len(s) > 3]
+        n_words = len(words)
+        n_sentences = max(len(sentences), 1)
+        return {
+            "word_count": n_words,
+            "sentence_count": n_sentences,
+            "avg_sentence_length": n_words / n_sentences,
+            "vocab_richness": len({w.lower() for w in words}) / max(n_words, 1),
+        }
+
+    def process(self, text: str, title: str | None = "") -> dict[str, Any]:
+        """Run Agent 1 on raw review text.
+
+        ``has_url`` is computed on the RAW text: the PRD computed it after URLs
+        were stripped, so it was always ``False``.
+        """
+        raw = text or ""
+        clean = self.clean_text(raw)
+        lang = self.detect_language(clean)
+        is_earbud, earbud_conf = self.is_earbud_content(clean, title)
+        return {
+            "clean_text": clean,
+            "language": lang,
+            "is_english": lang == "en",
+            "is_earbud": is_earbud,
+            "earbud_confidence": earbud_conf,
+            "has_url": bool(_URL_RE.search(raw)),
+            **self.quality_metrics(clean),
+        }

backend/app/ml/agents/quality.py

@@ -0,0 +1,131 @@
+"""Agent 5 — Review Quality Assessor (PRD §9, Agent 5).
+
+Interpretable heuristics (not a learned model) scoring how trustworthy a review
+is as a signal. The result weights reviews in product aggregation.
+
+Fixes vs the PRD sketch (see DECISIONS.md):
+* The PRD formula referenced ``quality_score`` inside its own definition via a
+  ``'quality_score' in dir()`` hack (always False there) and double-counted the
+  verified flag. The composite is now a plain weighted average.
+* ``exp(-age/730)`` is an e-folding time, not a "2-year half-life".
+  Freshness now uses ``0.5 ** (age / 730)``, and future dates clip to age 0
+  (previously produced freshness > 1).
+* Missing metadata (``None``) is imputed with training-set means stored in the
+  fusion artifact, never hard zeros (PRD §10.1, weakness #15).
+"""
+
+from __future__ import annotations
+
+import math
+from datetime import UTC, datetime
+from typing import Any, Final
+
+#: Dataset end date — freshness reference point for timestamped corpus reviews.
+CUTOFF_DATE: Final = datetime(2023, 9, 1, tzinfo=UTC)
+FRESHNESS_HALF_LIFE_DAYS: Final = 730.0
+HELPFUL_VOTES_SATURATION: Final = 100
+LENGTH_SATURATION_WORDS: Final = 500
+
+WEIGHTS: Final[dict[str, float]] = {
+    "helpfulness": 0.35,
+    "freshness": 0.25,
+    "length": 0.25,
+    "verified": 0.15,
+}
+QUALITY_FLOOR: Final = 0.05
+
+#: Keys the imputation mapping must provide (training-set means of the agent outputs).
+IMPUTATION_KEYS: Final[tuple[str, ...]] = (
+    "helpfulness_score",
+    "freshness_score",
+    "verified_purchase",
+)
+
+
+def _clip01(value: float) -> float:
+    return min(max(value, 0.0), 1.0)
+
+
+class QualityAssessor:
+    """Agent 5: metadata -> trust weight."""
+
+    def __init__(self, imputation: dict[str, float] | None = None) -> None:
+        """
+        Args:
+            imputation: Training-set means for :data:`IMPUTATION_KEYS`, loaded
+                from ``fusion_ensemble.json``. May be ``None`` only when every
+                review carries full metadata (the offline training corpus).
+        """
+        if imputation is not None:
+            absent = [k for k in IMPUTATION_KEYS if k not in imputation]
+            if absent:
+                raise ValueError(f"imputation mapping missing keys: {absent}")
+        self.imputation = imputation
+
+    def _impute(self, key: str) -> float:
+        if self.imputation is None:
+            raise ValueError(
+                f"Metadata for {key!r} is missing and no imputation means were provided"
+            )
+        return float(self.imputation[key])
+
+    @staticmethod
+    def helpfulness(helpful_votes: int) -> float:
+        """Log-normalised helpful votes, saturating at 100 votes."""
+        votes = max(int(helpful_votes), 0)
+        return _clip01(math.log1p(votes) / math.log1p(HELPFUL_VOTES_SATURATION))
+
+    @staticmethod
+    def freshness(timestamp_ms: int) -> float:
+        """True half-life decay relative to :data:`CUTOFF_DATE`."""
+        review_date = datetime.fromtimestamp(timestamp_ms / 1000, tz=UTC)
+        age_days = max((CUTOFF_DATE - review_date).days, 0)
+        return float(0.5 ** (age_days / FRESHNESS_HALF_LIFE_DAYS))
+
+    def process(
+        self,
+        helpful_votes: int | None,
+        verified: bool | None,
+        timestamp_ms: int | None,
+        word_count: int,
+    ) -> dict[str, Any]:
+        """Score one review; ``None`` inputs are imputed with training means."""
+        imputed: list[str] = []
+
+        if helpful_votes is None:
+            help_score = self._impute("helpfulness_score")
+            imputed.append("helpful_votes")
+        else:
+            help_score = self.helpfulness(helpful_votes)
+
+        if timestamp_ms is None:
+            fresh_score = self._impute("freshness_score")
+            imputed.append("timestamp")
+        else:
+            fresh_score = self.freshness(timestamp_ms)
+
+        if verified is None:
+            verified_score = self._impute("verified_purchase")
+            imputed.append("verified_purchase")
+        else:
+            verified_score = 1.0 if verified else 0.0
+
+        length_score = _clip01(max(word_count, 0) / LENGTH_SATURATION_WORDS)
+
+        quality = (
+            WEIGHTS["helpfulness"] * help_score
+            + WEIGHTS["freshness"] * fresh_score
+            + WEIGHTS["length"] * length_score
+            + WEIGHTS["verified"] * verified_score
+        )
+        quality = min(max(quality, QUALITY_FLOOR), 1.0)
+
+        return {
+            "helpfulness_score": round(help_score, 3),
+            "freshness_score": round(fresh_score, 3),
+            "length_score": round(length_score, 3),
+            "verified_purchase": round(verified_score, 3),
+            "quality_score": round(quality, 3),
+            "trust_weight": round(quality, 3),
+            "imputed_fields": imputed,
+        }

backend/app/ml/calibration/__init__.py

@@ -0,0 +1 @@
+"""Conformal calibration (per review) and Empirical Bayes aggregation (per product)."""

backend/app/ml/calibration/conformal.py

@@ -0,0 +1,80 @@
+"""Production split-conformal interval predictor (PRD §10.3).
+
+Reads the pre-fitted quantile from ``conformal_qhat.json`` — no MAPIE, no
+sklearn, only ``json`` + ``numpy`` (PRD weakness #16). Fitting happens offline
+in ``training/vera_training/calibration.py`` (notebook 06).
+
+Two nonconformity methods are supported (see DECISIONS.md):
+
+* ``absolute``   — score ``|y − ŷ|``; interval ``ŷ ± q̂``. Constant width, so
+  "confidence" only varies through clipping at the scale edges.
+* ``normalized`` — score ``|y − ŷ| / (σ̂ + floor)`` with σ̂ = ensemble std;
+  interval ``ŷ ± q̂·(σ̂ + floor)``. Same finite-sample coverage guarantee, but
+  widths adapt per review, which makes the confidence value informative.
+
+JSON schema::
+
+    {"method": "absolute"|"normalized", "q_hat": float, "alpha": float,
+     "nominal_coverage": float, "empirical_coverage": float,
+     "calibration_n": int, "sigma_floor": float (normalized only)}
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any, Final
+
+import numpy as np
+
+from app.ml.constants import SCORE_MAX, SCORE_MIN, SCORE_RANGE
+
+METHODS: Final = frozenset({"absolute", "normalized"})
+
+
+def interval_confidence(lower: float, upper: float) -> float:
+    """Map interval width to [0, 1]: a 0.5-wide interval -> 0.875, 2-wide -> 0.5."""
+    return float(np.clip(1.0 - (upper - lower) / SCORE_RANGE, 0.0, 1.0))
+
+
+class ConformalPredictor:
+    """Online conformal predictor; loaded once per worker process."""
+
+    def __init__(self, qhat_path: str | Path) -> None:
+        data: dict[str, Any] = json.loads(Path(qhat_path).read_text(encoding="utf-8"))
+        self.method: str = data.get("method", "absolute")
+        if self.method not in METHODS:
+            raise ValueError(f"Unknown conformal method {self.method!r}")
+        self.q_hat = float(data["q_hat"])
+        if not np.isfinite(self.q_hat) or self.q_hat < 0:
+            raise ValueError(f"q_hat must be a finite non-negative float, got {self.q_hat}")
+        self.nominal_coverage = float(data["nominal_coverage"])
+        self.sigma_floor = float(data.get("sigma_floor", 0.0))
+        if self.method == "normalized" and self.sigma_floor <= 0:
+            raise ValueError("normalized conformal requires sigma_floor > 0")
+        self._meta = data
+
+    def half_width(self, ensemble_std: float = 0.0) -> float:
+        """Interval half-width for one prediction."""
+        if self.method == "normalized":
+            return self.q_hat * (max(float(ensemble_std), 0.0) + self.sigma_floor)
+        return self.q_hat
+
+    def predict_interval(
+        self, base_pred: float, ensemble_std: float = 0.0
+    ) -> tuple[float, float, float]:
+        """Return ``(lower, upper, confidence)`` clipped to the 1–5 scale."""
+        hw = self.half_width(ensemble_std)
+        lower = float(np.clip(base_pred - hw, SCORE_MIN, SCORE_MAX))
+        upper = float(np.clip(base_pred + hw, SCORE_MIN, SCORE_MAX))
+        return lower, upper, interval_confidence(lower, upper)
+
+    def health_info(self) -> dict[str, Any]:
+        """Calibration metadata for diagnostics."""
+        return {
+            "method": self.method,
+            "q_hat": self.q_hat,
+            "nominal_coverage": self.nominal_coverage,
+            "empirical_coverage": self._meta.get("empirical_coverage"),
+            "calibration_n": self._meta.get("calibration_n"),
+        }

backend/app/ml/calibration/empirical_bayes.py

@@ -0,0 +1,276 @@
+"""Product-level Empirical Bayes aggregation (PRD §11).
+
+A product with one 5.0 review must not outrank a product with 500 reviews
+averaging 4.6. Each product's quality-weighted mean is shrunk toward the global
+mean in proportion to how noisy that mean is (normal–normal model).
+
+Model, for product j with review scores y_ij and trust weights w_ij::
+
+    m_j   = Σ w_ij y_ij / Σ w_ij                  (trust-weighted mean)
+    n_eff = (Σ w_ij)² / Σ w_ij²                   (Kish effective sample size)
+    σ²_j  = max(weighted var_j, 0.05)             (noise floor, PRD weakness #5)
+            — conservative default variance when the product has one review
+    s²_j  = σ²_j / n_eff                          (sampling variance of m_j)
+    B_j   = s²_j / (s²_j + τ²)                    (shrinkage factor)
+    θ̂_j   = (1 − B_j)·m_j + B_j·μ                  (EB estimate)
+    sd_j  = sqrt(B_j · τ²)                        (posterior sd = uncertainty)
+
+Prior (fitted across all products, method of moments)::
+
+    μ  = mean of all review scores
+    τ² = max(Var_j(m_j) − mean_j(s²_j), 0.01)     (between-product variance)
+
+Corrections vs the PRD sketch (see DECISIONS.md): the PRD set τ² to the variance
+of *individual review scores* (that is total variance, not between-product
+variance), used τ² as σ² for single-review products, and reported an ad-hoc
+``sqrt(σ²/n + B²τ²)`` instead of the posterior standard deviation.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Iterable, Mapping, Sequence
+from dataclasses import dataclass, field
+from typing import Any, Final
+
+import numpy as np
+
+from app.ml.constants import EARBUD_ASPECTS, SCORE_MAX, SCORE_MIN
+
+SIGMA_SQ_FLOOR: Final = 0.05
+TAU_SQ_FLOOR: Final = 0.01
+#: Below this many products the between-product variance is not estimable.
+MIN_PRODUCTS_FOR_PRIOR: Final = 3
+#: Weakly informative fallbacks: product means within ~±1 star, reviews within ~±0.7.
+DEFAULT_TAU_SQ: Final = 0.25
+DEFAULT_SIGMA_SQ: Final = 0.5
+
+_SENTIMENT_TIE_ORDER: Final = ("negative", "positive", "neutral")
+
+
+@dataclass(frozen=True)
+class ReviewSignal:
+    """The per-review quantities aggregation needs."""
+
+    product_asin: str
+    corrected_score: float
+    quality_weight: float
+    star_rating: float
+    aspects: Mapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True)
+class EBPrior:
+    global_mean: float
+    tau_sq: float
+    pooled_sigma_sq: float
+    n_products: int
+
+
+@dataclass(frozen=True)
+class ProductAggregate:
+    product_asin: str
+    corrected_score: float
+    uncertainty: float
+    review_count: int
+    trust_weighted_score: float
+    star_average: float | None
+    eb_shrinkage_factor: float
+    aspect_summary: dict[str, Any]
+
+
+@dataclass(frozen=True)
+class _ProductStats:
+    n: int
+    weighted_mean: float
+    weighted_var: float | None  # None when n == 1
+    n_eff: float
+
+
+def _product_stats(scores: Sequence[float], weights: Sequence[float]) -> _ProductStats:
+    y = np.asarray(scores, dtype=np.float64)
+    w = np.asarray(weights, dtype=np.float64)
+    if y.size == 0:
+        raise ValueError("cannot aggregate a product with no reviews")
+    if y.shape != w.shape:
+        raise ValueError("scores and weights must have the same length")
+    w = np.where(np.isfinite(w) & (w > 0), w, 0.0)
+    if w.sum() <= 0:  # all weights unusable -> uniform
+        w = np.ones_like(y)
+    w = w / w.sum()
+
+    mean = float(np.dot(w, y))
+    n_eff = float(1.0 / np.sum(w**2))
+    var: float | None = None
+    if y.size > 1:
+        denom = 1.0 - float(np.sum(w**2))  # unbiased reliability-weighted variance
+        var = float(np.dot(w, (y - mean) ** 2) / denom) if denom > 1e-12 else 0.0
+    return _ProductStats(n=int(y.size), weighted_mean=mean, weighted_var=var, n_eff=n_eff)
+
+
+class EmpiricalBayesAggregator:
+    """Fit a prior across products, then shrink each product's mean toward it."""
+
+    def __init__(self) -> None:
+        self.prior: EBPrior | None = None
+
+    # ------------------------------------------------------------------
+    def fit(self, grouped: Mapping[str, tuple[Sequence[float], Sequence[float]]]) -> EBPrior:
+        """Estimate μ, τ² and the pooled within-product variance.
+
+        Args:
+            grouped: ``asin -> (corrected_scores, quality_weights)``.
+        """
+        if not grouped:
+            raise ValueError("fit() needs at least one product")
+        stats = {asin: _product_stats(*sw) for asin, sw in grouped.items()}
+        all_scores = np.concatenate([np.asarray(s, dtype=np.float64) for s, _ in grouped.values()])
+
+        multi = [s for s in stats.values() if s.weighted_var is not None]
+        if multi:
+            dof = np.array([s.n - 1 for s in multi], dtype=np.float64)
+            variances = np.array([s.weighted_var for s in multi], dtype=np.float64)
+            pooled = float(np.dot(dof, variances) / dof.sum())
+        else:
+            pooled = DEFAULT_SIGMA_SQ
+        pooled = max(pooled, SIGMA_SQ_FLOOR)
+
+        if len(stats) >= MIN_PRODUCTS_FOR_PRIOR:
+            means = np.array([s.weighted_mean for s in stats.values()])
+            sampling = np.array([self._sigma_sq(s, pooled) / s.n_eff for s in stats.values()])
+            tau_sq = max(float(means.var(ddof=1) - sampling.mean()), TAU_SQ_FLOOR)
+        else:
+            tau_sq = DEFAULT_TAU_SQ
+
+        self.prior = EBPrior(
+            global_mean=float(all_scores.mean()),
+            tau_sq=tau_sq,
+            pooled_sigma_sq=pooled,
+            n_products=len(stats),
+        )
+        return self.prior
+
+    @staticmethod
+    def _sigma_sq(stats: _ProductStats, pooled_sigma_sq: float) -> float:
+        # A one-review product has no within-product variance estimate. The
+        # observed pooled variance can be deceptively near zero when existing
+        # multi-review products are unanimous, so using it here would allow an
+        # isolated 5.0 to outrank a well-supported 4.6. Use the documented
+        # weakly informative variance instead; observed variance takes over as
+        # soon as a product has two reviews.
+        raw = (
+            stats.weighted_var
+            if stats.weighted_var is not None
+            else max(pooled_sigma_sq, DEFAULT_SIGMA_SQ)
+        )
+        return max(raw, SIGMA_SQ_FLOOR)
+
+    # ------------------------------------------------------------------
+    def aggregate_product(
+        self,
+        corrected_scores: Sequence[float],
+        quality_weights: Sequence[float],
+        aspect_sentiments: Sequence[Mapping[str, Any]],
+        star_ratings: Sequence[float] | None = None,
+        product_asin: str = "",
+    ) -> ProductAggregate:
+        """Compute the EB-shrunk aggregate for one product. Requires :meth:`fit` first."""
+        if self.prior is None:
+            raise RuntimeError("call fit() before aggregate_product()")
+        prior = self.prior
+        stats = _product_stats(corrected_scores, quality_weights)
+
+        s_sq = self._sigma_sq(stats, prior.pooled_sigma_sq) / stats.n_eff
+        shrink = float(np.clip(s_sq / (s_sq + prior.tau_sq), 0.0, 1.0))
+        eb_score = (1.0 - shrink) * stats.weighted_mean + shrink * prior.global_mean
+        posterior_sd = float(np.sqrt(shrink * prior.tau_sq))
+
+        stars = list(star_ratings) if star_ratings is not None else []
+        return ProductAggregate(
+            product_asin=product_asin,
+            corrected_score=round(float(np.clip(eb_score, SCORE_MIN, SCORE_MAX)), 2),
+            uncertainty=round(posterior_sd, 3),
+            review_count=stats.n,
+            trust_weighted_score=round(stats.weighted_mean, 2),
+            star_average=round(float(np.mean(stars)), 2) if stars else None,
+            eb_shrinkage_factor=round(shrink, 3),
+            aspect_summary=aggregate_aspects(aspect_sentiments),
+        )
+
+    def aggregate_all(
+        self, reviews: Iterable[ReviewSignal]
+    ) -> tuple[EBPrior, dict[str, ProductAggregate]]:
+        """Fit the prior on every product and aggregate each one."""
+        by_asin: dict[str, list[ReviewSignal]] = {}
+        for review in reviews:
+            by_asin.setdefault(review.product_asin, []).append(review)
+        if not by_asin:
+            raise ValueError("aggregate_all() needs at least one review")
+
+        prior = self.fit(
+            {
+                asin: ([r.corrected_score for r in rs], [r.quality_weight for r in rs])
+                for asin, rs in by_asin.items()
+            }
+        )
+        aggregates = {
+            asin: self.aggregate_product(
+                [r.corrected_score for r in rs],
+                [r.quality_weight for r in rs],
+                [r.aspects for r in rs],
+                star_ratings=[r.star_rating for r in rs],
+                product_asin=asin,
+            )
+            for asin, rs in by_asin.items()
+        }
+        return prior, aggregates
+
+
+def rank_products(aggregates: Mapping[str, ProductAggregate]) -> list[str]:
+    """ASINs best-first: EB score desc, then more reviews, then lower uncertainty, then ASIN."""
+    return sorted(
+        aggregates,
+        key=lambda a: (
+            -aggregates[a].corrected_score,
+            -aggregates[a].review_count,
+            aggregates[a].uncertainty,
+            a,
+        ),
+    )
+
+
+def aggregate_aspects(aspect_sentiments: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
+    """Per-aspect mention rate, sentiment counts, net sentiment, and majority label.
+
+    ``net_sentiment`` = (positive − negative) / mentions, in [−1, 1] (drives the
+    radar chart). Ties for the majority label resolve negative > positive >
+    neutral, so a split verdict surfaces the complaint. Reviews whose aspect
+    dict lacks a key (e.g. legacy rows) count as "not mentioned".
+    """
+    total = len(aspect_sentiments)
+    summary: dict[str, Any] = {}
+    for aspect in EARBUD_ASPECTS:
+        mentions = [
+            a[aspect]["sentiment"]
+            for a in aspect_sentiments
+            if isinstance(a.get(aspect), Mapping) and a[aspect].get("detected")
+        ]
+        counts = Counter(mentions)
+        n = len(mentions)
+        if n == 0:
+            dominant = "not_mentioned"
+        else:
+            top = max(counts.values())
+            dominant = next(s for s in _SENTIMENT_TIE_ORDER if counts.get(s, 0) == top)
+        summary[aspect] = {
+            "mention_count": n,
+            "detected_fraction": round(n / total, 3) if total else 0.0,
+            "positive": counts.get("positive", 0),
+            "negative": counts.get("negative", 0),
+            "neutral": counts.get("neutral", 0),
+            "net_sentiment": (
+                round((counts.get("positive", 0) - counts.get("negative", 0)) / n, 3) if n else 0.0
+            ),
+            "dominant_sentiment": dominant,
+        }
+    return summary

backend/app/ml/constants.py

@@ -0,0 +1,142 @@
+"""Constants shared by production agents, the training package, and tests.
+
+Aspect keys are snake_case everywhere (API, DB JSON, features, labels). The
+natural-language phrase in :data:`ASPECT_PHRASES` is only used as the second
+sentence fed to the ABSA model.
+"""
+
+from __future__ import annotations
+
+from typing import Final, Literal
+
+#: Canonical, ordered aspect keys. ORDER IS PART OF THE FUSION FEATURE CONTRACT —
+#: never reorder without retraining the ensemble.
+EARBUD_ASPECTS: Final[tuple[str, ...]] = (
+    "sound_quality",
+    "bass",
+    "battery_life",
+    "comfort",
+    "fit",
+    "build_quality",
+    "connectivity",
+    "microphone",
+    "noise_cancellation",
+    "value",
+)
+
+#: Aspect key -> phrase given to the ABSA model as the aspect sentence.
+ASPECT_PHRASES: Final[dict[str, str]] = {key: key.replace("_", " ") for key in EARBUD_ASPECTS}
+
+#: Keyword pre-filter per aspect. Matched on word boundaries at the START of a
+#: word (so "isolat" matches "isolation" but "fit" does not match "benefit").
+#: "cheap" is deliberately only under `value`: "cheap plastic" is still caught
+#: by `build_quality` through "plastic".
+ASPECT_KEYWORDS: Final[dict[str, tuple[str, ...]]] = {
+    "sound_quality": (
+        "sound",
+        "audio",
+        "music",
+        "clarity",
+        "treble",
+        "highs",
+        "mids",
+        "tinny",
+        "muddy",
+    ),
+    "bass": ("bass", "low end", "boom", "thump", "sub-bass", "subwoofer"),
+    "battery_life": ("battery", "charge", "charging", "playtime", "endurance", "hours of"),
+    "comfort": ("comfort", "uncomfortable", "ear pain", "hurt", "wear", "wearing", "sore"),
+    "fit": ("fit", "fits", "stay in", "stays in", "fall out", "falls out", "ear tip", "seal"),
+    # Durability counts as build quality in the labeling protocol ("broke after a week",
+    # "connector bent"), which the original 7 keywords missed (train-split recall 0.35).
+    "build_quality": (
+        "build",
+        "built",
+        "construct",
+        "durab",
+        "sturd",
+        "flims",
+        "fragil",
+        "well made",
+        "well-made",
+        "cheaply made",
+        "poorly made",
+        "plastic",
+        "material",
+        "hinge",
+        "crack",
+        "broke",
+        "break",
+        "snapped",
+        "fell apart",
+        "falling apart",
+        "came apart",
+        "cord",
+        "cable",
+        "wire",
+        "jack",
+        "connector",
+        "stopped working",
+        "quit working",
+        "died",
+        "defect",
+        "peel",
+        "wore out",
+        "worn out",
+    ),
+    # "pair" / "range" were removed: "a pair of headphones" and "price range" made
+    # connectivity precision 0.52 on the train split.
+    "connectivity": (
+        "bluetooth",
+        "connect",
+        "disconnect",
+        "reconnect",
+        "pairing",
+        "paired",
+        "pair with",
+        "pair to",
+        "latency",
+        "lag",
+        "delay",
+        "drops out",
+        "dropped out",
+        "dropout",
+        "cuts out",
+        "cutting out",
+        "signal",
+        "out of range",
+        "sync",
+        "stutter",
+        "multipoint",
+    ),
+    "microphone": ("mic", "microphone", "calls", "phone call", "voice", "recording"),
+    "noise_cancellation": (
+        "anc",
+        "noise cancel",
+        "noise-cancel",
+        "passive isolation",
+        "isolat",
+        "block out",
+        "blocks out",
+        "transparency mode",
+    ),
+    "value": ("price", "value", "worth", "money", "cheap", "expensive", "cost", "budget"),
+}
+
+Sentiment = Literal["positive", "negative", "neutral", "not_mentioned"]
+
+#: Numeric encoding of aspect sentiment for fusion features (PRD §10.1).
+SENTIMENT_TO_NUMERIC: Final[dict[str, int]] = {
+    "negative": -1,
+    "neutral": 0,
+    "positive": 1,
+    "not_mentioned": 0,
+}
+
+#: Valid experience-score label values produced by the labeler (PRD §7.2).
+LABEL_SCORE_VALUES: Final[tuple[float, ...]] = (1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0)
+
+SCORE_MIN: Final[float] = 1.0
+SCORE_MAX: Final[float] = 5.0
+#: Largest possible distance on the 1–5 scale.
+SCORE_RANGE: Final[float] = SCORE_MAX - SCORE_MIN

backend/app/ml/features.py

@@ -0,0 +1,125 @@
+"""Fusion feature contract (PRD §10.1).
+
+This module is the SINGLE source of truth for the 40-dimensional feature vector.
+Notebook 05 (training), notebook 06 (calibration), and the production pipeline
+all call :func:`build_feature_vector`, so train/serve skew is impossible by
+construction. Changing the order or meaning of any feature requires retraining
+the ensemble and re-fitting the calibrator.
+
+Imputation of missing metadata happens upstream in Agent 5
+(:class:`~app.ml.agents.quality.QualityAssessor`); the PRD's
+``build_feature_vector(imputation=...)`` parameter was never actually used.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Final
+
+import numpy as np
+
+from app.ml.constants import EARBUD_ASPECTS, SENTIMENT_TO_NUMERIC
+
+FEATURE_NAMES: Final[tuple[str, ...]] = (
+    # Base signals (3)
+    "star_rating",
+    "experience_score",
+    "domain_score",
+    # Consistency signals (4)
+    "star_text_gap",
+    "absolute_gap",
+    "star_inflated",
+    "consistency_score",
+    # Quality signals (4)
+    "quality_score",
+    "helpfulness_score",
+    "freshness_score",
+    "verified_purchase",
+    # Text quality signals (4)
+    "word_count_norm",
+    "vocab_richness",
+    "avg_sentence_length_norm",
+    "experience_model_uncertainty",
+    # Aspect detection (10) + aspect sentiment (10)
+    *(f"aspect_detected__{a}" for a in EARBUD_ASPECTS),
+    *(f"aspect_sentiment__{a}" for a in EARBUD_ASPECTS),
+    # Interactions (5)
+    "star_x_experience",
+    "abs_star_minus_domain",
+    "experience_x_quality",
+    "consistency_x_trust",
+    "domain_confidence_x_domain_score",
+)
+
+N_FEATURES: Final = len(FEATURE_NAMES)
+
+#: Feature-name groups used by the ablation study (PRD §13.2).
+FEATURE_GROUPS: Final[dict[str, tuple[str, ...]]] = {
+    "star": ("star_rating",),
+    "experience": ("experience_score", "experience_model_uncertainty", "star_x_experience"),
+    "domain": ("domain_score", "abs_star_minus_domain", "domain_confidence_x_domain_score"),
+    "consistency": ("star_text_gap", "absolute_gap", "star_inflated", "consistency_score"),
+    "quality": (
+        "quality_score",
+        "helpfulness_score",
+        "freshness_score",
+        "verified_purchase",
+        "word_count_norm",
+        "vocab_richness",
+        "avg_sentence_length_norm",
+        "experience_x_quality",
+        "consistency_x_trust",
+    ),
+    "absa": tuple(n for n in FEATURE_NAMES if n.startswith("aspect_")),
+}
+
+
+def build_feature_vector(agent_outputs: dict[str, Any]) -> np.ndarray:
+    """Assemble the fusion features from Agents 1–6.
+
+    Args:
+        agent_outputs: Mapping with keys ``preprocessor``, ``experience_extractor``,
+            ``absa``, ``consistency``, ``quality``, ``domain_rating`` (each the
+            agent's ``process``/``predict`` output) and ``star_rating``.
+
+    Returns:
+        ``float32`` array of shape ``(N_FEATURES,)``.
+    """
+    pre = agent_outputs["preprocessor"]
+    exp = agent_outputs["experience_extractor"]
+    absa = agent_outputs["absa"]["aspects"]
+    cons = agent_outputs["consistency"]
+    qual = agent_outputs["quality"]
+    dom = agent_outputs["domain_rating"]
+    star = float(agent_outputs["star_rating"])
+
+    feats: list[float] = [
+        star,
+        exp["predicted_score"],
+        dom["domain_score"],
+        cons["star_text_gap"],
+        cons["absolute_gap"],
+        float(cons["star_inflated"]),
+        cons["consistency_score"],
+        qual["quality_score"],
+        qual["helpfulness_score"],
+        qual["freshness_score"],
+        float(qual["verified_purchase"]),
+        pre["word_count"] / 500,
+        pre["vocab_richness"],
+        pre["avg_sentence_length"] / 20,
+        exp["model_uncertainty"],
+    ]
+    feats += [float(absa[a]["detected"]) for a in EARBUD_ASPECTS]
+    feats += [float(SENTIMENT_TO_NUMERIC[absa[a]["sentiment"]]) for a in EARBUD_ASPECTS]
+    feats += [
+        star * exp["predicted_score"],
+        abs(star - dom["domain_score"]),
+        exp["predicted_score"] * qual["quality_score"],
+        cons["consistency_score"] * qual["trust_weight"],
+        dom["domain_confidence"] * dom["domain_score"],
+    ]
+
+    vector = np.asarray(feats, dtype=np.float32)
+    if vector.shape != (N_FEATURES,):  # guards against silent contract drift
+        raise AssertionError(f"feature vector has {vector.shape[0]} dims, expected {N_FEATURES}")
+    return vector

backend/app/ml/models/__init__.py

@@ -0,0 +1 @@
+"""Model artifact management: tokenizer wrapper, ONNX session helpers, HF Hub loader."""

backend/app/ml/models/loader.py

@@ -0,0 +1,128 @@
+"""Model artifact registry and HuggingFace Hub download (PRD §17.3).
+
+Production policy (user decision, see DECISIONS.md): the worker **hard-fails**
+if any artifact is missing. There is no heuristic fallback.
+
+CLI (from ``backend/``)::
+
+    python -m app.ml.models.loader          # download missing artifacts, then verify
+    python -m app.ml.models.loader --check  # verify only, no network
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from pathlib import Path
+from typing import Final
+
+logger = logging.getLogger(__name__)
+
+#: Every file the production pipeline needs, with the notebook that produces it.
+REQUIRED_ARTIFACTS: Final[dict[str, str]] = {
+    "experience_model.onnx": "notebooks/04_finetune_experience.ipynb (INT8 ONNX)",
+    "experience_tokenizer.json": "notebooks/04_finetune_experience.ipynb",
+    "absa_model.onnx": "notebooks/04_finetune_experience.ipynb (ABSA export section)",
+    "absa_tokenizer.json": "notebooks/04_finetune_experience.ipynb (ABSA export section)",
+    "absa_config.json": "notebooks/04_finetune_experience.ipynb (ABSA export section)",
+    "aspect_detector.joblib": "training/scripts/train_all.py --stages aspects",
+    "domain_tfidf.joblib": "notebooks/05_train_fusion.ipynb (domain model section)",
+    "fusion_ensemble.json": "notebooks/05_train_fusion.ipynb",
+    "conformal_qhat.json": "notebooks/06_calibration.ipynb",
+}
+
+
+class ModelArtifactsMissingError(RuntimeError):
+    """Raised when required model artifacts are absent after any download attempt."""
+
+    def __init__(self, model_dir: Path, missing: list[str]) -> None:
+        self.model_dir = model_dir
+        self.missing = missing
+        details = "\n".join(
+            f"  - {name}  <- produced by {REQUIRED_ARTIFACTS[name]}" for name in missing
+        )
+        super().__init__(
+            f"{len(missing)} required model artifact(s) missing in {model_dir}:\n{details}\n"
+            "Run notebooks 04 (Colab) and 05-06 (local), then upload the files to "
+            "HF_MODEL_REPO or copy them into MODEL_DIR."
+        )
+
+
+def missing_artifacts(model_dir: Path) -> list[str]:
+    """Return the names of required artifacts not present (or empty) in ``model_dir``."""
+    return [
+        name
+        for name in REQUIRED_ARTIFACTS
+        if not (model_dir / name).is_file() or (model_dir / name).stat().st_size == 0
+    ]
+
+
+def artifact_status(model_dir: Path) -> dict[str, bool]:
+    """Map each required artifact to whether it is present. Used by ``/health``."""
+    missing = set(missing_artifacts(model_dir))
+    return {name: name not in missing for name in REQUIRED_ARTIFACTS}
+
+
+def ensure_models(model_dir: Path, hf_repo: str = "", hf_token: str = "") -> dict[str, Path]:
+    """Download missing artifacts from the Hub (if configured) and verify all exist.
+
+    Args:
+        model_dir: Local artifact directory (created if absent).
+        hf_repo: ``user/repo`` on HuggingFace Hub; empty disables downloading.
+        hf_token: Optional token for private repos.
+
+    Returns:
+        Mapping of artifact name to its local path.
+
+    Raises:
+        ModelArtifactsMissingError: If anything is still missing afterwards.
+    """
+    model_dir.mkdir(parents=True, exist_ok=True)
+    missing = missing_artifacts(model_dir)
+
+    if missing and hf_repo:
+        from huggingface_hub import hf_hub_download  # imported lazily: network-only path
+
+        for name in missing:
+            logger.info("Downloading %s from %s ...", name, hf_repo)
+            try:
+                hf_hub_download(
+                    repo_id=hf_repo,
+                    filename=name,
+                    local_dir=model_dir,
+                    token=hf_token or None,
+                )
+            except Exception as exc:  # report every failure, then fail once below
+                logger.error("Failed to download %s: %s", name, exc)
+        missing = missing_artifacts(model_dir)
+
+    if missing:
+        raise ModelArtifactsMissingError(model_dir, missing)
+
+    logger.info("All %d model artifacts ready in %s", len(REQUIRED_ARTIFACTS), model_dir)
+    return {name: model_dir / name for name in REQUIRED_ARTIFACTS}
+
+
+def _main(argv: list[str] | None = None) -> int:
+    from app.config import settings
+
+    parser = argparse.ArgumentParser(description="Download / verify VERA model artifacts.")
+    parser.add_argument("--check", action="store_true", help="verify only; never download")
+    args = parser.parse_args(argv)
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
+
+    try:
+        ensure_models(
+            settings.MODEL_DIR,
+            hf_repo="" if args.check else settings.HF_MODEL_REPO,
+            hf_token=settings.HF_TOKEN,
+        )
+    except ModelArtifactsMissingError as exc:
+        logger.error("%s", exc)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(_main())

backend/app/ml/models/onnx_session.py

@@ -0,0 +1,63 @@
+"""Thin onnxruntime helpers shared by the transformer agents (Agents 2 and 3)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Protocol
+
+import numpy as np
+
+
+class InferenceSessionLike(Protocol):
+    """Subset of ``onnxruntime.InferenceSession`` the agents rely on (eases testing)."""
+
+    def get_inputs(self) -> list[Any]: ...
+
+    def run(
+        self, output_names: list[str] | None, input_feed: dict[str, np.ndarray]
+    ) -> list[Any]: ...
+
+
+def create_session(model_path: str | Path, intra_op_threads: int = 2) -> InferenceSessionLike:
+    """Create a CPU inference session with full graph optimisation."""
+    import onnxruntime as ort  # local import keeps module import cheap for tests
+
+    path = Path(model_path)
+    if not path.is_file():
+        raise FileNotFoundError(f"ONNX model not found: {path}")
+    opts = ort.SessionOptions()
+    opts.intra_op_num_threads = intra_op_threads
+    opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+    session: InferenceSessionLike = ort.InferenceSession(
+        str(path), sess_options=opts, providers=["CPUExecutionProvider"]
+    )
+    return session
+
+
+def build_feed(
+    session: InferenceSessionLike, encoded: dict[str, np.ndarray]
+) -> dict[str, np.ndarray]:
+    """Select exactly the inputs the graph declares.
+
+    DeBERTa-v3 has ``type_vocab_size=0``, so its ONNX export has no
+    ``token_type_ids`` input; feeding an undeclared input makes onnxruntime
+    raise ``INVALID_ARGUMENT``. Conversely, BERT-style exports require it.
+    """
+    feed: dict[str, np.ndarray] = {}
+    for spec in session.get_inputs():
+        name = spec.name
+        if name in encoded:
+            feed[name] = encoded[name].astype(np.int64, copy=False)
+        elif name == "token_type_ids":
+            feed[name] = np.zeros_like(encoded["input_ids"], dtype=np.int64)
+        else:
+            raise KeyError(f"ONNX graph expects unsupported input {name!r}")
+    return feed
+
+
+def softmax(logits: np.ndarray, axis: int = -1) -> np.ndarray:
+    """Numerically stable softmax (the PRD's ``exp(x)/sum(exp(x))`` overflows for large logits)."""
+    shifted = logits - np.max(logits, axis=axis, keepdims=True)
+    exp = np.exp(shifted)
+    result: np.ndarray = exp / np.sum(exp, axis=axis, keepdims=True)
+    return result

backend/app/ml/models/tokenizer_wrapper.py

@@ -0,0 +1,82 @@
+"""Torch-free tokenizer (PRD §4.4).
+
+Drop-in replacement for ``transformers.AutoTokenizer`` in production. The
+HuggingFace ``tokenizers`` library is a small Rust extension that reads the same
+``tokenizer.json`` a fine-tuned model ships with, but never imports torch.
+
+Differences from the PRD sketch (see DECISIONS.md):
+* Loads from a local ``tokenizer.json`` file (downloaded with the ONNX model)
+  instead of ``Tokenizer.from_pretrained`` — no network at inference time, and
+  the base ``microsoft/deberta-v3-base`` repo does not ship a ``tokenizer.json``.
+* Pads to the longest sequence in the batch instead of always to 512 tokens.
+  The ONNX graphs have dynamic axes, so this is numerically identical and
+  several times faster on a shared CPU.
+* The pad id/token are read from the vocabulary instead of hardcoded to 0.
+* Pair inputs use ``only_first`` truncation so the (short) aspect sentence is
+  never cut — only the review body is truncated.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+from tokenizers import Encoding, Tokenizer
+
+_PAD_CANDIDATES = ("[PAD]", "<pad>", "<PAD>")
+DEFAULT_MAX_LENGTH = 512
+
+
+class LightTokenizer:
+    """Rust-backed tokenizer producing int64 numpy batches for onnxruntime."""
+
+    def __init__(self, tokenizer_path: str | Path, max_length: int | None = None) -> None:
+        """
+        Args:
+            tokenizer_path: ``tokenizer.json`` exported with the model.
+            max_length: Truncation length. ``None`` uses the length stored in the
+                file's truncation settings (written at export so serving matches
+                training), falling back to 512.
+        """
+        path = Path(tokenizer_path)
+        if not path.is_file():
+            raise FileNotFoundError(f"tokenizer.json not found: {path}")
+        self._tok = Tokenizer.from_file(str(path))
+        stored = (self._tok.truncation or {}).get("max_length")
+        self.max_length = int(max_length or stored or DEFAULT_MAX_LENGTH)
+
+        pad_token = next((t for t in _PAD_CANDIDATES if self._tok.token_to_id(t) is not None), None)
+        if pad_token is None:
+            raise ValueError(f"No padding token {_PAD_CANDIDATES} in vocabulary of {path}")
+        self.pad_token = pad_token
+        self.pad_id = int(self._tok.token_to_id(pad_token))
+        self._tok.enable_padding(pad_id=self.pad_id, pad_token=self.pad_token)
+
+    # ------------------------------------------------------------------
+    def __call__(self, text: str, text_pair: str | None = None) -> dict[str, np.ndarray]:
+        """Encode one text (or one text pair) into a batch of size 1."""
+        if text_pair is None:
+            return self.encode_batch([text])
+        return self.encode_pairs([(text, text_pair)])
+
+    def encode_batch(self, texts: list[str]) -> dict[str, np.ndarray]:
+        """Encode single sequences, padded to the longest in the batch."""
+        if not texts:
+            raise ValueError("encode_batch() needs at least one text")
+        self._tok.enable_truncation(max_length=self.max_length, strategy="longest_first")
+        return self._to_arrays(self._tok.encode_batch(texts))
+
+    def encode_pairs(self, pairs: list[tuple[str, str]]) -> dict[str, np.ndarray]:
+        """Encode (first, second) pairs; only the first sequence is ever truncated."""
+        if not pairs:
+            raise ValueError("encode_pairs() needs at least one pair")
+        self._tok.enable_truncation(max_length=self.max_length, strategy="only_first")
+        return self._to_arrays(self._tok.encode_batch(pairs))
+
+    @staticmethod
+    def _to_arrays(encodings: list[Encoding]) -> dict[str, np.ndarray]:
+        return {
+            "input_ids": np.asarray([e.ids for e in encodings], dtype=np.int64),
+            "attention_mask": np.asarray([e.attention_mask for e in encodings], dtype=np.int64),
+            "token_type_ids": np.asarray([e.type_ids for e in encodings], dtype=np.int64),
+        }

backend/app/ml/pipeline.py

@@ -0,0 +1,218 @@
+"""VERA pipeline orchestrator (PRD §8.2).
+
+Runs Agents 1–6, fuses their signals with Agent 7, and wraps the point estimate
+in a conformal interval. Product-level Empirical Bayes aggregation runs
+separately (``app.worker.tasks.aggregate_products_task``) because it needs every
+review of every product.
+
+Model policy: :class:`VERAPipeline` **hard-fails** at construction when any
+artifact is missing (user decision, DECISIONS.md) — see
+:mod:`app.ml.models.loader`.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from app.ml.agents.absa import ABSAModule
+from app.ml.agents.aspect_detector import AspectDetector
+from app.ml.agents.consistency import ConsistencyScorer
+from app.ml.agents.domain_rating import DomainRatingModel
+from app.ml.agents.experience_extractor import ExperienceExtractor
+from app.ml.agents.fusion import FusionModel
+from app.ml.agents.preprocessor import TextPreprocessor
+from app.ml.agents.quality import QualityAssessor
+from app.ml.calibration.conformal import ConformalPredictor
+from app.ml.constants import EARBUD_ASPECTS
+from app.ml.features import build_feature_vector
+from app.ml.models.loader import ensure_models
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class SignalAgents:
+    """Agents 1–6. ``experience`` may be ``None`` when callers supply its output
+    directly (offline training uses out-of-fold predictions)."""
+
+    preprocessor: TextPreprocessor
+    experience: ExperienceExtractor | None
+    absa: ABSAModule
+    consistency: ConsistencyScorer
+    quality: QualityAssessor
+    domain: DomainRatingModel
+
+
+def extract_signals(
+    agents: SignalAgents,
+    *,
+    text: str,
+    star_rating: float,
+    helpful_votes: int | None = None,
+    verified_purchase: bool | None = None,
+    timestamp_ms: int | None = None,
+    product_title: str | None = "",
+    experience_output: dict[str, Any] | None = None,
+    aspect_mentions: dict[str, bool] | None = None,
+) -> dict[str, Any]:
+    """Run Agents 1–6 and return the ``agent_outputs`` mapping for fusion.
+
+    Shared by production and the training notebooks so both build features
+    through exactly the same code path. Training supplies out-of-fold
+    ``experience_output`` and ``aspect_mentions``; production computes both.
+    """
+    pre = agents.preprocessor.process(text, product_title)
+    clean = pre["clean_text"]
+
+    if experience_output is None:
+        if agents.experience is None:
+            raise ValueError("experience_output is required when no ExperienceExtractor is loaded")
+        experience_output = agents.experience.predict(clean)
+
+    quality = agents.quality.process(
+        helpful_votes=helpful_votes,
+        verified=verified_purchase,
+        timestamp_ms=timestamp_ms,
+        word_count=pre["word_count"],
+    )
+    return {
+        "preprocessor": pre,
+        "experience_extractor": experience_output,
+        "absa": agents.absa.process(clean, aspect_mentions),
+        "consistency": agents.consistency.process(
+            star_rating, experience_output["predicted_score"]
+        ),
+        "quality": quality,
+        "domain_rating": agents.domain.process(clean),
+        "star_rating": float(star_rating),
+    }
+
+
+def build_explanation(agent_outputs: dict[str, Any], corrected_score: float) -> str:
+    """Human-readable summary for the ScoreCard (PRD §16.2)."""
+    star = agent_outputs["star_rating"]
+    cons = agent_outputs["consistency"]
+    aspects = agent_outputs["absa"]["aspects"]
+    pre = agent_outputs["preprocessor"]
+
+    def names(sentiment: str) -> list[str]:
+        return [a.replace("_", " ") for a in EARBUD_ASPECTS if aspects[a]["sentiment"] == sentiment]
+
+    parts: list[str] = []
+    if cons["star_inflated"]:
+        parts.append(
+            f"The text implies about {corrected_score:.1f} stars despite the {star:g}-star rating."
+        )
+    elif cons["star_deflated"]:
+        parts.append(
+            f"The text is more positive than the {star:g}-star rating suggests "
+            f"(about {corrected_score:.1f})."
+        )
+    else:
+        parts.append("The star rating and the review text broadly agree.")
+
+    negatives, positives = names("negative"), names("positive")
+    if negatives:
+        parts.append(f"Concerns: {', '.join(negatives)}.")
+    if positives:
+        parts.append(f"Praised: {', '.join(positives)}.")
+    if not pre["is_english"]:
+        parts.append(f"Warning: detected language '{pre['language']}'; scores are English-trained.")
+    if not pre["is_earbud"]:
+        parts.append("Warning: this review may not be about earbuds or headphones.")
+    return " ".join(parts)
+
+
+class VERAPipeline:
+    """End-to-end per-review inference. Construct once per worker process."""
+
+    def __init__(self, model_dir: Path, hf_repo: str = "", hf_token: str = "") -> None:
+        started = time.perf_counter()
+        paths = ensure_models(model_dir, hf_repo=hf_repo, hf_token=hf_token)  # hard-fail
+
+        self.fusion = FusionModel(paths["fusion_ensemble.json"])
+        self.conformal = ConformalPredictor(paths["conformal_qhat.json"])
+        self.agents = SignalAgents(
+            preprocessor=TextPreprocessor(),
+            experience=ExperienceExtractor(
+                paths["experience_model.onnx"], paths["experience_tokenizer.json"]
+            ),
+            absa=ABSAModule(
+                paths["absa_model.onnx"],
+                paths["absa_tokenizer.json"],
+                paths["absa_config.json"],
+                detector=AspectDetector(paths["aspect_detector.joblib"]),
+            ),
+            consistency=ConsistencyScorer(),
+            quality=QualityAssessor(imputation=self.fusion.imputation_means),
+            domain=DomainRatingModel(paths["domain_tfidf.joblib"]),
+        )
+        logger.info("VERA pipeline loaded in %.1fs", time.perf_counter() - started)
+
+    @classmethod
+    def from_components(
+        cls, agents: SignalAgents, fusion: FusionModel, conformal: ConformalPredictor
+    ) -> VERAPipeline:
+        """Assemble from pre-built parts (tests, notebooks)."""
+        instance = cls.__new__(cls)
+        instance.agents = agents
+        instance.fusion = fusion
+        instance.conformal = conformal
+        return instance
+
+    def process(
+        self,
+        *,
+        text: str,
+        star_rating: float,
+        helpful_votes: int | None = None,
+        verified_purchase: bool | None = None,
+        product_title: str | None = "",
+        timestamp_ms: int | None = None,
+    ) -> dict[str, Any]:
+        """Score one review (PRD §3.1 output contract)."""
+        started = time.perf_counter()
+        outputs = extract_signals(
+            self.agents,
+            text=text,
+            star_rating=star_rating,
+            helpful_votes=helpful_votes,
+            verified_purchase=verified_purchase,
+            timestamp_ms=timestamp_ms,
+            product_title=product_title,
+        )
+        fused = self.fusion.predict(build_feature_vector(outputs))
+        lower, upper, confidence = self.conformal.predict_interval(
+            fused["score"], fused["ensemble_std"]
+        )
+        corrected = round(fused["score"], 3)
+
+        pre, cons, qual = outputs["preprocessor"], outputs["consistency"], outputs["quality"]
+        return {
+            "corrected_score": corrected,
+            "interval": [round(lower, 3), round(upper, 3)],
+            "confidence": round(confidence, 3),
+            "aspects": outputs["absa"]["aspects"],
+            # Agent 4 definition: star − text-derived experience score (positive = inflated).
+            "star_text_gap": cons["star_text_gap"],
+            "review_quality": qual["quality_score"],
+            "ensemble_std": round(fused["ensemble_std"], 4),
+            "flags": {
+                "language": pre["language"],
+                "is_english": pre["is_english"],
+                "is_earbud": pre["is_earbud"],
+                "earbud_confidence": pre["earbud_confidence"],
+                "has_url": pre["has_url"],
+                "star_inflated": cons["star_inflated"],
+                "star_deflated": cons["star_deflated"],
+                "experience_score": round(outputs["experience_extractor"]["predicted_score"], 3),
+                "domain_score": outputs["domain_rating"]["domain_score"],
+                "imputed_fields": qual["imputed_fields"],
+            },
+            "explanation": build_explanation(outputs, corrected),
+            "latency_ms": round((time.perf_counter() - started) * 1000, 1),
+        }

backend/app/models/__init__.py

@@ -0,0 +1,9 @@
+"""SQLAlchemy ORM models. Importing this package registers every table on ``Base``."""
+
+from app.models.calibration import CalibrationCheckpoint
+from app.models.job import Job
+from app.models.label_run import LabelRun
+from app.models.product import Product
+from app.models.review import Review, ReviewStatus
+
+__all__ = ["CalibrationCheckpoint", "Job", "LabelRun", "Product", "Review", "ReviewStatus"]

backend/app/models/calibration.py

@@ -0,0 +1,24 @@
+"""`calibration_checkpoints` table — history of fitted conformal q_hat values (PRD §10.3)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, Float, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class CalibrationCheckpoint(Base):
+    __tablename__ = "calibration_checkpoints"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    method: Mapped[str] = mapped_column(String(20), nullable=False, default="absolute")
+    q_hat: Mapped[float] = mapped_column(Float, nullable=False)
+    empirical_coverage: Mapped[float] = mapped_column(Float, nullable=False)
+    nominal_coverage: Mapped[float] = mapped_column(Float, nullable=False)
+    calibration_set_size: Mapped[int] = mapped_column(Integer, nullable=False)
+    fitted_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )

backend/app/models/job.py

@@ -0,0 +1,49 @@
+"""`jobs` table — transactional outbox for background work.
+
+Why an outbox: Postgres and Redis cannot commit atomically. The API writes the
+review AND its job row in one database transaction, so a job can never be lost
+between "review saved" and "task queued". Redis only carries wake-up messages;
+the database owns attempts, back-off, and completion. Duplicate deliveries are
+harmless because consumers lock the job row and skip finished jobs.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from enum import StrEnum
+from uuid import uuid4
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class JobKind(StrEnum):
+    REVIEW = "review"  # run the ML pipeline for one review
+    AGGREGATE = "aggregate"  # recompute every product aggregate + ranking
+
+
+class Job(Base):
+    __tablename__ = "jobs"
+    __table_args__ = (Index("idx_jobs_dispatch", "finished_at", "available_at"),)
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))
+    kind: Mapped[str] = mapped_column(String(20), nullable=False)
+    review_id: Mapped[int | None] = mapped_column(
+        ForeignKey("reviews.id", ondelete="CASCADE"), unique=True
+    )
+    attempts: Mapped[int] = mapped_column(Integer, default=0, server_default="0", nullable=False)
+    #: Earliest time the job may run (pushed forward by retry back-off).
+    available_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    #: Last time a wake-up was sent to Redis; NULL means "send now".
+    last_dispatched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    def __repr__(self) -> str:
+        return f"<Job {self.id} kind={self.kind} attempts={self.attempts} done={self.finished_at}>"

backend/app/models/label_run.py

@@ -0,0 +1,28 @@
+"""`label_runs` table — audit trail of LLM-generated labels (PRD §7, §15)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import DateTime, Float, Index, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base, JSONType
+
+
+class LabelRun(Base):
+    __tablename__ = "label_runs"
+    __table_args__ = (Index("idx_label_runs_hash", "review_text_hash"),)
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    review_text_hash: Mapped[str] = mapped_column(String(64), nullable=False)  # sha256 hex
+    labeler: Mapped[str] = mapped_column(String(100), nullable=False)
+    experience_score: Mapped[float | None] = mapped_column(Float)
+    aspects: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+    has_irony: Mapped[bool | None] = mapped_column()
+    review_quality: Mapped[str | None] = mapped_column(String(10))
+    raw_response: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )

backend/app/models/product.py

@@ -0,0 +1,44 @@
+"""`products` table — Empirical-Bayes product aggregates (PRD §11, §15)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import DateTime, Float, Index, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.database import Base, JSONType
+
+if TYPE_CHECKING:
+    from app.models.review import Review
+
+
+class Product(Base):
+    __tablename__ = "products"
+    __table_args__ = (Index("idx_products_rank", "rank_position"),)
+
+    asin: Mapped[str] = mapped_column(String(20), primary_key=True)
+    title: Mapped[str | None] = mapped_column(Text)
+
+    corrected_score: Mapped[float | None] = mapped_column(Float)
+    uncertainty: Mapped[float | None] = mapped_column(Float)
+    trust_weighted_score: Mapped[float | None] = mapped_column(Float)
+    star_average: Mapped[float | None] = mapped_column(Float)
+    eb_shrinkage_factor: Mapped[float | None] = mapped_column(Float)
+    review_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, server_default="0"
+    )
+    aspect_summary: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+
+    rank_position: Mapped[int | None] = mapped_column(Integer)
+    previous_rank_position: Mapped[int | None] = mapped_column(Integer)
+    last_aggregated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    reviews: Mapped[list[Review]] = relationship(back_populates="product")
+
+    def __repr__(self) -> str:
+        return f"<Product asin={self.asin!r} score={self.corrected_score} n={self.review_count}>"

backend/app/models/review.py

@@ -0,0 +1,87 @@
+"""`reviews` table — one submitted review and its pipeline output (PRD §15)."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from enum import StrEnum
+from typing import TYPE_CHECKING, Any
+
+from sqlalchemy import (
+    CheckConstraint,
+    DateTime,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.database import Base, JSONType
+
+if TYPE_CHECKING:
+    from app.models.product import Product
+
+
+class ReviewStatus(StrEnum):
+    """Processing lifecycle of a review."""
+
+    PENDING = "pending"
+    PROCESSING = "processing"
+    DONE = "done"
+    FAILED = "failed"
+
+
+class Review(Base):
+    __tablename__ = "reviews"
+    __table_args__ = (
+        CheckConstraint("star_rating >= 1 AND star_rating <= 5", name="ck_reviews_star_range"),
+        CheckConstraint(
+            "status IN ('pending', 'processing', 'done', 'failed')", name="ck_reviews_status"
+        ),
+        Index("idx_reviews_asin", "product_asin"),
+        Index("idx_reviews_status", "status"),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # --- Input ----------------------------------------------------------------
+    text: Mapped[str] = mapped_column(Text, nullable=False)
+    star_rating: Mapped[float] = mapped_column(Float, nullable=False)
+    # NULL means "not supplied" — the pipeline imputes training means (PRD §10.1).
+    helpful_votes: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    verified_purchase: Mapped[bool | None] = mapped_column(nullable=True)
+    product_title: Mapped[str | None] = mapped_column(Text, nullable=True)
+    product_asin: Mapped[str | None] = mapped_column(
+        String(20), ForeignKey("products.asin", ondelete="SET NULL"), nullable=True
+    )
+
+    # --- Pipeline output ------------------------------------------------------
+    corrected_score: Mapped[float | None] = mapped_column(Float)
+    interval_lower: Mapped[float | None] = mapped_column(Float)
+    interval_upper: Mapped[float | None] = mapped_column(Float)
+    confidence_value: Mapped[float | None] = mapped_column(Float)
+    star_text_gap: Mapped[float | None] = mapped_column(Float)
+    review_quality: Mapped[float | None] = mapped_column(Float)
+    aspects: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+    flags: Mapped[dict[str, Any] | None] = mapped_column(JSONType)
+    explanation: Mapped[str | None] = mapped_column(Text)
+    error_message: Mapped[str | None] = mapped_column(Text)
+
+    # --- Bookkeeping ----------------------------------------------------------
+    status: Mapped[str] = mapped_column(
+        String(20), nullable=False, default=ReviewStatus.PENDING.value
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )
+
+    product: Mapped[Product | None] = relationship(back_populates="reviews")
+
+    def __repr__(self) -> str:
+        return f"<Review id={self.id} status={self.status} asin={self.product_asin!r}>"

backend/app/schemas/__init__.py

@@ -0,0 +1 @@
+"""Pydantic request and response schemas for the public API."""

backend/app/schemas/product.py

@@ -0,0 +1,39 @@
+"""Product API contracts."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict
+
+from app.schemas.review import ReviewResponse
+
+
+class ProductResponse(BaseModel):
+    """One row of the product ranking."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    asin: str
+    title: str | None
+    corrected_score: float | None
+    uncertainty: float | None
+    trust_weighted_score: float | None
+    star_average: float | None
+    review_count: int
+    rank_position: int | None
+    previous_rank_position: int | None
+    aspect_summary: dict[str, Any] | None
+
+
+class ProductDetailResponse(ProductResponse):
+    """Product aggregate plus one page of completed reviews."""
+
+    eb_shrinkage_factor: float | None
+    last_aggregated_at: datetime | None
+    reviews: list[ReviewResponse]
+    reviews_total: int
+    reviews_limit: int
+    reviews_offset: int
+    reviews_has_more: bool

backend/app/schemas/review.py

@@ -0,0 +1,95 @@
+"""Review API contracts."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.ml.constants import Sentiment
+
+
+class AspectSentiment(BaseModel):
+    """Inference result for one earbud aspect."""
+
+    detected: bool
+    sentiment: Sentiment
+    confidence: float = Field(ge=0.0, le=1.0)
+    prob_neg: float | None = Field(default=None, ge=0.0, le=1.0)
+    prob_neu: float | None = Field(default=None, ge=0.0, le=1.0)
+    prob_pos: float | None = Field(default=None, ge=0.0, le=1.0)
+
+
+class ReviewCreate(BaseModel):
+    """A review submitted for asynchronous scoring."""
+
+    text: str = Field(min_length=20, max_length=10_000)
+    star_rating: float = Field(ge=1.0, le=5.0)
+    helpful_votes: int | None = Field(default=None, ge=0)
+    verified_purchase: bool | None = None
+    product_title: str | None = Field(default=None, max_length=1_000)
+    product_asin: str | None = Field(default=None, max_length=20)
+
+    @field_validator("text")
+    @classmethod
+    def text_must_contain_non_whitespace(cls, value: str) -> str:
+        value = value.strip()
+        if not value:
+            raise ValueError("text must contain non-whitespace characters")
+        return value
+
+    @field_validator("product_title", "product_asin", mode="before")
+    @classmethod
+    def strip_optional_strings(cls, value: object) -> str | None:
+        if value is None:
+            return None
+        if not isinstance(value, str):
+            raise ValueError("must be a string")
+        return value.strip() or None
+
+
+class ReviewTaskStatus(BaseModel):
+    """Polling response returned after a review is submitted."""
+
+    review_id: int
+    task_id: str | None
+    status: str
+    message: str
+
+
+class ReviewResponse(BaseModel):
+    """Persisted review plus any completed pipeline result."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    text: str
+    star_rating: float
+    helpful_votes: int | None
+    verified_purchase: bool | None
+    product_title: str | None
+    product_asin: str | None
+    corrected_score: float | None
+    interval_lower: float | None
+    interval_upper: float | None
+    confidence_value: float | None
+    star_text_gap: float | None
+    review_quality: float | None
+    aspects: dict[str, AspectSentiment] | None
+    flags: dict[str, Any] | None
+    explanation: str | None
+    error_message: str | None
+    status: str
+    created_at: datetime
+    updated_at: datetime
+
+
+class ReviewLifecycle(StrEnum):
+    """Public lifecycle values; mirrors the persisted ORM enum."""
+
+    PENDING = "pending"
+    PROCESSING = "processing"
+    DONE = "done"
+    FAILED = "failed"

backend/app/worker/__init__.py

@@ -0,0 +1 @@
+"""Celery application and asynchronous VERA tasks."""

backend/app/worker/celery_app.py

@@ -0,0 +1,71 @@
+"""Celery application (PRD §14.4), tuned for the Upstash free tier.
+
+UPSTASH BUDGET: the free tier allows 10,000 Redis commands/day. Celery's default
+0.1 s broker polling would burn ~864K/day, so ``polling_interval`` is 2.0 s.
+Even that is ~43K polls/day for ONE worker — above the free quota. Run a single
+worker, and expect to need a paid Upstash plan (or local Redis) for sustained
+use. Results are ignored (the DB stores outcomes), saving result-backend writes.
+
+Worker start::
+
+    cd backend && celery -A app.worker.celery_app worker --loglevel=info
+
+Startup behaviour:
+* ``worker_init`` loads the ML pipeline and exits the process if any artifact is
+  missing (hard-fail policy, DECISIONS.md) — before consuming any message.
+* ``worker_ready`` starts the outbox dispatcher thread (:mod:`app.worker.outbox`).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from celery import Celery
+from celery.signals import worker_init, worker_ready
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+celery_app = Celery("vera", broker=settings.broker_url, include=["app.worker.tasks"])
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    timezone="UTC",
+    enable_utc=True,
+    task_ignore_result=True,
+    broker_transport_options={"visibility_timeout": 3600, "polling_interval": 2.0},
+    broker_connection_timeout=3,
+    broker_connection_retry_on_startup=True,
+    worker_prefetch_multiplier=1,
+    task_acks_late=True,
+    task_reject_on_worker_lost=True,
+    # Serialize inference so this worker does not duplicate the loaded FP32
+    # models. Deployment memory must be sized and verified separately; the
+    # free-tier 512 MB target is below the documented production requirement.
+    worker_pool="solo",
+    worker_concurrency=1,
+)
+
+
+@worker_init.connect
+def load_models_or_exit(**_: Any) -> None:
+    """Fail fast: a worker without valid artifacts must not consume jobs."""
+    from app.worker.tasks import get_pipeline
+
+    try:
+        get_pipeline()
+    except Exception as exc:
+        logger.critical("VERA model initialisation failed: %s", exc)
+        # SystemExit (a BaseException) escapes Celery's signal error handling.
+        raise SystemExit(1) from exc
+
+
+@worker_ready.connect
+def start_outbox(**_: Any) -> None:
+    """Run the outbox dispatcher alongside the worker (no extra process needed)."""
+    from app.worker.outbox import start_background_dispatcher
+
+    start_background_dispatcher()
+    logger.info("Outbox dispatcher thread started")

backend/app/worker/outbox.py

@@ -0,0 +1,93 @@
+"""Outbox dispatcher: turns unfinished `jobs` rows into Celery wake-ups.
+
+Runs as a daemon thread inside the worker (started on ``worker_ready`` in
+:mod:`app.worker.celery_app`), so a free-tier deployment needs no extra process.
+It can also run standalone::
+
+    cd backend && python -m app.worker.outbox
+
+A job is (re)sent when it is unfinished, due (``available_at <= now``), and was
+never dispatched or was dispatched more than :data:`REDISPATCH_AFTER` ago — this
+recovers messages lost to a Redis outage or a crashed worker. Importing this
+module never loads ML models.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import threading
+from datetime import UTC, datetime, timedelta
+from typing import Final
+
+from sqlalchemy import or_
+
+from app.database import SessionLocal
+from app.models.job import Job
+
+logger = logging.getLogger(__name__)
+
+POLL_INTERVAL_SECONDS: Final = 5.0
+REDISPATCH_AFTER: Final = timedelta(seconds=60)
+RUN_JOB_TASK: Final = "vera.run_job"
+
+
+def send_wakeup(job_id: str) -> None:
+    """Publish one wake-up message. Raises if the broker is unreachable."""
+    from app.worker.celery_app import celery_app  # lazy: avoid import cycle
+
+    celery_app.send_task(RUN_JOB_TASK, args=[job_id], retry=False)
+
+
+def dispatch_pending(limit: int = 100) -> int:
+    """Send wake-ups for due jobs. Returns the number sent."""
+    now = datetime.now(UTC)
+    sent = 0
+    with SessionLocal() as db:
+        jobs = (
+            db.query(Job)
+            .filter(
+                Job.finished_at.is_(None),
+                Job.available_at <= now,
+                or_(
+                    Job.last_dispatched_at.is_(None),
+                    Job.last_dispatched_at < now - REDISPATCH_AFTER,
+                ),
+            )
+            .order_by(Job.available_at, Job.id)
+            .with_for_update(skip_locked=True)
+            .limit(limit)
+            .all()
+        )
+        for job in jobs:
+            send_wakeup(job.id)
+            job.last_dispatched_at = now
+            sent += 1
+        db.commit()
+    return sent
+
+
+def run_forever(stop: threading.Event | None = None) -> None:
+    """Poll until ``stop`` is set. Errors are logged; the loop never dies."""
+    stop = stop or threading.Event()
+    while not stop.is_set():
+        try:
+            count = dispatch_pending()
+            if count:
+                logger.info("Outbox dispatched %d job(s)", count)
+        except Exception:
+            logger.exception("Outbox dispatch failed; will retry")
+        stop.wait(POLL_INTERVAL_SECONDS)
+
+
+def start_background_dispatcher() -> threading.Event:
+    """Start :func:`run_forever` in a daemon thread; set the returned event to stop it."""
+    stop = threading.Event()
+    threading.Thread(target=run_forever, args=(stop,), name="vera-outbox", daemon=True).start()
+    return stop
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+    with contextlib.suppress(KeyboardInterrupt):
+        run_forever()

backend/app/worker/tasks.py

@@ -0,0 +1,238 @@
+"""Celery task: execute one outbox job (review inference or product aggregation).
+
+Delivery model (see :mod:`app.models.job`): Redis messages are only wake-ups
+and may be duplicated or lost; the ``jobs`` row is the source of truth. Every
+run locks its job row, skips finished jobs, and records retries durably with
+exponential back-off. :mod:`app.worker.outbox` re-sends anything unfinished.
+
+ML models load lazily through :func:`get_pipeline`, so importing this module
+(e.g. from tests) never touches model artifacts.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+from functools import lru_cache
+from typing import TYPE_CHECKING, Any, Final
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.config import settings
+from app.database import SessionLocal
+from app.models.job import Job, JobKind
+from app.models.product import Product
+from app.models.review import Review, ReviewStatus
+from app.worker.celery_app import celery_app
+
+if TYPE_CHECKING:
+    from app.ml.pipeline import VERAPipeline
+
+logger = logging.getLogger(__name__)
+
+MAX_REVIEW_ATTEMPTS: Final = 4
+MAX_AGGREGATE_ATTEMPTS: Final = 10
+MAX_BACKOFF_SECONDS: Final = 300
+#: Arbitrary constant key for the Postgres advisory lock around aggregation.
+AGGREGATION_LOCK_KEY: Final = 19842026
+FAILED_REVIEW_MESSAGE: Final = "Review analysis failed. Please try again later."
+
+
+@lru_cache(maxsize=1)
+def get_pipeline() -> VERAPipeline:
+    """Load production artifacts once per worker process. Never falls back."""
+    from app.ml.pipeline import VERAPipeline
+
+    return VERAPipeline(settings.MODEL_DIR, settings.HF_MODEL_REPO, settings.HF_TOKEN)
+
+
+def _as_utc(value: datetime) -> datetime:
+    """Normalise DB datetimes: SQLite returns naive values, Postgres aware ones."""
+    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
+
+
+def backoff_seconds(attempts: int) -> int:
+    """5 s, 10 s, 20 s, … capped at :data:`MAX_BACKOFF_SECONDS`."""
+    return min(MAX_BACKOFF_SECONDS, 5 * 2 ** min(attempts, 6))
+
+
+# ----------------------------------------------------------------------------
+# Job handlers — run inside the caller's transaction; the caller commits.
+# ----------------------------------------------------------------------------
+def process_review(db: Session, review_id: int) -> None:
+    """Run the pipeline for one review and store its result.
+
+    The review row stays locked for the whole inference, so a crash rolls back
+    cleanly and a duplicate delivery cannot write concurrently. The status is not
+    committed as "processing" separately; clients see "pending" until "done".
+    """
+    review = db.query(Review).filter(Review.id == review_id).with_for_update().one_or_none()
+    if review is None or review.status in (ReviewStatus.DONE.value, ReviewStatus.FAILED.value):
+        return
+
+    result = get_pipeline().process(
+        text=review.text,
+        star_rating=review.star_rating,
+        helpful_votes=review.helpful_votes,
+        verified_purchase=review.verified_purchase,
+        product_title=review.product_title,
+    )
+    review.corrected_score = result["corrected_score"]
+    review.interval_lower, review.interval_upper = result["interval"]
+    review.confidence_value = result["confidence"]
+    review.star_text_gap = result["star_text_gap"]
+    review.review_quality = result["review_quality"]
+    review.aspects = result["aspects"]
+    review.flags = result["flags"]
+    review.explanation = result["explanation"]
+    review.error_message = None
+    review.status = ReviewStatus.DONE.value
+
+    # Reviews without an ASIN never create aggregation work (PRD weakness #3).
+    if review.product_asin:
+        db.add(Job(kind=JobKind.AGGREGATE.value))
+
+
+def aggregate_products(db: Session, current_job_id: str | None = None) -> int:
+    """Recompute EB aggregates and ranks for ALL products. Returns product count.
+
+    The EB prior is a population quantity, so every product is refreshed under
+    one lock (Postgres advisory transaction lock — released on commit/rollback,
+    no TTL to expire mid-run). Other unfinished aggregate jobs visible *before*
+    the review snapshot is read are marked finished: their reviews are already
+    committed and therefore included, so re-running them would be pure waste.
+    """
+    from app.ml.calibration.empirical_bayes import (
+        EmpiricalBayesAggregator,
+        ReviewSignal,
+        rank_products,
+    )
+
+    if db.get_bind().dialect.name == "postgresql":
+        db.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": AGGREGATION_LOCK_KEY})
+
+    superseded = (
+        db.query(Job)
+        .filter(
+            Job.kind == JobKind.AGGREGATE.value,
+            Job.finished_at.is_(None),
+            Job.id != (current_job_id or ""),
+        )
+        .with_for_update(skip_locked=True)
+        .all()
+    )
+
+    rows = (
+        db.query(
+            Review.product_asin,
+            Review.corrected_score,
+            Review.review_quality,
+            Review.star_rating,
+            Review.aspects,
+        )
+        .filter(
+            Review.status == ReviewStatus.DONE.value,
+            Review.product_asin.is_not(None),
+            Review.corrected_score.is_not(None),
+        )
+        .all()
+    )
+    now = datetime.now(UTC)
+    for job in superseded:
+        job.finished_at = now
+
+    signals = [
+        ReviewSignal(
+            product_asin=row.product_asin,
+            corrected_score=float(row.corrected_score),
+            quality_weight=float(row.review_quality) if row.review_quality is not None else 0.5,
+            star_rating=float(row.star_rating),
+            aspects=row.aspects or {},
+        )
+        for row in rows
+    ]
+    if not signals:
+        return 0
+
+    _, aggregates = EmpiricalBayesAggregator().aggregate_all(signals)
+    for rank, asin in enumerate(rank_products(aggregates), start=1):
+        aggregate = aggregates[asin]
+        product = db.get(Product, asin)
+        if product is None:  # FK makes this impossible unless data was edited by hand
+            raise RuntimeError(f"Completed review references missing product {asin!r}")
+        product.previous_rank_position = product.rank_position
+        product.rank_position = rank
+        product.corrected_score = aggregate.corrected_score
+        product.uncertainty = aggregate.uncertainty
+        product.trust_weighted_score = aggregate.trust_weighted_score
+        product.star_average = aggregate.star_average
+        product.eb_shrinkage_factor = aggregate.eb_shrinkage_factor
+        product.review_count = aggregate.review_count
+        product.aspect_summary = aggregate.aspect_summary
+        product.last_aggregated_at = now
+    return len(aggregates)
+
+
+# ----------------------------------------------------------------------------
+# Celery entry point
+# ----------------------------------------------------------------------------
+def _record_failure(db: Session, job_id: str) -> None:
+    """Increment attempts; schedule a retry or mark the job (and review) failed."""
+    job = db.query(Job).filter(Job.id == job_id).with_for_update().one_or_none()
+    if job is None or job.finished_at is not None:
+        return  # a duplicate delivery finished it meanwhile
+
+    job.attempts += 1
+    limit = MAX_REVIEW_ATTEMPTS if job.kind == JobKind.REVIEW.value else MAX_AGGREGATE_ATTEMPTS
+    now = datetime.now(UTC)
+    if job.attempts >= limit:
+        job.finished_at = now
+        if job.kind == JobKind.REVIEW.value and job.review_id is not None:
+            review = db.get(Review, job.review_id)
+            if review is not None and review.status != ReviewStatus.DONE.value:
+                review.status = ReviewStatus.FAILED.value
+                review.error_message = FAILED_REVIEW_MESSAGE
+        logger.error("Job %s (%s) failed permanently after %d attempts", job.id, job.kind, limit)
+    else:
+        job.available_at = now + timedelta(seconds=backoff_seconds(job.attempts))
+        job.last_dispatched_at = None  # let the outbox re-send once due
+    db.commit()
+
+
+def execute_job(job_id: str) -> str:
+    """Run one job to completion or record its failure. Returns an outcome label."""
+    with SessionLocal() as db:
+        try:
+            job = (
+                db.query(Job)
+                .filter(Job.id == job_id)
+                .with_for_update(skip_locked=True)
+                .one_or_none()
+            )
+            if job is None or job.finished_at is not None:
+                return "ignored"  # unknown, finished, or locked by a concurrent delivery
+            if _as_utc(job.available_at) > datetime.now(UTC):
+                return "deferred"
+
+            if job.kind == JobKind.REVIEW.value and job.review_id is not None:
+                process_review(db, job.review_id)
+            elif job.kind == JobKind.AGGREGATE.value:
+                aggregate_products(db, current_job_id=job.id)
+            else:
+                raise ValueError(f"Unknown job kind {job.kind!r}")
+
+            job.finished_at = datetime.now(UTC)
+            db.commit()
+            return "done"
+        except Exception:
+            db.rollback()
+            logger.exception("Job %s failed; recording durable retry", job_id)
+            _record_failure(db, job_id)
+            return "retry_or_failed"
+
+
+@celery_app.task(name="vera.run_job", ignore_result=True)
+def run_job(job_id: str) -> dict[str, Any]:
+    """Celery wrapper around :func:`execute_job`."""
+    return {"job_id": job_id, "status": execute_job(job_id)}

backend/migrations/env.py

@@ -0,0 +1,52 @@
+"""Alembic environment configured from VERA's runtime settings."""
+
+from __future__ import annotations
+
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import engine_from_config, pool
+
+import app.models  # noqa: F401  # Register all ORM metadata before autogeneration.
+from app.config import settings
+from app.database import Base
+
+config = context.config
+# ConfigParser treats "%" as interpolation; URL-encoded passwords contain it.
+config.set_main_option("sqlalchemy.url", settings.DATABASE_URL.replace("%", "%%"))
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def run_migrations_offline() -> None:
+    """Run migrations without an Engine, e.g. to generate SQL."""
+    context.configure(
+        url=settings.DATABASE_URL,
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+        compare_type=True,
+    )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online() -> None:
+    """Run migrations against the configured database."""
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section, {}),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+    with connectable.connect() as connection:
+        context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

backend/migrations/script.py.mako

@@ -0,0 +1,24 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = ${repr(branch_labels)}
+depends_on = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}