initial deployment for HF Spaces

.dockerignore

@@ -0,0 +1,54 @@
+# Environments
+venv/
+.venv/
+env/
+.env
+
+# Python
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Data & Models (not baked into image — loaded at runtime)
+model_cache/
+tests/
+.ipynb_checkpoints/
+
+# Version Control
+.git/
+.github/
+.gitignore
+
+# IDEs
+.vscode/
+.idea/
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# HF Spaces — README is required at repo root, not in image
+# README.md is NOT excluded so HF can read the frontmatter
+
+
+model_cache/
+__pycache__/
+*.pyc
+.env

.env

@@ -0,0 +1,47 @@
+# Server
+HOST=0.0.0.0
+PORT=8000
+ENV=development
+LOG_LEVEL=INFO
+
+# JWT Authentication
+JWT_ACCESS_SECRET=hubble-access-secret-change-in-production-32chars
+JWT_REFRESH_SECRET=hubble-refresh-secret-change-in-production-32chars
+JWT_ACCESS_EXPIRES_MINUTES=15
+JWT_REFRESH_EXPIRES_DAYS=7
+
+# Security
+BCRYPT_ROUNDS=12
+CORS_ORIGINS=http://localhost:3000,http://localhost:3001
+
+# MongoDB (Atlas Cloud)
+MONGODB_URI=mongodb+srv://sajithjaganathan7_db_user:Winter_bear_07@cluster0.jxdvukx.mongodb.net/hubble?appName=Cluster0
+MONGODB_DB_NAME=hubble
+
+# Redis (Redis Cloud)
+REDIS_URL=redis://default:dongH74t41QfBN0TO0e5ylWAVThXZoLR@redis-13470.crce281.ap-south-1-3.ec2.cloud.redislabs.com:13470
+REDIS_CACHE_TTL=300
+
+# Gemini API Keys (User should provide real keys for this)
+GEMINI_API_KEYS=AIzaSyBL6onsP6Z-wG32nFgy8Bi7uGDGIopbRDE
+GEMINI_MODEL=gemini-2.5-flash
+
+# LangSmith Tracing
+LANGSMITH_API_KEY=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
+LANGSMITH_PROJECT=hubble-moderation
+LANGSMITH_TRACING_V2=true
+
+# Model Configuration
+MODEL_CACHE_DIR=./model_cache
+ONNX_ENABLED=false
+TEXT_MODEL_NAME=unitary/toxic-bert
+IMAGE_MODEL_NAME=google/efficientnet-b0
+CLIP_MODEL_NAME=openai/clip-vit-base-patch32
+
+# Risk Thresholds
+RISK_LOW_MAX=30
+RISK_MEDIUM_MAX=65
+
+# Video Processing
+VIDEO_MAX_FRAMES=10
+VIDEO_FPS_SAMPLE=1

.env.example

@@ -0,0 +1,48 @@
+# ============================================
+# Hubble AI Engine - Environment Configuration
+# ============================================
+# On Hugging Face Spaces, set these as Secrets
+# in the Space settings (Settings > Secrets).
+# They are injected as environment variables at runtime.
+
+# Server
+HOST=0.0.0.0
+PORT=7860
+ENV=production
+LOG_LEVEL=INFO
+
+# MongoDB (async via motor)
+MONGODB_URI=mongodb+srv://<user>:<password>@<cluster>.mongodb.net/hubble
+MONGODB_DB_NAME=hubble
+
+# Redis
+REDIS_URL=redis://default:<password>@<host>:<port>
+REDIS_CACHE_TTL=300
+
+# Gemini API Keys (comma-separated for rotation)
+GEMINI_API_KEYS=your-key-1,your-key-2,your-key-3
+GEMINI_MODEL=gemini-2.5-flash
+
+# LangSmith Observability (optional)
+LANGSMITH_API_KEY=your-langsmith-api-key
+LANGSMITH_PROJECT=hubble-moderation
+LANGSMITH_TRACING_V2=true
+
+# JWT Secrets (use long random strings)
+JWT_ACCESS_SECRET=your-access-secret-at-least-32-chars
+JWT_REFRESH_SECRET=your-refresh-secret-at-least-32-chars
+
+# Model Configuration
+MODEL_CACHE_DIR=/tmp/model_cache
+ONNX_ENABLED=false
+TEXT_MODEL_NAME=unitary/toxic-bert
+IMAGE_MODEL_NAME=google/efficientnet-b0
+CLIP_MODEL_NAME=openai/clip-vit-base-patch32
+
+# Risk Thresholds
+RISK_LOW_MAX=30
+RISK_MEDIUM_MAX=65
+
+# Video Processing
+VIDEO_MAX_FRAMES=10
+VIDEO_FPS_SAMPLE=1

.gitattributes

@@ -0,0 +1,36 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.onnx.data filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,34 @@
+# Hugging Face Spaces — Hubble AI Engine
+FROM python:3.12-slim
+
+# HF Spaces requires port 7860
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV HOST=0.0.0.0
+ENV PORT=7860
+
+WORKDIR /app
+
+# System dependencies for OpenCV and native builds
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libgl1 \
+    libglib2.0-0 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir torch torchvision --index-url https://download.pytorch.org/whl/cpu && \
+    pip install --no-cache-dir -r requirements.txt --extra-index-url https://download.pytorch.org/whl/cpu
+
+COPY . .
+
+# HF Spaces runs containers as a non-root user (UID 1000)
+RUN useradd -m -u 1000 user
+RUN mkdir -p /tmp/model_cache && chown -R user:user /tmp/model_cache && chown -R user:user /app
+USER user
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,11 @@
+---
+title: SentinelAI
+emoji: 📈
+colorFrom: green
+colorTo: blue
+sdk: docker
+pinned: false
+license: mit
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app/__init__.py

@@ -0,0 +1,2 @@
+# app/__init__.py
+"""Hubble AI Engine — Production-grade cyberbullying detection pipeline."""

app/api/__init__.py

@@ -0,0 +1,2 @@
+# app/api/__init__.py
+"""API layer: routes, schemas, and versioning."""

app/api/router.py

@@ -0,0 +1,27 @@
+# app/api/router.py
+# Root router — aggregates all versioned API routes
+
+from fastapi import APIRouter
+from app.api.v1.health import router as health_router
+from app.api.v1.analyze import router as analyze_router
+from app.api.v1.history import router as history_router
+from app.api.v1.auth import router as auth_router
+from app.api.v1.users import router as users_router
+from app.api.v1.scan import router as scan_router
+from app.api.v1.alerts import router as alerts_router
+
+# Main API router
+api_router = APIRouter()
+
+# ── Unauthenticated: health + raw AI pipeline ──
+api_router.include_router(health_router, prefix="")
+api_router.include_router(analyze_router, prefix="/api/v1")
+api_router.include_router(history_router, prefix="/api/v1")
+
+# ── Auth ──
+api_router.include_router(auth_router, prefix="/api/v1")
+
+# ── Authenticated business logic ──
+api_router.include_router(users_router, prefix="/api/v1")
+api_router.include_router(scan_router, prefix="/api/v1")
+api_router.include_router(alerts_router, prefix="/api/v1")

app/api/schemas/__init__.py

@@ -0,0 +1,2 @@
+# app/api/schemas/__init__.py
+"""Pydantic request/response schemas."""

app/api/schemas/requests.py

@@ -0,0 +1,43 @@
+# app/api/schemas/requests.py
+# Pydantic request models for API endpoints
+
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class TextAnalysisRequest(BaseModel):
+    """Request body for text analysis."""
+    text: str = Field(..., min_length=1, max_length=10000, description="Text content to analyze")
+    user_id: Optional[str] = Field(None, description="Optional user ID for history tracking")
+    source_app: Optional[str] = Field(None, description="Source application (e.g., 'instagram', 'whatsapp')")
+    metadata: Optional[dict] = Field(None, description="Additional metadata")
+
+    model_config = {
+        "json_schema_extra": {
+            "examples": [
+                {
+                    "text": "You are so ugly nobody likes you",
+                    "user_id": "user_123",
+                    "source_app": "instagram",
+                }
+            ]
+        }
+    }
+
+
+class ImageAnalysisRequest(BaseModel):
+    """Metadata for image analysis (file sent as multipart)."""
+    user_id: Optional[str] = Field(None, description="Optional user ID for history tracking")
+    source_app: Optional[str] = Field(None, description="Source application")
+
+
+class VideoAnalysisRequest(BaseModel):
+    """Metadata for video analysis (file sent as multipart)."""
+    user_id: Optional[str] = Field(None, description="Optional user ID for history tracking")
+    source_app: Optional[str] = Field(None, description="Source application")
+
+
+class HistoryRequest(BaseModel):
+    """Parameters for history queries."""
+    limit: int = Field(20, ge=1, le=100, description="Maximum results to return")
+    skip: int = Field(0, ge=0, description="Number of results to skip")

app/api/schemas/responses.py

@@ -0,0 +1,136 @@
+# app/api/schemas/responses.py
+# Pydantic response models for API endpoints
+
+from pydantic import BaseModel, Field
+from typing import Optional, Literal, Any
+from datetime import datetime
+
+
+class DecisionDetail(BaseModel):
+    """Details of the moderation decision."""
+    action: Literal["ALLOWED", "WARNING", "BLOCKED"]
+    reason: str
+    severity: str
+    should_alert_parent: bool = False
+    escalation_notes: Optional[str] = None
+
+
+class DeepAnalysisDetail(BaseModel):
+    """Details from the deep analysis stage (only present for HIGH risk)."""
+    is_confirmed: bool
+    severity: str
+    reasoning: str
+    categories: list[str] = []
+    recommended_action: str
+    confidence: float
+    clip_scores: dict = {}
+
+
+class RiskDetail(BaseModel):
+    """Breakdown of risk scoring."""
+    score: float
+    level: Literal["LOW", "MEDIUM", "HIGH"]
+    components: dict = {}
+    repeat_offender: bool = False
+
+
+class FilterDetail(BaseModel):
+    """Fast filter stage output."""
+    is_flagged: bool
+    scores: dict[str, float] = {}
+    max_score: float
+    max_label: str
+    categories: list[str] = []
+
+
+class AnalysisResponse(BaseModel):
+    """
+    Unified response for all /analyze/* endpoints.
+
+    This is the primary contract between the AI engine
+    and the Node.js backend (and any external consumers).
+    """
+    request_id: str = Field(..., description="Unique request identifier")
+    input_type: Literal["text", "image", "video"]
+    status: Literal["ALLOWED", "WARNING", "BLOCKED"]
+    risk_level: Literal["LOW", "MEDIUM", "HIGH"]
+    risk_score: float = Field(..., ge=0, le=100, description="Composite risk score 0-100")
+    categories: list[str] = Field(default_factory=list, description="Detected abuse categories")
+    confidence: float = Field(..., ge=0, le=1, description="Overall confidence 0-1")
+
+    # Detailed breakdowns
+    decision: DecisionDetail
+    risk_detail: RiskDetail
+    filter_detail: FilterDetail
+    deep_analysis: Optional[DeepAnalysisDetail] = None
+
+    # Metadata
+    processing_time_ms: int = Field(..., description="Total processing time in milliseconds")
+    trace_id: Optional[str] = Field(None, description="LangSmith trace ID for observability")
+    cached: bool = Field(False, description="Whether this result was served from cache")
+
+    model_config = {
+        "json_schema_extra": {
+            "examples": [
+                {
+                    "request_id": "req_abc123",
+                    "input_type": "text",
+                    "status": "WARNING",
+                    "risk_level": "MEDIUM",
+                    "risk_score": 45.2,
+                    "categories": ["insult", "toxic"],
+                    "confidence": 0.82,
+                    "decision": {
+                        "action": "WARNING",
+                        "reason": "Content flagged as potentially harmful",
+                        "severity": "medium",
+                        "should_alert_parent": False,
+                    },
+                    "risk_detail": {
+                        "score": 45.2,
+                        "level": "MEDIUM",
+                        "components": {
+                            "base_score": 42.0,
+                            "multi_category_penalty": 3.2,
+                            "repeat_offender_boost": 0.0,
+                        },
+                        "repeat_offender": False,
+                    },
+                    "filter_detail": {
+                        "is_flagged": True,
+                        "scores": {"toxic": 0.78, "insult": 0.65},
+                        "max_score": 0.78,
+                        "max_label": "toxic",
+                        "categories": ["toxic", "insult"],
+                    },
+                    "deep_analysis": None,
+                    "processing_time_ms": 156,
+                    "trace_id": None,
+                    "cached": False,
+                }
+            ]
+        }
+    }
+
+
+class HealthResponse(BaseModel):
+    """Health check response."""
+    status: str
+    version: str
+    models: dict[str, bool]
+    services: dict[str, bool]
+    uptime_seconds: float
+
+
+class HistoryResponse(BaseModel):
+    """Moderation history response."""
+    user_id: str
+    total: int
+    results: list[dict[str, Any]]
+
+
+class ErrorResponse(BaseModel):
+    """Standard error response."""
+    error: str
+    detail: str
+    request_id: Optional[str] = None

app/api/v1/__init__.py

@@ -0,0 +1,2 @@
+# app/api/v1/__init__.py
+"""API v1 endpoints."""

app/api/v1/alerts.py

@@ -0,0 +1,151 @@
+# app/api/v1/alerts.py
+# Alert endpoints for parents and children
+
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional
+from app.core.dependencies import get_current_user
+from app.db.models.user import UserDocument, UserRole
+from app.db.models.alert import AlertDocument, AlertStatus
+from app.observability.logging import get_logger
+from datetime import datetime
+
+logger = get_logger(__name__)
+router = APIRouter(prefix="/alerts", tags=["Alerts"])
+
+
+@router.get("")
+async def list_alerts(
+    page: int = 1,
+    limit: int = 20,
+    status: Optional[str] = None,
+    severity: Optional[str] = None,
+    user: UserDocument = Depends(get_current_user),
+):
+    """List alerts. Parents see all their children's alerts; children see their own."""
+    skip = (page - 1) * limit
+
+    if user.role == UserRole.PARENT:
+        query = AlertDocument.find(AlertDocument.parent_id == str(user.id))
+    else:
+        query = AlertDocument.find(AlertDocument.child_id == str(user.id))
+
+    if status:
+        query = query.find(AlertDocument.status == AlertStatus(status))
+    if severity:
+        from app.db.models.alert import AlertSeverity
+        query = query.find(AlertDocument.severity == AlertSeverity(severity))
+
+    alerts = await query.sort(-AlertDocument.created_at).skip(skip).limit(limit).to_list()
+    return {
+        "success": True,
+        "page": page,
+        "limit": limit,
+        "alerts": [_fmt(a) for a in alerts],
+    }
+
+
+@router.get("/{alert_id}")
+async def get_alert(
+    alert_id: str,
+    user: UserDocument = Depends(get_current_user),
+):
+    alert = await AlertDocument.get(alert_id)
+    if not alert:
+        raise HTTPException(status_code=404, detail="Alert not found")
+    _check_access(alert, user)
+    return {"success": True, "alert": _fmt(alert)}
+
+
+class AcknowledgeRequest(BaseModel):
+    resolution_notes: Optional[str] = None
+
+
+@router.post("/{alert_id}/acknowledge")
+async def acknowledge_alert(
+    alert_id: str,
+    user: UserDocument = Depends(get_current_user),
+):
+    alert = await AlertDocument.get(alert_id)
+    if not alert:
+        raise HTTPException(status_code=404, detail="Alert not found")
+    _check_access(alert, user)
+    alert.status = AlertStatus.ACKNOWLEDGED
+    alert.acknowledged_at = datetime.utcnow()
+    alert.updated_at = datetime.utcnow()
+    await alert.save()
+    return {"success": True, "alert": _fmt(alert)}
+
+
+@router.post("/{alert_id}/resolve")
+async def resolve_alert(
+    alert_id: str,
+    body: AcknowledgeRequest,
+    user: UserDocument = Depends(get_current_user),
+):
+    if user.role != UserRole.PARENT:
+        raise HTTPException(status_code=403, detail="Only parents can resolve alerts")
+    alert = await AlertDocument.get(alert_id)
+    if not alert:
+        raise HTTPException(status_code=404, detail="Alert not found")
+    if alert.parent_id != str(user.id):
+        raise HTTPException(status_code=403, detail="Cannot resolve this alert")
+    alert.status = AlertStatus.RESOLVED
+    alert.resolved_at = datetime.utcnow()
+    alert.resolved_by = str(user.id)
+    alert.resolution_notes = body.resolution_notes
+    alert.updated_at = datetime.utcnow()
+    await alert.save()
+    return {"success": True, "alert": _fmt(alert)}
+
+
+@router.get("/stats/summary")
+async def alert_stats(user: UserDocument = Depends(get_current_user)):
+    if user.role != UserRole.PARENT:
+        raise HTTPException(status_code=403, detail="Only parents can view stats")
+    alerts = await AlertDocument.find(
+        AlertDocument.parent_id == str(user.id)
+    ).to_list()
+
+    by_severity: dict = {}
+    by_status: dict = {}
+    by_category: dict = {}
+    for a in alerts:
+        by_severity[a.severity.value] = by_severity.get(a.severity.value, 0) + 1
+        by_status[a.status.value] = by_status.get(a.status.value, 0) + 1
+        for c in a.categories:
+            by_category[c] = by_category.get(c, 0) + 1
+
+    return {
+        "success": True,
+        "total": len(alerts),
+        "by_severity": by_severity,
+        "by_status": by_status,
+        "by_category": by_category,
+    }
+
+
+# ──────────────────────────────────────────────
+# Helpers
+# ──────────────────────────────────────────────
+
+def _check_access(alert: AlertDocument, user: UserDocument):
+    if alert.parent_id != str(user.id) and alert.child_id != str(user.id):
+        raise HTTPException(status_code=403, detail="Cannot access this alert")
+
+
+def _fmt(a: AlertDocument) -> dict:
+    return {
+        "id": str(a.id),
+        "child_id": a.child_id,
+        "parent_id": a.parent_id,
+        "title": a.title,
+        "message": a.message,
+        "guidance": a.guidance,
+        "severity": a.severity.value,
+        "categories": a.categories,
+        "status": a.status.value,
+        "created_at": a.created_at.isoformat(),
+        "acknowledged_at": a.acknowledged_at.isoformat() if a.acknowledged_at else None,
+        "resolved_at": a.resolved_at.isoformat() if a.resolved_at else None,
+    }

app/api/v1/analyze.py

@@ -0,0 +1,330 @@
+# app/api/v1/analyze.py
+# Core analysis endpoints — text, image, video
+
+import time
+import uuid
+from dataclasses import asdict
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException
+
+from app.api.schemas.requests import TextAnalysisRequest
+from app.api.schemas.responses import (
+    AnalysisResponse,
+    DecisionDetail,
+    RiskDetail,
+    FilterDetail,
+    DeepAnalysisDetail,
+    ErrorResponse,
+)
+from app.services.mongo_service import mongo_service
+from app.services.redis_service import redis_service
+from app.pipeline.workflow import get_workflow, PipelineState
+from app.observability.logging import get_logger
+from app.config import get_settings
+
+logger = get_logger(__name__)
+settings = get_settings()
+
+router = APIRouter(tags=["Analysis"])
+
+
+# ──────────────────────────────────────────────
+# Helper: Convert pipeline state to API response
+# ──────────────────────────────────────────────
+
+def _build_response(
+    state: dict,
+    input_type: str,
+    request_id: str,
+    start_time: float,
+    cached: bool = False,
+) -> AnalysisResponse:
+    """Convert pipeline output state into the unified AnalysisResponse."""
+
+    risk = state.get("risk_score")
+    decision = state.get("decision")
+    filter_result = state.get("filter_result")
+    deep_result = state.get("deep_result")
+
+    # Build nested models
+    decision_detail = DecisionDetail(
+        action=decision.action if decision else "WARNING",
+        reason=decision.reason if decision else "Pipeline incomplete",
+        severity=decision.severity if decision else "medium",
+        should_alert_parent=decision.should_alert_parent if decision else False,
+        escalation_notes=decision.escalation_notes if decision else None,
+    )
+
+    risk_detail = RiskDetail(
+        score=risk.score if risk else 0.0,
+        level=risk.level if risk else "LOW",
+        components=risk.components if risk else {},
+        repeat_offender=risk.repeat_offender if risk else False,
+    )
+
+    filter_detail = FilterDetail(
+        is_flagged=filter_result.is_flagged if filter_result else False,
+        scores=filter_result.scores if filter_result else {},
+        max_score=filter_result.max_score if filter_result else 0.0,
+        max_label=filter_result.max_label if filter_result else "",
+        categories=filter_result.categories if filter_result else [],
+    )
+
+    deep_analysis = None
+    if deep_result:
+        deep_analysis = DeepAnalysisDetail(
+            is_confirmed=deep_result.is_confirmed,
+            severity=deep_result.severity,
+            reasoning=deep_result.reasoning,
+            categories=deep_result.categories,
+            recommended_action=deep_result.recommended_action,
+            confidence=deep_result.confidence,
+            clip_scores=deep_result.clip_scores,
+        )
+
+    processing_time = int((time.time() - start_time) * 1000)
+
+    return AnalysisResponse(
+        request_id=request_id,
+        input_type=input_type,
+        status=decision_detail.action,
+        risk_level=risk_detail.level,
+        risk_score=risk_detail.score,
+        categories=filter_detail.categories,
+        confidence=filter_detail.max_score,
+        decision=decision_detail,
+        risk_detail=risk_detail,
+        filter_detail=filter_detail,
+        deep_analysis=deep_analysis,
+        processing_time_ms=processing_time,
+        trace_id=None,  # TODO: capture from LangSmith
+        cached=cached,
+    )
+
+
+# ──────────────────────────────────────────────
+# POST /analyze/text
+# ──────────────────────────────────────────────
+
+@router.post(
+    "/analyze/text",
+    response_model=AnalysisResponse,
+    responses={400: {"model": ErrorResponse}, 503: {"model": ErrorResponse}},
+    summary="Analyze text content for cyberbullying",
+)
+async def analyze_text(request: TextAnalysisRequest):
+    """
+    Full moderation pipeline for text content.
+
+    Pipeline: Preprocess → RoBERTa Filter → Risk Score → [Deep Analysis] → Decision
+    """
+    request_id = f"req_{uuid.uuid4().hex[:12]}"
+    start_time = time.time()
+
+    logger.info("analyze_text_started", request_id=request_id, text_length=len(request.text))
+
+    # Validate text length
+    if len(request.text) > settings.max_text_length:
+        raise HTTPException(status_code=400, detail=f"Text too long (max {settings.max_text_length} chars)")
+
+    # Check cache
+    cached_result = await redis_service.get_cached_result(request.text, "text")
+    if cached_result:
+        logger.info("analyze_text_cached", request_id=request_id)
+        cached_result["request_id"] = request_id
+        cached_result["cached"] = True
+        cached_result["processing_time_ms"] = int((time.time() - start_time) * 1000)
+        return AnalysisResponse(**cached_result)
+
+    # Run pipeline
+    try:
+        workflow = get_workflow()
+        initial_state: PipelineState = {
+            "input_type": "text",
+            "raw_content": request.text,
+            "user_id": request.user_id,
+        }
+
+        result_state = await workflow.ainvoke(initial_state)
+
+        # Check for pipeline errors
+        if result_state.get("error"):
+            raise HTTPException(status_code=500, detail=result_state["error"])
+
+        response = _build_response(result_state, "text", request_id, start_time)
+
+        # Cache the result
+        await redis_service.cache_result(
+            request.text, "text", response.model_dump()
+        )
+
+        # Log to MongoDB
+        await _log_moderation(request_id, "text", request.user_id, response)
+
+        return response
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("analyze_text_failed", request_id=request_id, error=str(e))
+        raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
+
+
+# ──────────────────────────────────────────────
+# POST /analyze/image
+# ──────────────────────────────────────────────
+
+@router.post(
+    "/analyze/image",
+    response_model=AnalysisResponse,
+    responses={400: {"model": ErrorResponse}, 503: {"model": ErrorResponse}},
+    summary="Analyze image content for harmful material",
+)
+async def analyze_image(
+    file: UploadFile = File(..., description="Image file to analyze"),
+    user_id: str | None = Form(None, description="Optional user ID"),
+    source_app: str | None = Form(None, description="Source application"),
+):
+    """
+    Full moderation pipeline for image content.
+
+    Pipeline: Preprocess → EfficientNet Filter → Risk Score → [CLIP + Gemini] → Decision
+    """
+    request_id = f"req_{uuid.uuid4().hex[:12]}"
+    start_time = time.time()
+
+    # Validate file type
+    if not file.content_type or not file.content_type.startswith("image/"):
+        raise HTTPException(status_code=400, detail="File must be an image (JPEG, PNG, etc.)")
+
+    logger.info("analyze_image_started", request_id=request_id, filename=file.filename)
+
+    # Validate file size
+    if file.size and file.size > settings.max_image_size:
+        raise HTTPException(status_code=400, detail=f"Image too large (max {settings.max_image_size / 1024 / 1024}MB)")
+
+    try:
+        image_bytes = await file.read()
+
+        workflow = get_workflow()
+        initial_state: PipelineState = {
+            "input_type": "image",
+            "raw_content": image_bytes,
+            "user_id": user_id,
+        }
+
+        result_state = await workflow.ainvoke(initial_state)
+
+        if result_state.get("error"):
+            raise HTTPException(status_code=500, detail=result_state["error"])
+
+        response = _build_response(result_state, "image", request_id, start_time)
+
+        # Log to MongoDB
+        await _log_moderation(request_id, "image", user_id, response)
+
+        return response
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("analyze_image_failed", request_id=request_id, error=str(e))
+        raise HTTPException(status_code=500, detail=f"Image analysis failed: {str(e)}")
+
+
+# ──────────────────────────────────────────────
+# POST /analyze/video
+# ──────────────────────────────────────────────
+
+@router.post(
+    "/analyze/video",
+    response_model=AnalysisResponse,
+    responses={400: {"model": ErrorResponse}, 503: {"model": ErrorResponse}},
+    summary="Analyze video content for harmful material",
+)
+async def analyze_video(
+    file: UploadFile = File(..., description="Video file to analyze"),
+    user_id: str | None = Form(None, description="Optional user ID"),
+    source_app: str | None = Form(None, description="Source application"),
+):
+    """
+    Full moderation pipeline for video content.
+
+    Pipeline: Extract frames → Per-frame EfficientNet → Aggregate risk → [Deep Analysis] → Decision
+    """
+    request_id = f"req_{uuid.uuid4().hex[:12]}"
+    start_time = time.time()
+
+    # Validate file type
+    if not file.content_type or not file.content_type.startswith("video/"):
+        raise HTTPException(status_code=400, detail="File must be a video (MP4, AVI, etc.)")
+
+    logger.info("analyze_video_started", request_id=request_id, filename=file.filename)
+
+    # Validate file size
+    if file.size and file.size > settings.max_video_size:
+        raise HTTPException(status_code=400, detail=f"Video too large (max {settings.max_video_size / 1024 / 1024}MB)")
+
+    try:
+        video_bytes = await file.read()
+
+        workflow = get_workflow()
+        initial_state: PipelineState = {
+            "input_type": "video",
+            "raw_content": video_bytes,
+            "user_id": user_id,
+        }
+
+        result_state = await workflow.ainvoke(initial_state)
+
+        if result_state.get("error"):
+            raise HTTPException(status_code=500, detail=result_state["error"])
+
+        response = _build_response(result_state, "video", request_id, start_time)
+
+        # Log to MongoDB
+        await _log_moderation(request_id, "video", user_id, response)
+
+        return response
+
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("analyze_video_failed", request_id=request_id, error=str(e))
+        raise HTTPException(status_code=500, detail=f"Video analysis failed: {str(e)}")
+
+
+# ──────────────────────────────────────────────
+# Helper: Log moderation result to MongoDB
+# ──────────────────────────────────────────────
+
+async def _log_moderation(
+    request_id: str,
+    input_type: str,
+    user_id: str | None,
+    response: AnalysisResponse,
+) -> None:
+    """Log the moderation result and update user history."""
+    try:
+        log_entry = {
+            "request_id": request_id,
+            "input_type": input_type,
+            "user_id": user_id,
+            "status": response.status,
+            "risk_level": response.risk_level,
+            "risk_score": response.risk_score,
+            "categories": response.categories,
+            "processing_time_ms": response.processing_time_ms,
+        }
+        await mongo_service.log_moderation(log_entry)
+
+        # Update user history
+        if user_id:
+            await mongo_service.update_user_history(user_id, {
+                "risk_level": response.risk_level,
+                "categories": response.categories,
+            })
+            # Invalidate cached history
+            await redis_service.invalidate_user_history(user_id)
+
+    except Exception as e:
+        logger.warning("moderation_logging_failed", error=str(e))

app/api/v1/auth.py

@@ -0,0 +1,217 @@
+# app/api/v1/auth.py
+# Auth endpoints: register (parent), create-child, login, refresh, logout
+
+from datetime import datetime
+from fastapi import APIRouter, HTTPException, status, Depends
+from pydantic import BaseModel, EmailStr, field_validator
+from app.db.models.user import UserDocument, UserRole
+from app.core.security import hash_password, verify_password, create_access_token, create_refresh_token, decode_refresh_token
+from app.core.dependencies import get_current_user
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+router = APIRouter(prefix="/auth", tags=["Auth"])
+
+
+# ──────────────────────────────────────────────
+# Request / Response Schemas
+# ──────────────────────────────────────────────
+
+class RegisterParentRequest(BaseModel):
+    email: EmailStr
+    username: str
+    password: str
+    first_name: str
+    last_name: str
+    phone: str | None = None
+    consent_given: bool = True
+
+    @field_validator("username")
+    @classmethod
+    def username_length(cls, v: str) -> str:
+        if len(v) < 3 or len(v) > 30:
+            raise ValueError("Username must be 3–30 characters")
+        return v.strip()
+
+    @field_validator("password")
+    @classmethod
+    def password_length(cls, v: str) -> str:
+        if len(v) < 8:
+            raise ValueError("Password must be at least 8 characters")
+        return v
+
+
+class CreateChildRequest(BaseModel):
+    username: str
+    password: str
+    first_name: str
+    last_name: str
+
+
+class LoginRequest(BaseModel):
+    login: str   # email or username
+    password: str
+
+
+class RefreshRequest(BaseModel):
+    refresh_token: str
+
+
+class TokenResponse(BaseModel):
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+    expires_in: int  # seconds
+
+
+def _token_response(user: UserDocument) -> dict:
+    access = create_access_token(str(user.id), user.role.value)
+    refresh = create_refresh_token(str(user.id), user.role.value)
+    return {
+        "access_token": access,
+        "refresh_token": refresh,
+        "token_type": "bearer",
+        "expires_in": 15 * 60,
+        "user": user.to_public(),
+    }
+
+
+# ──────────────────────────────────────────────
+# Endpoints
+# ──────────────────────────────────────────────
+
+@router.post("/register", status_code=status.HTTP_201_CREATED)
+async def register_parent(body: RegisterParentRequest):
+    """Register a new parent account."""
+    # Email uniqueness
+    existing = await UserDocument.find_one(UserDocument.email == body.email)
+    if existing:
+        raise HTTPException(status_code=409, detail="Email already registered")
+
+    # Username uniqueness
+    existing_u = await UserDocument.find_one(UserDocument.username == body.username)
+    if existing_u:
+        raise HTTPException(status_code=409, detail="Username already taken")
+
+    user = UserDocument(
+        email=body.email,
+        username=body.username,
+        password_hash=hash_password(body.password),
+        role=UserRole.PARENT,
+        first_name=body.first_name,
+        last_name=body.last_name,
+        phone=body.phone,
+        consent_given=body.consent_given,
+        consent_date=datetime.utcnow() if body.consent_given else None,
+    )
+    await user.insert()
+
+    resp = _token_response(user)
+    # Save refresh token
+    user.refresh_tokens.append(resp["refresh_token"])
+    await user.save()
+
+    logger.info("parent_registered", user_id=str(user.id))
+    return {"success": True, **resp}
+
+
+@router.post("/create-child", status_code=status.HTTP_201_CREATED)
+async def create_child(
+    body: CreateChildRequest,
+    current_user: UserDocument = Depends(get_current_user),
+):
+    """Parent creates a child account linked to their account."""
+    if current_user.role != UserRole.PARENT:
+        raise HTTPException(status_code=403, detail="Only parents can create child accounts")
+
+    existing_u = await UserDocument.find_one(UserDocument.username == body.username)
+    if existing_u:
+        raise HTTPException(status_code=409, detail="Username already taken")
+
+    child = UserDocument(
+        username=body.username,
+        password_hash=hash_password(body.password),
+        role=UserRole.CHILD,
+        first_name=body.first_name,
+        last_name=body.last_name,
+        parent_id=str(current_user.id),
+        parental_consent=True,
+        consent_given=True,
+    )
+    await child.insert()
+
+    # Link child to parent
+    current_user.children.append(str(child.id))
+    await current_user.save()
+
+    logger.info("child_created", child_id=str(child.id), parent_id=str(current_user.id))
+    return {"success": True, "user": child.to_public()}
+
+
+@router.post("/login")
+async def login(body: LoginRequest):
+    """Login with email or username + password."""
+    login_val = body.login.strip().lower()
+
+    # Try email first, then username
+    if "@" in login_val:
+        user = await UserDocument.find_one(UserDocument.email == login_val)
+    else:
+        user = await UserDocument.find_one(UserDocument.username == login_val)
+
+    if not user or not verify_password(body.password, user.password_hash):
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+
+    if not user.is_active:
+        raise HTTPException(status_code=403, detail="Account is deactivated")
+
+    resp = _token_response(user)
+    user.refresh_tokens.append(resp["refresh_token"])
+    user.last_login_at = datetime.utcnow()
+    await user.save()
+
+    logger.info("user_logged_in", user_id=str(user.id))
+    return {"success": True, **resp}
+
+
+@router.post("/refresh")
+async def refresh_token(body: RefreshRequest):
+    """Exchange a valid refresh token for a new token pair."""
+    payload = decode_refresh_token(body.refresh_token)
+    if not payload:
+        raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
+
+    user = await UserDocument.get(payload["sub"])
+    if not user or not user.is_active:
+        raise HTTPException(status_code=401, detail="User not found")
+
+    if body.refresh_token not in user.refresh_tokens:
+        raise HTTPException(status_code=401, detail="Refresh token revoked")
+
+    # Rotate: remove old, add new
+    user.refresh_tokens.remove(body.refresh_token)
+    resp = _token_response(user)
+    user.refresh_tokens.append(resp["refresh_token"])
+    await user.save()
+
+    return {"success": True, **resp}
+
+
+@router.post("/logout")
+async def logout(
+    body: RefreshRequest | None = None,
+    current_user: UserDocument = Depends(get_current_user),
+):
+    """Revoke refresh token(s). Omit body to logout all devices."""
+    if body and body.refresh_token in current_user.refresh_tokens:
+        current_user.refresh_tokens.remove(body.refresh_token)
+    else:
+        current_user.refresh_tokens.clear()
+    await current_user.save()
+    return {"success": True, "message": "Logged out"}
+
+
+@router.get("/me")
+async def get_me(current_user: UserDocument = Depends(get_current_user)):
+    """Return current authenticated user's profile."""
+    return {"success": True, "user": current_user.to_public()}

app/api/v1/health.py

@@ -0,0 +1,51 @@
+# app/api/v1/health.py
+# Health check endpoints
+
+import time
+from fastapi import APIRouter
+from app.api.schemas.responses import HealthResponse
+from app.models.model_registry import model_registry
+from app.services.redis_service import redis_service
+from app.services.mongo_service import mongo_service
+from app.services.gemini_service import gemini_service
+
+router = APIRouter(tags=["Health"])
+
+_startup_time = time.time()
+
+
+@router.get("/health", response_model=HealthResponse)
+async def health_check():
+    """
+    Full health check — reports status of all models and services.
+    """
+    model_status = model_registry.get_status()
+    all_models_ok = all(model_status.values())
+
+    service_status = {
+        "mongodb": mongo_service.is_connected,
+        "redis": redis_service.is_connected,
+        "gemini": gemini_service.is_initialized,
+    }
+
+    overall = "healthy" if all_models_ok else "degraded"
+
+    return HealthResponse(
+        status=overall,
+        version="4.0.0",
+        models=model_status,
+        services=service_status,
+        uptime_seconds=round(time.time() - _startup_time, 1),
+    )
+
+
+@router.get("/health/models")
+async def model_health():
+    """Detailed model status."""
+    return model_registry.get_status()
+
+
+@router.get("/health/ping")
+async def ping():
+    """Lightweight liveness probe."""
+    return {"status": "ok"}

app/api/v1/history.py

@@ -0,0 +1,68 @@
+# app/api/v1/history.py
+# Moderation history endpoints
+
+from fastapi import APIRouter, HTTPException, Query
+from app.api.schemas.responses import HistoryResponse
+from app.services.mongo_service import mongo_service
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+router = APIRouter(tags=["History"])
+
+
+@router.get(
+    "/history/{user_id}",
+    response_model=HistoryResponse,
+    summary="Get moderation history for a user",
+)
+async def get_user_history(
+    user_id: str,
+    limit: int = Query(20, ge=1, le=100, description="Max results"),
+    skip: int = Query(0, ge=0, description="Results to skip"),
+):
+    """
+    Retrieve moderation history for a specific user.
+
+    Returns past moderation decisions with timestamps,
+    risk scores, and categories.
+    """
+    if not mongo_service.is_connected:
+        raise HTTPException(
+            status_code=503,
+            detail="MongoDB not available — history querying disabled",
+        )
+
+    results = await mongo_service.get_moderation_history(user_id, limit=limit, skip=skip)
+
+    return HistoryResponse(
+        user_id=user_id,
+        total=len(results),
+        results=results,
+    )
+
+
+@router.get(
+    "/history/{user_id}/summary",
+    summary="Get aggregated user stats",
+)
+async def get_user_summary(user_id: str):
+    """
+    Get aggregated moderation statistics for a user.
+
+    Includes total scans, violations, warnings, and category breakdown.
+    """
+    if not mongo_service.is_connected:
+        raise HTTPException(status_code=503, detail="MongoDB not available")
+
+    history = await mongo_service.get_user_history(user_id)
+    if not history:
+        return {
+            "user_id": user_id,
+            "total_scans": 0,
+            "total_violations": 0,
+            "total_warnings": 0,
+            "violation_categories": {},
+        }
+
+    return history

app/api/v1/scan.py

@@ -0,0 +1,166 @@
+# app/api/v1/scan.py
+# Authenticated scan endpoints — wraps the AI pipeline and persists results
+
+import time
+import uuid
+from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends
+from pydantic import BaseModel
+from app.core.dependencies import get_current_user, require_role
+from app.db.models.user import UserDocument, UserRole
+from app.db.models.scan_result import ScanResultDocument, RiskLevel
+from app.db.models.alert import AlertDocument, AlertSeverity, AlertStatus
+from app.pipeline.workflow import get_workflow, PipelineState
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+router = APIRouter(prefix="/scan", tags=["Scan"])
+
+_SEVERITY_MAP = {
+    "LOW": AlertSeverity.LOW,
+    "MEDIUM": AlertSeverity.MEDIUM,
+    "HIGH": AlertSeverity.HIGH,
+    "CRITICAL": AlertSeverity.CRITICAL,
+}
+
+
+async def _run_pipeline_and_persist(
+    input_type: str,
+    raw_content,
+    user: UserDocument,
+) -> dict:
+    """Run the AI pipeline, persist ScanResult and optional Alert, return result dict."""
+    start = time.time()
+
+    workflow = get_workflow()
+    state: PipelineState = {
+        "input_type": input_type,
+        "raw_content": raw_content,
+        "user_id": str(user.id),
+    }
+    result = await workflow.ainvoke(state)
+
+    if result.get("error"):
+        raise HTTPException(status_code=500, detail=result["error"])
+
+    risk = result.get("risk_score")
+    decision = result.get("decision")
+    filter_result = result.get("filter_result")
+    deep_result = result.get("deep_result")
+
+    risk_level_str = risk.level if risk else "LOW"
+    risk_score = risk.score if risk else 0.0
+    action = decision.action if decision else "ALLOWED"
+    categories = filter_result.categories if filter_result else []
+    is_flagged = filter_result.is_flagged if filter_result else False
+    reasoning = deep_result.reasoning if deep_result else None
+    processing_ms = int((time.time() - start) * 1000)
+
+    # Persist ScanResult
+    scan_doc = ScanResultDocument(
+        user_id=str(user.id),
+        input_type=input_type,
+        content_preview=(raw_content[:200] if isinstance(raw_content, str) else None),
+        risk_level=RiskLevel(risk_level_str),
+        risk_score=risk_score,
+        categories=categories,
+        is_flagged=is_flagged,
+        action=action,
+        reasoning=reasoning,
+        processing_time_ms=processing_ms,
+        deep_analysis_used=deep_result is not None,
+    )
+    await scan_doc.insert()
+
+    # Create Alert if child account and content is flagged
+    if is_flagged and user.role == UserRole.CHILD and user.parent_id:
+        severity = _SEVERITY_MAP.get(risk_level_str, AlertSeverity.LOW)
+        alert = AlertDocument(
+            child_id=str(user.id),
+            parent_id=user.parent_id,
+            scan_result_id=str(scan_doc.id),
+            severity=severity,
+            categories=categories,
+            severity_score=risk_score,
+        )
+        alert.generate_content()
+        await alert.insert()
+        logger.info("alert_created", alert_id=str(alert.id), child_id=str(user.id))
+
+    return {
+        "request_id": f"req_{uuid.uuid4().hex[:12]}",
+        "input_type": input_type,
+        "scan_id": str(scan_doc.id),
+        "status": action,
+        "risk_level": risk_level_str,
+        "risk_score": risk_score,
+        "categories": categories,
+        "is_flagged": is_flagged,
+        "reasoning": reasoning,
+        "processing_time_ms": processing_ms,
+    }
+
+
+# ──────────────────────────────────────────────
+# Endpoints
+# ──────────────────────────────────────────────
+
+class ScanTextRequest(BaseModel):
+    text: str
+
+
+@router.post("/text")
+async def scan_text(
+    body: ScanTextRequest,
+    user: UserDocument = Depends(get_current_user),
+):
+    """Scan text content. Requires authentication."""
+    if not body.text.strip():
+        raise HTTPException(status_code=400, detail="Text cannot be empty")
+    result = await _run_pipeline_and_persist("text", body.text, user)
+    return {"success": True, **result}
+
+
+@router.post("/image")
+async def scan_image(
+    file: UploadFile = File(...),
+    user: UserDocument = Depends(get_current_user),
+):
+    """Scan image content. Requires authentication."""
+    if not file.content_type or not file.content_type.startswith("image/"):
+        raise HTTPException(status_code=400, detail="File must be an image")
+
+    image_bytes = await file.read()
+    result = await _run_pipeline_and_persist("image", image_bytes, user)
+    return {"success": True, **result}
+
+
+@router.get("/history")
+async def get_scan_history(
+    page: int = 1,
+    limit: int = 20,
+    user: UserDocument = Depends(get_current_user),
+):
+    """Get scan history for the current user."""
+    skip = (page - 1) * limit
+    scans = await ScanResultDocument.find(
+        ScanResultDocument.user_id == str(user.id)
+    ).sort(-ScanResultDocument.created_at).skip(skip).limit(limit).to_list()
+
+    return {
+        "success": True,
+        "page": page,
+        "limit": limit,
+        "results": [
+            {
+                "id": str(s.id),
+                "input_type": s.input_type,
+                "risk_level": s.risk_level.value,
+                "risk_score": s.risk_score,
+                "action": s.action,
+                "is_flagged": s.is_flagged,
+                "categories": s.categories,
+                "created_at": s.created_at.isoformat(),
+            }
+            for s in scans
+        ],
+    }

app/api/v1/users.py

@@ -0,0 +1,61 @@
+# app/api/v1/users.py
+# User profile and parent-child management endpoints
+
+from fastapi import APIRouter, HTTPException, Depends
+from pydantic import BaseModel
+from typing import Optional
+from app.core.dependencies import get_current_user
+from app.db.models.user import UserDocument, UserRole
+
+router = APIRouter(prefix="/users", tags=["Users"])
+
+
+@router.get("/me")
+async def get_profile(user: UserDocument = Depends(get_current_user)):
+    return {"success": True, "user": user.to_public()}
+
+
+class UpdateProfileRequest(BaseModel):
+    first_name: Optional[str] = None
+    last_name: Optional[str] = None
+    phone: Optional[str] = None
+
+
+@router.patch("/me")
+async def update_profile(
+    body: UpdateProfileRequest,
+    user: UserDocument = Depends(get_current_user),
+):
+    if body.first_name:
+        user.first_name = body.first_name
+    if body.last_name:
+        user.last_name = body.last_name
+    if body.phone is not None:
+        user.phone = body.phone
+    await user.save()
+    return {"success": True, "user": user.to_public()}
+
+
+@router.get("/children")
+async def list_children(user: UserDocument = Depends(get_current_user)):
+    """Parent: list all linked child accounts."""
+    if user.role != UserRole.PARENT:
+        raise HTTPException(status_code=403, detail="Only parents can list children")
+    children = await UserDocument.find(
+        UserDocument.parent_id == str(user.id)
+    ).to_list()
+    return {"success": True, "children": [c.to_public() for c in children]}
+
+
+@router.get("/children/{child_id}")
+async def get_child(
+    child_id: str,
+    user: UserDocument = Depends(get_current_user),
+):
+    """Parent: get a specific child's profile."""
+    if user.role != UserRole.PARENT:
+        raise HTTPException(status_code=403, detail="Only parents can view child profiles")
+    child = await UserDocument.get(child_id)
+    if not child or child.parent_id != str(user.id):
+        raise HTTPException(status_code=404, detail="Child not found")
+    return {"success": True, "child": child.to_public()}

app/config.py

@@ -0,0 +1,99 @@
+# app/config.py
+# Centralized configuration via Pydantic Settings
+
+from pathlib import Path
+from functools import lru_cache
+from pydantic_settings import BaseSettings
+from pydantic import Field
+
+
+class Settings(BaseSettings):
+    """Application settings loaded from environment variables."""
+
+    # --- Server ---
+    host: str = "0.0.0.0"
+    port: int = 7860
+    env: str = "production"
+    log_level: str = "INFO"
+
+    # --- JWT ---
+    jwt_access_secret: str = Field(default="change-me-in-production-at-least-32-chars!!")
+    jwt_refresh_secret: str = Field(default="change-me-in-production-at-least-32-chars!!")
+    jwt_access_expires_minutes: int = 15
+    jwt_refresh_expires_days: int = 7
+
+    # --- Security ---
+    bcrypt_rounds: int = 12
+    cors_origins: str = "*"
+
+    # --- MongoDB ---
+    mongodb_uri: str = Field(default="")
+    mongodb_db_name: str = "hubble"
+
+    # --- Redis ---
+    redis_url: str = Field(default="")
+    redis_cache_ttl: int = 300  # seconds
+
+    # --- Gemini ---
+    gemini_api_keys: str = ""  # comma-separated
+    gemini_model: str = "gemini-2.5-flash"
+
+    # --- LangSmith ---
+    langsmith_api_key: str = ""
+    langsmith_project: str = "hubble-moderation"
+    langsmith_tracing_v2: bool = True
+
+    # --- Models ---
+    model_cache_dir: str = "/tmp/model_cache"
+    onnx_enabled: bool = False
+    text_model_name: str = "unitary/toxic-bert"
+    image_model_name: str = "google/efficientnet-b0"
+    clip_model_name: str = "openai/clip-vit-base-patch32"
+
+    # --- Risk Thresholds ---
+    risk_low_max: int = 30
+    risk_medium_max: int = 65
+
+    # --- Content Limits ---
+    max_text_length: int = 10000
+    max_image_size: int = 10 * 1024 * 1024   # 10MB
+    max_video_size: int = 50 * 1024 * 1024   # 50MB
+
+    # --- Video Processing ---
+    video_max_frames: int = 10
+    video_fps_sample: int = 1
+
+    model_config = {
+        "env_file": ".env",
+        "env_file_encoding": "utf-8",
+        "extra": "ignore",
+    }
+
+    @property
+    def gemini_keys_list(self) -> list[str]:
+        """Parse comma-separated Gemini API keys."""
+        if not self.gemini_api_keys:
+            return []
+        return [k.strip() for k in self.gemini_api_keys.split(",") if k.strip()]
+
+    @property
+    def cors_origins_list(self) -> list[str]:
+        """Parse comma-separated CORS origins."""
+        return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
+
+    @property
+    def model_cache_path(self) -> Path:
+        """Resolved path for model cache directory."""
+        path = Path(self.model_cache_dir)
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+
+    @property
+    def is_production(self) -> bool:
+        return self.env == "production"
+
+
+@lru_cache()
+def get_settings() -> Settings:
+    """Cached settings singleton."""
+    return Settings()

app/core/__init__.py

@@ -0,0 +1 @@
+# app/core/__init__.py

app/core/dependencies.py

@@ -0,0 +1,48 @@
+# app/core/dependencies.py
+# FastAPI dependency injection — auth guards, role checks, shared helpers
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from app.core.security import decode_access_token
+from app.db.models.user import UserDocument, UserRole
+
+_bearer = HTTPBearer(auto_error=False)
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
+) -> UserDocument:
+    """Extract and validate JWT; return the authenticated UserDocument."""
+    if not credentials:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Not authenticated",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    payload = decode_access_token(credentials.credentials)
+    if not payload:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+        )
+
+    user = await UserDocument.get(payload["sub"])
+    if not user or not user.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="User not found or deactivated",
+        )
+    return user
+
+
+def require_role(*roles: UserRole):
+    """Factory that returns a dependency requiring one of the given roles."""
+    async def _check(user: UserDocument = Depends(get_current_user)) -> UserDocument:
+        if user.role not in roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"Role '{user.role}' is not permitted for this action",
+            )
+        return user
+    return _check

app/core/security.py

@@ -0,0 +1,79 @@
+# app/core/security.py
+# Password hashing (argon2-cffi) + JWT creation/verification (python-jose)
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from argon2 import PasswordHasher
+from argon2.exceptions import VerifyMismatchError, VerificationError, InvalidHashError
+from jose import JWTError, jwt
+from app.config import get_settings
+
+settings = get_settings()
+
+# ──────────────────────────────────────────────
+# Password hashing — Argon2id (modern, OWASP recommended)
+# ──────────────────────────────────────────────
+
+_ph = PasswordHasher()
+
+
+def hash_password(plain: str) -> str:
+    return _ph.hash(plain)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    try:
+        return _ph.verify(hashed, plain)
+    except (VerifyMismatchError, VerificationError, InvalidHashError):
+        return False
+
+
+# ──────────────────────────────────────────────
+# JWT — HS256 access + refresh tokens
+# ──────────────────────────────────────────────
+
+ALGORITHM = "HS256"
+
+
+def _create_token(data: dict, secret: str, expires_delta: timedelta) -> str:
+    payload = data.copy()
+    payload["exp"] = datetime.now(timezone.utc) + expires_delta
+    return jwt.encode(payload, secret, algorithm=ALGORITHM)
+
+
+def create_access_token(user_id: str, role: str) -> str:
+    return _create_token(
+        {"sub": user_id, "role": role, "type": "access"},
+        settings.jwt_access_secret,
+        timedelta(minutes=settings.jwt_access_expires_minutes),
+    )
+
+
+def create_refresh_token(user_id: str, role: str) -> str:
+    return _create_token(
+        {"sub": user_id, "role": role, "type": "refresh"},
+        settings.jwt_refresh_secret,
+        timedelta(days=settings.jwt_refresh_expires_days),
+    )
+
+
+def decode_access_token(token: str) -> Optional[dict]:
+    """Returns payload dict or None on failure."""
+    try:
+        payload = jwt.decode(token, settings.jwt_access_secret, algorithms=[ALGORITHM])
+        if payload.get("type") != "access":
+            return None
+        return payload
+    except JWTError:
+        return None
+
+
+def decode_refresh_token(token: str) -> Optional[dict]:
+    """Returns payload dict or None on failure."""
+    try:
+        payload = jwt.decode(token, settings.jwt_refresh_secret, algorithms=[ALGORITHM])
+        if payload.get("type") != "refresh":
+            return None
+        return payload
+    except JWTError:
+        return None

app/db/__init__.py

@@ -0,0 +1 @@
+# app/db/__init__.py

app/db/connection.py

@@ -0,0 +1,36 @@
+# app/db/connection.py
+# Beanie ODM initialization — reuses the existing mongo_service Motor client
+
+from beanie import init_beanie
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+async def connect_db() -> None:
+    """Initialize Beanie ODM using the already-connected mongo_service client."""
+    from app.services.mongo_service import mongo_service
+    from app.db.models.user import UserDocument
+    from app.db.models.scan_result import ScanResultDocument
+    from app.db.models.alert import AlertDocument
+
+    if not mongo_service._connected or mongo_service.client is None:
+        logger.error("beanie_init_skipped", reason="mongo_service not connected")
+        return
+
+    try:
+        db = mongo_service.client[mongo_service.settings.mongodb_db_name]
+        await init_beanie(
+            database=db,
+            document_models=[UserDocument, ScanResultDocument, AlertDocument],
+            allow_index_dropping=False,
+        )
+        logger.info("beanie_initialized", db=mongo_service.settings.mongodb_db_name)
+    except Exception as e:
+        logger.error("beanie_init_failed", error=str(e))
+        raise
+
+
+async def close_db() -> None:
+    """No-op — Motor client is closed by mongo_service.disconnect()."""
+    pass

app/db/models/__init__.py

@@ -0,0 +1,10 @@
+# app/db/models/__init__.py
+from app.db.models.user import UserDocument, UserRole
+from app.db.models.scan_result import ScanResultDocument, RiskLevel
+from app.db.models.alert import AlertDocument, AlertSeverity, AlertStatus
+
+__all__ = [
+    "UserDocument", "UserRole",
+    "ScanResultDocument", "RiskLevel",
+    "AlertDocument", "AlertSeverity", "AlertStatus",
+]

app/db/models/alert.py

@@ -0,0 +1,75 @@
+# app/db/models/alert.py
+# Alert Beanie document
+
+from __future__ import annotations
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+from beanie import Document
+from pydantic import Field
+
+
+class AlertSeverity(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class AlertStatus(str, Enum):
+    PENDING = "pending"
+    ACKNOWLEDGED = "acknowledged"
+    RESOLVED = "resolved"
+
+
+class AlertDocument(Document):
+    """Parent alert generated when flagged content is detected for a child."""
+
+    child_id: str
+    parent_id: str
+    scan_result_id: str
+
+    title: str
+    message: str
+    guidance: str = ""
+
+    severity: AlertSeverity = AlertSeverity.LOW
+    categories: list[str] = Field(default_factory=list)
+    severity_score: float = 0.0
+
+    status: AlertStatus = AlertStatus.PENDING
+    parent_notified: bool = False
+    child_notified: bool = False
+
+    acknowledged_at: Optional[datetime] = None
+    resolved_at: Optional[datetime] = None
+    resolved_by: Optional[str] = None
+    resolution_notes: Optional[str] = None
+
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+    class Settings:
+        name = "alerts"
+
+    def generate_content(self) -> None:
+        """Populate title, message, guidance based on severity and categories."""
+        category_text = ", ".join(self.categories) if self.categories else "potentially harmful content"
+        score = int(self.severity_score * 100)
+
+        if self.severity == AlertSeverity.LOW:
+            self.title = "Mild Concern Detected"
+            self.message = f"Content flagged for: {category_text}. Score: {score}/100."
+            self.guidance = "This content contains some concerning elements. Consider talking about online safety."
+        elif self.severity == AlertSeverity.MEDIUM:
+            self.title = "Moderate Concern Detected"
+            self.message = f"Concerning content detected: {category_text}. Score: {score}/100."
+            self.guidance = "This content shows signs of potential cyberbullying. We recommend discussing this with your child."
+        elif self.severity == AlertSeverity.HIGH:
+            self.title = "⚠️ High Severity Alert"
+            self.message = f"Serious concern detected: {category_text}. Score: {score}/100."
+            self.guidance = "Immediate discussion with your child is recommended. Consider reaching out to school counselors."
+        else:  # CRITICAL
+            self.title = "🚨 CRITICAL ALERT - Immediate Action Required"
+            self.message = f"Critical content detected: {category_text}. Score: {score}/100."
+            self.guidance = "If there are threats of violence or self-harm, please contact emergency services immediately."

app/db/models/scan_result.py

@@ -0,0 +1,43 @@
+# app/db/models/scan_result.py
+# ScanResult Beanie document
+
+from __future__ import annotations
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+from beanie import Document
+from pydantic import Field
+
+
+class RiskLevel(str, Enum):
+    LOW = "LOW"
+    MEDIUM = "MEDIUM"
+    HIGH = "HIGH"
+    CRITICAL = "CRITICAL"
+
+
+class ScanResultDocument(Document):
+    """Persisted result of a content scan through the AI pipeline."""
+
+    user_id: str
+    input_type: str                     # text | image | video
+    content_preview: Optional[str] = None  # first 200 chars of text, or filename
+
+    # Risk
+    risk_level: RiskLevel = RiskLevel.LOW
+    risk_score: float = 0.0
+    categories: list[str] = Field(default_factory=list)
+    is_flagged: bool = False
+
+    # Decision
+    action: str = "ALLOWED"             # ALLOWED | WARNED | BLOCKED | ESCALATED
+    reasoning: Optional[str] = None
+
+    # Pipeline metadata
+    processing_time_ms: int = 0
+    deep_analysis_used: bool = False
+
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+
+    class Settings:
+        name = "scan_results"

app/db/models/user.py

@@ -0,0 +1,74 @@
+# app/db/models/user.py
+# Beanie User document — mirrors the Mongoose User model from Backend/
+
+from __future__ import annotations
+from datetime import datetime
+from enum import Enum
+from typing import Optional
+from beanie import Document, Indexed
+from pydantic import EmailStr, Field
+
+
+class UserRole(str, Enum):
+    PARENT = "parent"
+    CHILD = "child"
+    ADMIN = "admin"
+
+
+class UserDocument(Document):
+    """
+    User document stored in MongoDB.
+    Supports Parent → Child relationship for monitoring.
+    """
+
+    email: Optional[EmailStr] = None
+    username: Indexed(str, unique=True)  # type: ignore[valid-type]
+    password_hash: str  # bcrypt hash — never returned in API responses
+    role: UserRole = UserRole.PARENT
+    first_name: str
+    last_name: str
+    phone: Optional[str] = None
+    date_of_birth: Optional[datetime] = None
+
+    # Parent-Child relationship
+    parent_id: Optional[str] = None       # set for child accounts
+    children: list[str] = Field(default_factory=list)  # set for parent accounts
+
+    # Consent & Privacy
+    consent_given: bool = False
+    consent_date: Optional[datetime] = None
+    parental_consent: Optional[bool] = None  # for child accounts
+
+    # Account status
+    is_active: bool = True
+    is_verified: bool = False
+    last_login_at: Optional[datetime] = None
+
+    # Security — refresh tokens stored for rotation/revocation
+    refresh_tokens: list[str] = Field(default_factory=list)
+    password_changed_at: Optional[datetime] = None
+
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+
+    class Settings:
+        name = "users"
+        # Atlas already has the correct indexes from the Mongoose model.
+        # Beanie will use the field-level Indexed() annotations and won't
+        # try to re-create conflicting ones when allow_index_dropping=False.
+
+    def to_public(self) -> dict:
+        """Return safe user dict without sensitive fields."""
+        return {
+            "id": str(self.id),
+            "email": self.email,
+            "username": self.username,
+            "role": self.role.value,
+            "first_name": self.first_name,
+            "last_name": self.last_name,
+            "is_active": self.is_active,
+            "is_verified": self.is_verified,
+            "parent_id": self.parent_id,
+            "children": self.children,
+            "created_at": self.created_at.isoformat(),
+        }

app/dependencies.py

@@ -0,0 +1,29 @@
+# app/dependencies.py
+# FastAPI dependency injection
+
+from app.models.model_registry import model_registry
+from app.services.redis_service import redis_service
+from app.services.mongo_service import mongo_service
+
+
+async def require_models():
+    """Dependency: ensure models are loaded."""
+    status = model_registry.get_status()
+    if not status.get("text_model") and not status.get("image_model"):
+        from fastapi import HTTPException
+        raise HTTPException(
+            status_code=503,
+            detail="AI models not loaded. Server is starting up.",
+        )
+    return model_registry
+
+
+async def require_mongo():
+    """Dependency: ensure MongoDB is connected."""
+    if not mongo_service.is_connected:
+        from fastapi import HTTPException
+        raise HTTPException(
+            status_code=503,
+            detail="MongoDB not available.",
+        )
+    return mongo_service

app/main.py

@@ -0,0 +1,154 @@
+# app/main.py
+# FastAPI application factory — entry point for the Hubble AI Engine
+
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from app.config import get_settings
+from app.observability.logging import setup_logging, get_logger
+from app.observability.langsmith import setup_langsmith
+from app.models.model_registry import model_registry
+from app.services.redis_service import redis_service
+from app.services.mongo_service import mongo_service
+from app.services.gemini_service import gemini_service
+from app.pipeline.workflow import get_workflow
+from app.api.router import api_router
+from app.db.connection import connect_db, close_db
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """
+    Application lifespan manager.
+
+    Startup:
+    1. Configure logging
+    2. Setup LangSmith tracing
+    3. Connect MongoDB & Redis
+    4. Initialize Gemini service
+    5. Load all ML models
+    6. Compile LangGraph workflow
+
+    Shutdown:
+    1. Disconnect MongoDB & Redis
+    """
+    logger = get_logger(__name__)
+    settings = get_settings()
+
+    # ── Startup ──
+    logger.info("=" * 60)
+    logger.info("[STARTUP] HUBBLE AI ENGINE — Starting up...")
+    logger.info("=" * 60)
+
+    # 1. LangSmith
+    langsmith_ok = setup_langsmith()
+    logger.info("langsmith", enabled=langsmith_ok)
+
+    # 2. MongoDB (legacy motor service + Beanie ODM)
+    logger.info("connecting_mongodb")
+    await mongo_service.connect()
+    await connect_db()   # initializes Beanie document models
+
+    # 3. Redis
+    logger.info("connecting_redis")
+    await redis_service.connect()
+
+    # 4. Gemini
+    logger.info("initializing_gemini")
+    gemini_service.initialize()
+
+    # 5. ML Models
+    logger.info("loading_models")
+    model_status = await model_registry.load_all()
+    logger.info("models_loaded", status=model_status)
+
+    # 6. LangGraph Workflow
+    logger.info("compiling_workflow")
+    get_workflow()
+
+    logger.info("=" * 60)
+    logger.info("[READY] HUBBLE AI ENGINE — Ready!")
+    logger.info(f"   Environment: {settings.env}")
+    logger.info(f"   Port: {settings.port}")
+    logger.info(f"   Docs: http://localhost:{settings.port}/docs")
+    logger.info("=" * 60)
+
+    yield  # ── Application runs here ──
+
+    # ── Shutdown ──
+    logger.info("[SHUTDOWN] HUBBLE — Shutting down...")
+    await redis_service.disconnect()
+    await mongo_service.disconnect()
+    await close_db()
+    logger.info("Shutdown complete")
+
+
+def create_app() -> FastAPI:
+    """Create and configure the FastAPI application."""
+    settings = get_settings()
+
+    # Configure logging first
+    setup_logging()
+
+    app = FastAPI(
+        title="Hubble AI Engine — Cyberbullying Detection API",
+        description=(
+            "Production-grade content moderation pipeline with layered AI analysis. "
+            "Supports text, image, and video inputs with risk-based routing."
+        ),
+        version="4.0.0",
+        docs_url="/docs",
+        redoc_url="/redoc",
+        lifespan=lifespan,
+    )
+
+    # CORS — use origins from config
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_origins_list if settings.is_production else ["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    # Mount routes
+    app.include_router(api_router)
+
+    @app.get("/", include_in_schema=False)
+    async def root():
+        return JSONResponse({
+            "name": "Hubble Unified API",
+            "version": "5.0.0",
+            "docs": "/docs",
+            "health": "/health",
+            "endpoints": {
+                "auth": "POST /api/v1/auth/{register|login|refresh|logout}",
+                "users": "GET /api/v1/users/me",
+                "scan_text": "POST /api/v1/scan/text",
+                "scan_image": "POST /api/v1/scan/image",
+                "scan_history": "GET /api/v1/scan/history",
+                "alerts": "GET /api/v1/alerts",
+                "analyze_text": "POST /api/v1/analyze/text  (raw, no auth)",
+                "analyze_image": "POST /api/v1/analyze/image (raw, no auth)",
+            },
+        })
+
+    return app
+
+
+# Create the app instance
+app = create_app()
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    settings = get_settings()
+    uvicorn.run(
+        "app.main:app",
+        host=settings.host,
+        port=settings.port,
+        reload=not settings.is_production,
+    )

app/models/__init__.py

@@ -0,0 +1,2 @@
+# app/models/__init__.py
+"""Model loading, ONNX optimization, and inference."""

app/models/clip_model.py

@@ -0,0 +1,145 @@
+# app/models/clip_model.py
+# CLIP model for multimodal text-image alignment (deep analysis only)
+
+from PIL import Image
+import numpy as np
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class CLIPModel:
+    """
+    CLIP (Contrastive Language-Image Pre-Training) model.
+
+    Used in the deep analysis path to compute semantic alignment
+    between text descriptions and image content. This helps detect
+    subtle multimodal threats (e.g., threatening text overlaid on images).
+    """
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.model = None
+        self.preprocess = None
+        self.tokenizer = None
+        self._loaded = False
+        self.device = None
+
+    def load(self) -> None:
+        """Load the CLIP model and preprocessor."""
+        import torch
+        try:
+            import open_clip
+
+            model_name = self.settings.clip_model_name
+            cache_dir = self.settings.model_cache_path / "clip"
+            cache_dir.mkdir(parents=True, exist_ok=True)
+
+            self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+            logger.info("loading_clip_model", model=model_name)
+
+            # Use OpenCLIP for flexibility
+            self.model, _, self.preprocess = open_clip.create_model_and_transforms(
+                "ViT-B-32",
+                pretrained="laion2b_s34b_b79k",
+            )
+            self.model = self.model.to(self.device)
+            self.model.eval()
+
+            self.tokenizer = open_clip.get_tokenizer("ViT-B-32")
+
+            self._loaded = True
+            logger.info("clip_model_loaded")
+
+        except ImportError:
+            logger.warning("clip_not_available", reason="open_clip not installed")
+            self._loaded = False
+        except Exception as e:
+            logger.error("clip_load_failed", error=str(e))
+            self._loaded = False
+
+    def compute_similarity(self, image: Image.Image, texts: list[str]) -> dict:
+        """
+        Compute cosine similarity between an image and a list of text descriptions.
+
+        Args:
+            image: PIL Image.
+            texts: List of text descriptions to compare against.
+
+        Returns:
+            Dict with similarities, best_match, and best_score.
+        """
+        if not self._loaded:
+            return {"error": "CLIP model not loaded", "similarities": []}
+
+        import torch
+
+        # Preprocess image
+        image_input = self.preprocess(image).unsqueeze(0).to(self.device)
+
+        # Tokenize texts
+        text_tokens = self.tokenizer(texts).to(self.device)
+
+        with torch.no_grad():
+            image_features = self.model.encode_image(image_input)
+            text_features = self.model.encode_text(text_tokens)
+
+            # Normalize
+            image_features = image_features / image_features.norm(dim=-1, keepdim=True)
+            text_features = text_features / text_features.norm(dim=-1, keepdim=True)
+
+            # Cosine similarity
+            similarities = (image_features @ text_features.T).squeeze(0).cpu().numpy()
+
+        sim_list = similarities.tolist()
+        best_idx = int(np.argmax(sim_list))
+
+        return {
+            "similarities": dict(zip(texts, sim_list)),
+            "best_match": texts[best_idx],
+            "best_score": sim_list[best_idx],
+        }
+
+    def align_content(self, image: Image.Image, context_text: str | None = None) -> dict:
+        """
+        Analyze image alignment with harmful content categories.
+
+        Args:
+            image: Image to analyze.
+            context_text: Optional surrounding text context.
+
+        Returns:
+            Dict with category alignment scores.
+        """
+        harmful_descriptions = [
+            "a photo containing violence, fighting, or physical harm",
+            "a photo containing nudity or sexual content",
+            "a photo containing self-harm or suicide imagery",
+            "a photo containing hate symbols or extremist content",
+            "a photo containing drugs or substance abuse",
+            "a safe and appropriate photo for children",
+        ]
+
+        result = self.compute_similarity(image, harmful_descriptions)
+
+        if "error" in result:
+            return result
+
+        # Also check text-image alignment if context provided
+        text_alignment = None
+        if context_text:
+            text_result = self.compute_similarity(image, [context_text, "unrelated content"])
+            text_alignment = text_result["similarities"].get(context_text, 0.0)
+
+        return {
+            "category_scores": result["similarities"],
+            "most_aligned": result["best_match"],
+            "alignment_score": result["best_score"],
+            "text_image_alignment": text_alignment,
+        }
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded

app/models/image_model.py

@@ -0,0 +1,180 @@
+# app/models/image_model.py
+# EfficientNet-based image classification model with ONNX optimization
+
+from pathlib import Path
+import numpy as np
+from PIL import Image
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ImageClassificationModel:
+    """
+    Image content classifier using EfficientNet.
+
+    Detects violence, NSFW content, and other harmful imagery.
+    Supports ONNX (fast) and PyTorch (fallback) inference.
+    """
+
+    LABELS = ["safe", "violence", "nsfw", "self_harm", "hate_symbol"]
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.processor = None
+        self.onnx_session = None
+        self.pt_model = None
+        self.device = None
+        self._loaded = False
+        self._num_labels = len(self.LABELS)
+
+    def load(self) -> None:
+        """Load the image processor and model."""
+        from transformers import AutoImageProcessor, AutoModelForImageClassification
+
+        model_name = self.settings.image_model_name
+        cache_dir = self.settings.model_cache_path / "efficientnet"
+        onnx_path = cache_dir / "image_classifier.onnx"
+
+        logger.info("loading_image_model", model=model_name)
+
+        # Load image processor
+        try:
+            self.processor = AutoImageProcessor.from_pretrained(
+                model_name, cache_dir=cache_dir
+            )
+        except Exception:
+            # Fallback: use a generic processor
+            from transformers import AutoImageProcessor
+            self.processor = AutoImageProcessor.from_pretrained(
+                "google/efficientnet-b0", cache_dir=cache_dir
+            )
+
+        if self.settings.onnx_enabled and onnx_path.exists():
+            from app.models.onnx_utils import load_onnx_session
+            self.onnx_session = load_onnx_session(onnx_path)
+            logger.info("image_model_loaded", backend="onnx")
+        else:
+            self._load_pytorch(model_name, cache_dir)
+            if self.settings.onnx_enabled:
+                try:
+                    self._export_onnx(onnx_path)
+                    from app.models.onnx_utils import load_onnx_session
+                    self.onnx_session = load_onnx_session(onnx_path)
+                    self.pt_model = None
+                    logger.info("image_model_loaded", backend="onnx", note="exported")
+                except Exception as e:
+                    logger.warning("onnx_export_failed", error=str(e), fallback="pytorch")
+            else:
+                logger.info("image_model_loaded", backend="pytorch")
+
+        self._loaded = True
+
+    def _load_pytorch(self, model_name: str, cache_dir: Path) -> None:
+        """Load PyTorch model."""
+        import torch
+        from transformers import AutoModelForImageClassification
+
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        try:
+            self.pt_model = AutoModelForImageClassification.from_pretrained(
+                model_name, cache_dir=cache_dir
+            )
+        except Exception:
+            # If the model doesn't exist as a pretrained classifier, load base EfficientNet
+            self.pt_model = AutoModelForImageClassification.from_pretrained(
+                "google/efficientnet-b0", cache_dir=cache_dir
+            )
+        self.pt_model.to(self.device)
+        self.pt_model.eval()
+
+        # Update labels from model config if available
+        if hasattr(self.pt_model.config, "id2label"):
+            model_labels = list(self.pt_model.config.id2label.values())
+            if model_labels:
+                self._num_labels = len(model_labels)
+
+    def _export_onnx(self, onnx_path: Path) -> None:
+        """Export to ONNX."""
+        import torch
+        from app.models.onnx_utils import export_to_onnx
+
+        dummy_input = torch.randn(1, 3, 224, 224).to(self.device)
+        export_to_onnx(
+            model=self.pt_model,
+            sample_input={"pixel_values": dummy_input},
+            output_path=onnx_path,
+            input_names=["pixel_values"],
+            output_names=["logits"],
+        )
+
+    def predict(self, image: Image.Image) -> dict:
+        """
+        Classify an image for harmful content.
+
+        Args:
+            image: PIL Image (RGB).
+
+        Returns:
+            Dict with labels, scores, is_harmful, max_score, max_label.
+        """
+        if not self._loaded:
+            raise RuntimeError("Image model not loaded. Call load() first.")
+
+        # Preprocess with the model's processor
+        inputs = self.processor(images=image, return_tensors="np" if self.onnx_session else "pt")
+
+        if self.onnx_session:
+            return self._predict_onnx(inputs)
+        else:
+            return self._predict_pytorch(inputs)
+
+    def _predict_onnx(self, inputs) -> dict:
+        """ONNX inference."""
+        from app.models.onnx_utils import onnx_inference
+
+        pixel_values = inputs["pixel_values"].astype(np.float32)
+        outputs = onnx_inference(self.onnx_session, {"pixel_values": pixel_values})
+        logits = outputs[0][0]
+        return self._format_output(logits)
+
+    def _predict_pytorch(self, inputs) -> dict:
+        """PyTorch inference."""
+        import torch
+
+        inputs = {k: v.to(self.device) for k, v in inputs.items()}
+        with torch.no_grad():
+            outputs = self.pt_model(**inputs)
+            logits = outputs.logits[0].cpu().numpy()
+        return self._format_output(logits)
+
+    def _format_output(self, logits: np.ndarray) -> dict:
+        """Convert logits to prediction dict."""
+        # Softmax for single-label classification
+        exp_logits = np.exp(logits - np.max(logits))
+        scores = (exp_logits / exp_logits.sum()).tolist()
+
+        # Map to our labels (or use model's own labels)
+        if self.pt_model and hasattr(self.pt_model.config, "id2label"):
+            labels = [self.pt_model.config.id2label.get(i, f"class_{i}") for i in range(len(scores))]
+        else:
+            labels = [f"class_{i}" for i in range(len(scores))]
+
+        max_idx = int(np.argmax(scores))
+
+        # Determine if harmful (anything not classified as safe/non-violent)
+        safe_keywords = {"safe", "non-violence", "non_violence", "normal", "neutral"}
+        is_harmful = labels[max_idx].lower().replace("-", "_").replace(" ", "_") not in safe_keywords
+
+        return {
+            "labels": labels,
+            "scores": scores,
+            "is_harmful": is_harmful,
+            "max_score": scores[max_idx],
+            "max_label": labels[max_idx],
+        }
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded

app/models/model_registry.py

@@ -0,0 +1,99 @@
+# app/models/model_registry.py
+# Singleton model registry — loads and manages all ML models
+
+from app.models.text_model import TextToxicityModel
+from app.models.image_model import ImageClassificationModel
+from app.models.clip_model import CLIPModel
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class ModelRegistry:
+    """
+    Central registry for all ML models.
+
+    Provides lazy-loading and lifecycle management.
+    Models are loaded once and reused across requests.
+    """
+
+    def __init__(self):
+        self._text_model: TextToxicityModel | None = None
+        self._image_model: ImageClassificationModel | None = None
+        self._clip_model: CLIPModel | None = None
+
+    async def load_all(self) -> dict[str, bool]:
+        """
+        Load all models. Called during application startup.
+
+        Returns:
+            Dict of model name → loaded status.
+        """
+        results = {}
+
+        # Text model (required)
+        logger.info("registry_loading", model="text_toxicity")
+        try:
+            self._text_model = TextToxicityModel()
+            self._text_model.load()
+            results["text"] = True
+        except Exception as e:
+            logger.error("text_model_load_failed", error=str(e))
+            results["text"] = False
+
+        # Image model (required)
+        logger.info("registry_loading", model="image_classifier")
+        try:
+            self._image_model = ImageClassificationModel()
+            self._image_model.load()
+            results["image"] = True
+        except Exception as e:
+            logger.error("image_model_load_failed", error=str(e))
+            results["image"] = False
+
+        # CLIP model (optional — only for deep analysis)
+        logger.info("registry_loading", model="clip")
+        try:
+            self._clip_model = CLIPModel()
+            self._clip_model.load()
+            results["clip"] = self._clip_model.is_loaded
+        except Exception as e:
+            logger.warning("clip_model_load_failed", error=str(e))
+            results["clip"] = False
+
+        logger.info("registry_loaded", results=results)
+        return results
+
+    @property
+    def text_model(self) -> TextToxicityModel:
+        if self._text_model is None or not self._text_model.is_loaded:
+            raise RuntimeError("Text model not available")
+        return self._text_model
+
+    @property
+    def image_model(self) -> ImageClassificationModel:
+        if self._image_model is None or not self._image_model.is_loaded:
+            raise RuntimeError("Image model not available")
+        return self._image_model
+
+    @property
+    def clip_model(self) -> CLIPModel:
+        if self._clip_model is None:
+            raise RuntimeError("CLIP model not available")
+        return self._clip_model
+
+    @property
+    def clip_available(self) -> bool:
+        return self._clip_model is not None and self._clip_model.is_loaded
+
+    def get_status(self) -> dict:
+        """Get health status of all models."""
+        return {
+            "text_model": self._text_model.is_loaded if self._text_model else False,
+            "image_model": self._image_model.is_loaded if self._image_model else False,
+            "clip_model": self._clip_model.is_loaded if self._clip_model else False,
+        }
+
+
+# Global singleton
+model_registry = ModelRegistry()

app/models/onnx_utils.py

@@ -0,0 +1,120 @@
+# app/models/onnx_utils.py
+# ONNX export and inference utilities
+
+from pathlib import Path
+import numpy as np
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+def export_to_onnx(
+    model,
+    sample_input: dict,
+    output_path: Path,
+    input_names: list[str] | None = None,
+    output_names: list[str] | None = None,
+    dynamic_axes: dict | None = None,
+    opset_version: int = 14,
+) -> Path:
+    """
+    Export a PyTorch model to ONNX format.
+
+    Args:
+        model: PyTorch model (eval mode).
+        sample_input: Dict of tensor inputs for tracing.
+        output_path: Where to save the .onnx file.
+        input_names: Names for input tensors.
+        output_names: Names for output tensors.
+        dynamic_axes: Dynamic axes specification.
+        opset_version: ONNX opset version.
+
+    Returns:
+        Path to the exported ONNX model.
+    """
+    import torch
+
+    output_path = Path(output_path)
+    output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    if input_names is None:
+        input_names = list(sample_input.keys())
+    if output_names is None:
+        output_names = ["logits"]
+    if dynamic_axes is None:
+        dynamic_axes = {name: {0: "batch_size"} for name in input_names + output_names}
+
+    # Prepare ordered tuple of inputs
+    input_tuple = tuple(sample_input[name] for name in input_names)
+
+    model.eval()
+    with torch.no_grad():
+        torch.onnx.export(
+            model,
+            input_tuple,
+            str(output_path),
+            input_names=input_names,
+            output_names=output_names,
+            dynamic_axes=dynamic_axes,
+            opset_version=opset_version,
+            do_constant_folding=True,
+        )
+
+    logger.info("onnx_export_complete", path=str(output_path), size_mb=round(output_path.stat().st_size / 1e6, 1))
+    return output_path
+
+
+def load_onnx_session(model_path: Path, providers: list[str] | None = None):
+    """
+    Load an ONNX model as an InferenceSession.
+
+    Args:
+        model_path: Path to .onnx file.
+        providers: ONNX Runtime execution providers (defaults to CPU).
+
+    Returns:
+        ort.InferenceSession instance.
+    """
+    import onnxruntime as ort
+
+    if providers is None:
+        available = ort.get_available_providers()
+        # Prefer CUDA if available, else CPU
+        if "CUDAExecutionProvider" in available:
+            providers = ["CUDAExecutionProvider", "CPUExecutionProvider"]
+        else:
+            providers = ["CPUExecutionProvider"]
+
+    session = ort.InferenceSession(str(model_path), providers=providers)
+    logger.info(
+        "onnx_session_loaded",
+        path=str(model_path),
+        providers=providers,
+    )
+    return session
+
+
+def onnx_inference(session, inputs: dict[str, np.ndarray]) -> list[np.ndarray]:
+    """
+    Run inference on an ONNX session.
+
+    Args:
+        session: ONNX InferenceSession.
+        inputs: Dict mapping input names to numpy arrays.
+
+    Returns:
+        List of output numpy arrays.
+    """
+    # Ensure proper dtypes
+    feed = {}
+    for inp in session.get_inputs():
+        if inp.name in inputs:
+            arr = inputs[inp.name]
+            # Match expected dtype
+            if "int" in inp.type:
+                arr = arr.astype(np.int64)
+            else:
+                arr = arr.astype(np.float32)
+            feed[inp.name] = arr
+
+    return session.run(None, feed)

app/models/text_model.py

@@ -0,0 +1,178 @@
+# app/models/text_model.py
+# RoBERTa-based text toxicity model with ONNX optimization
+
+from pathlib import Path
+import numpy as np
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class TextToxicityModel:
+    """
+    Text toxicity classifier using a RoBERTa-based model.
+
+    Supports both ONNX (fast) and PyTorch (fallback) inference.
+    Model: unitary/toxic-bert (multi-label toxicity detection).
+
+    Labels: toxic, severe_toxic, obscene, threat, insult, identity_hate
+    """
+
+    LABELS = ["toxic", "severe_toxic", "obscene", "threat", "insult", "identity_hate"]
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.tokenizer = None
+        self.onnx_session = None
+        self.pt_model = None
+        self.device = None
+        self._loaded = False
+
+    def load(self) -> None:
+        """Load the tokenizer and model (ONNX preferred, PyTorch fallback)."""
+        from transformers import AutoTokenizer
+
+        model_name = self.settings.text_model_name
+        cache_dir = self.settings.model_cache_path / "roberta"
+        onnx_path = cache_dir / "text_toxicity.onnx"
+
+        logger.info("loading_text_model", model=model_name)
+
+        # Load tokenizer
+        self.tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
+
+        if self.settings.onnx_enabled and onnx_path.exists():
+            # Use existing ONNX model
+            from app.models.onnx_utils import load_onnx_session
+            self.onnx_session = load_onnx_session(onnx_path)
+            logger.info("text_model_loaded", backend="onnx")
+        elif self.settings.onnx_enabled:
+            # Load PyTorch, export to ONNX, then use ONNX
+            self._load_pytorch(model_name, cache_dir)
+            self._export_onnx(onnx_path)
+            # Switch to ONNX session
+            from app.models.onnx_utils import load_onnx_session
+            self.onnx_session = load_onnx_session(onnx_path)
+            self.pt_model = None  # Free PyTorch memory
+            logger.info("text_model_loaded", backend="onnx", note="exported_from_pytorch")
+        else:
+            # PyTorch only
+            self._load_pytorch(model_name, cache_dir)
+            logger.info("text_model_loaded", backend="pytorch")
+
+        self._loaded = True
+
+    def _load_pytorch(self, model_name: str, cache_dir: Path) -> None:
+        """Load the PyTorch model."""
+        import torch
+        from transformers import AutoModelForSequenceClassification
+
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.pt_model = AutoModelForSequenceClassification.from_pretrained(
+            model_name, cache_dir=cache_dir
+        )
+        self.pt_model.to(self.device)
+        self.pt_model.eval()
+
+    def _export_onnx(self, onnx_path: Path) -> None:
+        """Export current PyTorch model to ONNX."""
+        import torch
+        from app.models.onnx_utils import export_to_onnx
+
+        sample = self.tokenizer(
+            "test input for export",
+            return_tensors="pt",
+            padding="max_length",
+            truncation=True,
+            max_length=128,
+        )
+        sample = {k: v.to(self.device) for k, v in sample.items()}
+
+        export_to_onnx(
+            model=self.pt_model,
+            sample_input=sample,
+            output_path=onnx_path,
+            input_names=["input_ids", "attention_mask"],
+            output_names=["logits"],
+        )
+
+    def predict(self, text: str) -> dict:
+        """
+        Predict toxicity scores for input text.
+
+        Args:
+            text: Input text to classify.
+
+        Returns:
+            Dict with:
+                - labels: list of label names
+                - scores: list of per-label probabilities
+                - is_toxic: bool (any label > 0.5)
+                - max_score: float (highest toxicity probability)
+                - max_label: str (label with highest probability)
+        """
+        if not self._loaded:
+            raise RuntimeError("Text model not loaded. Call load() first.")
+
+        # Tokenize
+        encoding = self.tokenizer(
+            text,
+            return_tensors="np" if self.onnx_session else "pt",
+            padding="max_length",
+            truncation=True,
+            max_length=128,
+        )
+
+        if self.onnx_session:
+            return self._predict_onnx(encoding)
+        else:
+            return self._predict_pytorch(encoding)
+
+    def _predict_onnx(self, encoding: dict) -> dict:
+        """Run ONNX inference."""
+        from app.models.onnx_utils import onnx_inference
+
+        inputs = {
+            "input_ids": encoding["input_ids"].astype(np.int64),
+            "attention_mask": encoding["attention_mask"].astype(np.int64),
+        }
+        outputs = onnx_inference(self.onnx_session, inputs)
+        logits = outputs[0][0]  # (num_labels,)
+        return self._format_output(logits)
+
+    def _predict_pytorch(self, encoding: dict) -> dict:
+        """Run PyTorch inference."""
+        import torch
+
+        inputs = {k: v.to(self.device) for k, v in encoding.items()}
+        with torch.no_grad():
+            outputs = self.pt_model(**inputs)
+            logits = outputs.logits[0].cpu().numpy()
+        return self._format_output(logits)
+
+    def _format_output(self, logits: np.ndarray) -> dict:
+        """Convert raw logits to formatted prediction dict."""
+        # Sigmoid for multi-label classification
+        scores = 1 / (1 + np.exp(-logits))
+        scores = scores.tolist()
+
+        # Handle case where model has fewer outputs than expected labels
+        labels = self.LABELS[: len(scores)]
+
+        label_scores = dict(zip(labels, scores))
+        max_idx = int(np.argmax(scores))
+        is_toxic = any(s > 0.5 for s in scores)
+
+        return {
+            "labels": labels,
+            "scores": scores,
+            "label_scores": label_scores,
+            "is_toxic": is_toxic,
+            "max_score": scores[max_idx],
+            "max_label": labels[max_idx],
+        }
+
+    @property
+    def is_loaded(self) -> bool:
+        return self._loaded

app/observability/__init__.py

@@ -0,0 +1,2 @@
+# app/observability/__init__.py
+"""Observability: structured logging and LangSmith tracing."""

app/observability/langsmith.py

@@ -0,0 +1,59 @@
+# app/observability/langsmith.py
+# LangSmith tracing integration for pipeline observability
+
+import os
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+def setup_langsmith() -> bool:
+    """
+    Configure LangSmith tracing via environment variables.
+    LangChain/LangGraph automatically pick up these env vars.
+
+    Returns:
+        True if LangSmith is configured, False otherwise.
+    """
+    settings = get_settings()
+
+    if not settings.langsmith_api_key:
+        logger.info("langsmith_disabled", reason="No API key provided")
+        return False
+
+    os.environ["LANGCHAIN_TRACING_V2"] = str(settings.langsmith_tracing_v2).lower()
+    os.environ["LANGCHAIN_API_KEY"] = settings.langsmith_api_key
+    os.environ["LANGCHAIN_PROJECT"] = settings.langsmith_project
+
+    logger.info(
+        "langsmith_enabled",
+        project=settings.langsmith_project,
+    )
+    return True
+
+
+def get_trace_config(
+    run_name: str,
+    input_type: str,
+    metadata: dict | None = None,
+) -> dict:
+    """
+    Build configuration dict for LangGraph invoke calls.
+    This attaches metadata and tags to the LangSmith trace.
+
+    Args:
+        run_name: Human-readable name for this trace run.
+        input_type: The content type being analyzed (text/image/video).
+        metadata: Additional key-value metadata.
+
+    Returns:
+        Config dict to pass to workflow.invoke().
+    """
+    tags = [f"input:{input_type}", "hubble-moderation"]
+    config = {
+        "run_name": run_name,
+        "tags": tags,
+        "metadata": metadata or {},
+    }
+    return {"configurable": config}

app/observability/logging.py

@@ -0,0 +1,41 @@
+# app/observability/logging.py
+# Structured logging with structlog
+
+import sys
+import logging
+import structlog
+from app.config import get_settings
+
+
+def setup_logging() -> None:
+    """Configure structured logging for the application."""
+    settings = get_settings()
+
+    # Configure structlog
+    structlog.configure(
+        processors=[
+            structlog.contextvars.merge_contextvars,
+            structlog.processors.add_log_level,
+            structlog.processors.StackInfoRenderer(),
+            structlog.dev.set_exc_info,
+            structlog.processors.TimeStamper(fmt="iso"),
+            structlog.processors.JSONRenderer()
+            if settings.is_production
+            else structlog.dev.ConsoleRenderer(colors=True),
+        ],
+        wrapper_class=structlog.make_filtering_bound_logger(
+            logging.getLevelName(settings.log_level)
+        ),
+        context_class=dict,
+        logger_factory=structlog.PrintLoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+
+    # Silence noisy third-party loggers
+    for logger_name in ["uvicorn.access", "httpx", "httpcore"]:
+        logging.getLogger(logger_name).setLevel(logging.WARNING)
+
+
+def get_logger(name: str | None = None) -> structlog.BoundLogger:
+    """Get a structured logger instance."""
+    return structlog.get_logger(name or __name__)

app/pipeline/__init__.py

@@ -0,0 +1,2 @@
+# app/pipeline/__init__.py
+"""Core moderation pipeline: preprocess → filter → score → route → decide."""

app/pipeline/decision_engine.py

@@ -0,0 +1,142 @@
+# app/pipeline/decision_engine.py
+# Rule-based decision engine: final moderation verdict
+
+from dataclasses import dataclass, field
+from app.pipeline.risk_scorer import RiskScore
+from app.pipeline.deep_analyzer import DeepAnalysisResult
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class Decision:
+    """Final moderation decision."""
+    action: str  # ALLOWED, WARNING, BLOCKED
+    reason: str
+    severity: str  # low, medium, high, critical
+    categories: list[str] = field(default_factory=list)
+    should_alert_parent: bool = False
+    should_log: bool = True
+    escalation_notes: str | None = None
+
+
+class DecisionEngine:
+    """
+    Rule-based final decision engine.
+
+    Takes risk score + optional deep analysis and produces a final verdict.
+
+    Rules:
+    - LOW risk  → ALLOWED (no action)
+    - MEDIUM risk → WARNING (increment user warning count, log)
+    - HIGH risk + deep_confirmed → BLOCKED (alert parent, log, escalate if critical)
+    - HIGH risk + deep_not_confirmed → WARNING (false positive recovery)
+    - Repeat offender with MEDIUM → BLOCKED (escalate)
+    """
+
+    def decide(
+        self,
+        risk: RiskScore,
+        deep_result: DeepAnalysisResult | None = None,
+        user_history: dict | None = None,
+    ) -> Decision:
+        """
+        Produce final moderation decision.
+
+        Args:
+            risk: Composite risk score from the scoring engine.
+            deep_result: Optional deep analysis result (only for HIGH risk).
+            user_history: Optional user moderation history.
+
+        Returns:
+            Decision with action, reason, and metadata.
+        """
+
+        # === LOW RISK ===
+        if risk.level == "LOW":
+            decision = Decision(
+                action="ALLOWED",
+                reason="Content passed all safety checks",
+                severity="low",
+                should_log=False,  # Don't clutter logs with safe content
+            )
+
+        # === MEDIUM RISK ===
+        elif risk.level == "MEDIUM":
+            # Check for repeat offender escalation
+            if risk.repeat_offender:
+                decision = Decision(
+                    action="BLOCKED",
+                    reason="Repeat offender with moderately harmful content — escalated to block",
+                    severity="high",
+                    should_alert_parent=True,
+                    escalation_notes="User has repeated violation history. Medium-risk content escalated.",
+                )
+            else:
+                decision = Decision(
+                    action="WARNING",
+                    reason=f"Content flagged as potentially harmful (risk score: {risk.score})",
+                    severity="medium",
+                    should_alert_parent=False,
+                )
+
+        # === HIGH RISK ===
+        elif risk.level == "HIGH":
+            if deep_result and deep_result.is_confirmed:
+                # Deep analysis confirms the threat
+                severity = deep_result.severity
+                should_escalate = severity == "critical"
+
+                decision = Decision(
+                    action="BLOCKED",
+                    reason=deep_result.reasoning,
+                    severity=severity,
+                    categories=deep_result.categories,
+                    should_alert_parent=True,
+                    escalation_notes=(
+                        "CRITICAL: Immediate review required. "
+                        f"Recommended action: {deep_result.recommended_action}"
+                        if should_escalate
+                        else None
+                    ),
+                )
+            elif deep_result and not deep_result.is_confirmed:
+                # Deep analysis says it's a false positive
+                decision = Decision(
+                    action="WARNING",
+                    reason=(
+                        f"Content initially flagged as high-risk (score: {risk.score}) "
+                        f"but deep analysis did not confirm threat. "
+                        f"Reasoning: {deep_result.reasoning}"
+                    ),
+                    severity="low",
+                    should_alert_parent=False,
+                )
+            else:
+                # No deep analysis available — err on caution
+                decision = Decision(
+                    action="BLOCKED",
+                    reason=f"High-risk content detected (score: {risk.score}). Deep analysis unavailable.",
+                    severity="high",
+                    should_alert_parent=True,
+                    escalation_notes="Deep analysis was not performed. Manual review recommended.",
+                )
+
+        else:
+            # Fallback
+            decision = Decision(
+                action="WARNING",
+                reason="Unclassified risk level",
+                severity="medium",
+            )
+
+        logger.info(
+            "decision_made",
+            action=decision.action,
+            severity=decision.severity,
+            alert_parent=decision.should_alert_parent,
+            risk_score=risk.score,
+            risk_level=risk.level,
+        )
+        return decision

app/pipeline/deep_analyzer.py

@@ -0,0 +1,153 @@
+# app/pipeline/deep_analyzer.py
+# Deep analysis layer: CLIP + Gemini reasoning (HIGH risk only)
+
+from dataclasses import dataclass, field
+from PIL import Image
+from app.models.model_registry import model_registry
+from app.services.gemini_service import gemini_service
+from app.pipeline.fast_filter import FilterResult
+from app.utils.image_utils import image_to_base64
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class DeepAnalysisResult:
+    """Output from the deep analysis stage."""
+    is_confirmed: bool  # Does deep analysis confirm the threat?
+    severity: str  # low, medium, high, critical
+    reasoning: str  # Explanation from Gemini
+    categories: list[str] = field(default_factory=list)
+    recommended_action: str = "warn"  # allow, warn, block, escalate
+    confidence: float = 0.0
+    clip_scores: dict = field(default_factory=dict)
+    gemini_raw: dict = field(default_factory=dict)
+
+
+class DeepAnalyzer:
+    """
+    Deep analysis layer invoked only for HIGH-risk content.
+
+    Pipeline:
+    1. CLIP multimodal alignment (if image present)
+    2. Gemini reasoning via LangChain
+    3. Combine signals into final assessment
+
+    This layer trades speed for accuracy — expected latency: 1-3 seconds.
+    """
+
+    async def analyze_text(
+        self,
+        text: str,
+        filter_result: FilterResult,
+    ) -> DeepAnalysisResult:
+        """
+        Deep analysis for flagged text content.
+
+        Args:
+            text: Original text content.
+            filter_result: Results from the fast filter stage.
+
+        Returns:
+            DeepAnalysisResult with Gemini reasoning.
+        """
+        logger.info("deep_analysis_text_started")
+
+        # Prepare context from fast filter
+        context = {
+            "flagged_categories": filter_result.categories,
+            "max_score": filter_result.max_score,
+            "max_label": filter_result.max_label,
+            "all_scores": filter_result.scores,
+        }
+
+        # Invoke Gemini for contextual reasoning
+        gemini_result = await gemini_service.analyze_text(text, context)
+
+        result = self._build_result(gemini_result)
+
+        logger.info(
+            "deep_analysis_text_complete",
+            confirmed=result.is_confirmed,
+            severity=result.severity,
+            action=result.recommended_action,
+        )
+        return result
+
+    async def analyze_image(
+        self,
+        image: Image.Image,
+        filter_result: FilterResult,
+        context_text: str | None = None,
+    ) -> DeepAnalysisResult:
+        """
+        Deep analysis for flagged image content.
+
+        Args:
+            image: PIL Image.
+            filter_result: Results from the fast filter.
+            context_text: Optional text accompanying the image.
+
+        Returns:
+            DeepAnalysisResult with CLIP alignment + Gemini reasoning.
+        """
+        logger.info("deep_analysis_image_started")
+
+        # Step 1: CLIP multimodal alignment
+        clip_scores = {}
+        if model_registry.clip_available:
+            try:
+                clip_result = model_registry.clip_model.align_content(image, context_text)
+                clip_scores = clip_result
+                logger.info("clip_alignment_complete", most_aligned=clip_result.get("most_aligned"))
+            except Exception as e:
+                logger.warning("clip_alignment_failed", error=str(e))
+
+        # Step 2: Gemini image reasoning
+        context = {
+            "flagged_categories": filter_result.categories,
+            "max_score": filter_result.max_score,
+            "clip_alignment": clip_scores.get("most_aligned", "unknown"),
+        }
+
+        image_b64 = image_to_base64(image)
+        gemini_result = await gemini_service.analyze_image(image_b64, context)
+
+        result = self._build_result(gemini_result, clip_scores)
+
+        logger.info(
+            "deep_analysis_image_complete",
+            confirmed=result.is_confirmed,
+            severity=result.severity,
+        )
+        return result
+
+    def _build_result(
+        self,
+        gemini_result: dict,
+        clip_scores: dict | None = None,
+    ) -> DeepAnalysisResult:
+        """Build DeepAnalysisResult from Gemini response."""
+        if "error" in gemini_result:
+            # Gemini failed — err on the side of caution
+            return DeepAnalysisResult(
+                is_confirmed=True,  # Assume harmful if we can't verify
+                severity="medium",
+                reasoning=f"Deep analysis unavailable: {gemini_result['error']}. Defaulting to caution.",
+                recommended_action="warn",
+                confidence=0.3,
+                clip_scores=clip_scores or {},
+                gemini_raw=gemini_result,
+            )
+
+        return DeepAnalysisResult(
+            is_confirmed=gemini_result.get("is_confirmed", False),
+            severity=gemini_result.get("severity", "medium"),
+            reasoning=gemini_result.get("reasoning", "No reasoning provided"),
+            categories=gemini_result.get("categories", []),
+            recommended_action=gemini_result.get("recommended_action", "warn"),
+            confidence=gemini_result.get("confidence", 0.5),
+            clip_scores=clip_scores or {},
+            gemini_raw=gemini_result,
+        )

app/pipeline/fast_filter.py

@@ -0,0 +1,124 @@
+# app/pipeline/fast_filter.py
+# First-pass AI classification using ONNX-optimized models
+
+from dataclasses import dataclass, field
+from PIL import Image
+from app.models.model_registry import model_registry
+from app.pipeline.preprocessor import ProcessedText, ProcessedImage
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class FilterResult:
+    """Output from the fast filter stage."""
+    input_type: str  # "text", "image"
+    is_flagged: bool
+    scores: dict[str, float] = field(default_factory=dict)
+    max_score: float = 0.0
+    max_label: str = ""
+    categories: list[str] = field(default_factory=list)
+    confidence: float = 0.0
+
+
+class FastFilter:
+    """
+    Fast AI filter using ONNX-optimized models.
+
+    - RoBERTa for text toxicity (multi-label)
+    - EfficientNet for image classification
+
+    This is the first gate in the pipeline. Designed for speed (<200ms).
+    """
+
+    # Toxicity threshold for flagging
+    TEXT_FLAG_THRESHOLD = 0.4
+    IMAGE_FLAG_THRESHOLD = 0.5
+
+    def filter_text(self, processed: ProcessedText) -> FilterResult:
+        """
+        Run RoBERTa text toxicity inference.
+
+        Args:
+            processed: Preprocessed text input.
+
+        Returns:
+            FilterResult with per-category toxicity scores.
+        """
+        text_model = model_registry.text_model
+        prediction = text_model.predict(processed.cleaned)
+
+        # Determine which categories are flagged
+        flagged_categories = []
+        label_scores = prediction.get("label_scores", {})
+        for label, score in label_scores.items():
+            if score > self.TEXT_FLAG_THRESHOLD:
+                flagged_categories.append(label)
+
+        is_flagged = len(flagged_categories) > 0
+
+        result = FilterResult(
+            input_type="text",
+            is_flagged=is_flagged,
+            scores=label_scores,
+            max_score=prediction["max_score"],
+            max_label=prediction["max_label"],
+            categories=flagged_categories,
+            confidence=prediction["max_score"],
+        )
+
+        logger.info(
+            "fast_filter_text",
+            flagged=is_flagged,
+            max_label=result.max_label,
+            max_score=round(result.max_score, 3),
+            categories=flagged_categories,
+        )
+        return result
+
+    def filter_image(self, processed: ProcessedImage) -> FilterResult:
+        """
+        Run EfficientNet image classification inference.
+
+        Args:
+            processed: Preprocessed image input.
+
+        Returns:
+            FilterResult with image classification scores.
+        """
+        image_model = model_registry.image_model
+        prediction = image_model.predict(processed.image)
+
+        # Map model output to categories
+        scores = {}
+        for label, score in zip(prediction["labels"], prediction["scores"]):
+            scores[label] = score
+
+        flagged_categories = []
+        for label, score in scores.items():
+            if score > self.IMAGE_FLAG_THRESHOLD:
+                # Check if this is a harmful category
+                safe_labels = {"safe", "non-violence", "non_violence", "normal", "neutral"}
+                if label.lower().replace("-", "_").replace(" ", "_") not in safe_labels:
+                    flagged_categories.append(label)
+
+        is_flagged = prediction["is_harmful"]
+
+        result = FilterResult(
+            input_type="image",
+            is_flagged=is_flagged,
+            scores=scores,
+            max_score=prediction["max_score"],
+            max_label=prediction["max_label"],
+            categories=flagged_categories,
+            confidence=prediction["max_score"],
+        )
+
+        logger.info(
+            "fast_filter_image",
+            flagged=is_flagged,
+            max_label=result.max_label,
+            max_score=round(result.max_score, 3),
+        )
+        return result

app/pipeline/preprocessor.py

@@ -0,0 +1,141 @@
+# app/pipeline/preprocessor.py
+# Input preprocessing: normalization, frame extraction, cleaning
+
+import re
+from dataclasses import dataclass, field
+from PIL import Image
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class ProcessedText:
+    """Preprocessed text content."""
+    original: str
+    cleaned: str
+    word_count: int
+    char_count: int
+    language: str = "en"  # placeholder for language detection
+
+
+@dataclass
+class ProcessedImage:
+    """Preprocessed image content."""
+    image: Image.Image
+    width: int
+    height: int
+    format: str = "RGB"
+
+
+@dataclass
+class ProcessedVideo:
+    """Preprocessed video — a list of extracted frames."""
+    frames: list[ProcessedImage] = field(default_factory=list)
+    frame_count: int = 0
+    duration_seconds: float = 0.0
+    metadata: dict = field(default_factory=dict)
+
+
+class Preprocessor:
+    """
+    Input preprocessing for all content types.
+
+    - Text: cleaning, normalization
+    - Image: resize, format conversion
+    - Video: frame extraction + per-frame preprocessing
+    """
+
+    def __init__(self):
+        self.settings = get_settings()
+
+    def process_text(self, text: str) -> ProcessedText:
+        """
+        Clean and normalize input text.
+
+        - Strip excessive whitespace
+        - Remove zero-width characters
+        - Normalize unicode
+        """
+        import unicodedata
+
+        # Remove zero-width characters often used for obfuscation
+        cleaned = re.sub(r"[\u200b\u200c\u200d\ufeff]", "", text)
+
+        # Normalize unicode
+        cleaned = unicodedata.normalize("NFKC", cleaned)
+
+        # Collapse excessive whitespace
+        cleaned = re.sub(r"\s+", " ", cleaned).strip()
+
+        result = ProcessedText(
+            original=text,
+            cleaned=cleaned,
+            word_count=len(cleaned.split()),
+            char_count=len(cleaned),
+        )
+
+        logger.debug(
+            "text_preprocessed",
+            word_count=result.word_count,
+            char_count=result.char_count,
+        )
+        return result
+
+    def process_image(self, image_bytes: bytes) -> ProcessedImage:
+        """
+        Load and preprocess image from bytes.
+
+        - Convert to RGB
+        - Record dimensions
+        """
+        from app.utils.image_utils import load_image_from_bytes
+
+        image = load_image_from_bytes(image_bytes)
+        width, height = image.size
+
+        result = ProcessedImage(
+            image=image,
+            width=width,
+            height=height,
+        )
+
+        logger.debug("image_preprocessed", width=width, height=height)
+        return result
+
+    def process_video(self, video_bytes: bytes) -> ProcessedVideo:
+        """
+        Extract key frames from video.
+
+        Uses OpenCV to sample frames at configured intervals.
+        """
+        from app.utils.video_utils import extract_frames, get_video_metadata
+
+        metadata = get_video_metadata(video_bytes)
+        frames_pil = extract_frames(
+            video_bytes,
+            max_frames=self.settings.video_max_frames,
+            fps_sample=self.settings.video_fps_sample,
+        )
+
+        processed_frames = []
+        for frame in frames_pil:
+            w, h = frame.size
+            processed_frames.append(
+                ProcessedImage(image=frame, width=w, height=h)
+            )
+
+        result = ProcessedVideo(
+            frames=processed_frames,
+            frame_count=len(processed_frames),
+            duration_seconds=metadata.get("duration_seconds", 0.0),
+            metadata=metadata,
+        )
+
+        logger.debug(
+            "video_preprocessed",
+            frames_extracted=result.frame_count,
+            duration=result.duration_seconds,
+        )
+        return result

app/pipeline/risk_scorer.py

@@ -0,0 +1,166 @@
+# app/pipeline/risk_scorer.py
+# Composite risk scoring engine
+
+from dataclasses import dataclass
+from app.config import get_settings
+from app.pipeline.fast_filter import FilterResult
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class RiskScore:
+    """Composite risk assessment."""
+    score: float  # 0-100
+    level: str  # LOW, MEDIUM, HIGH
+    components: dict  # breakdown of scoring factors
+    repeat_offender: bool = False
+
+
+class RiskScorer:
+    """
+    Computes a composite risk score (0-100) from multiple signals.
+
+    Scoring formula:
+    - Base score from model confidence (weighted by category severity)
+    - Repeat offender boost (user history)
+    - Multi-category penalty (multiple harmful categories = higher risk)
+
+    Thresholds (configurable via env):
+    - 0-30:  LOW    → Allow
+    - 31-65: MEDIUM → Warning
+    - 66-100: HIGH  → Deep Analysis
+    """
+
+    # Category severity weights (how dangerous each type is)
+    CATEGORY_WEIGHTS = {
+        # Text categories (from RoBERTa toxic-bert)
+        "toxic": 0.6,
+        "severe_toxic": 1.0,
+        "obscene": 0.5,
+        "threat": 1.0,
+        "insult": 0.5,
+        "identity_hate": 0.9,
+        # Image categories
+        "violence": 0.9,
+        "nsfw": 0.8,
+        "self_harm": 1.0,
+        "hate_symbol": 0.9,
+        # Generic fallback
+        "harassment": 0.7,
+        "bullying": 0.7,
+    }
+
+    # Repeat offender thresholds
+    REPEAT_OFFENDER_VIOLATIONS = 3
+    REPEAT_OFFENDER_BOOST = 15  # points added
+
+    def __init__(self):
+        self.settings = get_settings()
+
+    def score(
+        self,
+        filter_result: FilterResult,
+        user_history: dict | None = None,
+    ) -> RiskScore:
+        """
+        Compute composite risk score.
+
+        Args:
+            filter_result: Output from fast filter stage.
+            user_history: Optional user moderation history.
+
+        Returns:
+            RiskScore with level classification.
+        """
+        # 1. Base score from model confidence
+        base_score = self._compute_base_score(filter_result)
+
+        # 2. Multi-category penalty
+        multi_cat_penalty = self._multi_category_penalty(filter_result)
+
+        # 3. Repeat offender boost
+        repeat_boost, is_repeat = self._repeat_offender_boost(user_history)
+
+        # 4. Combine
+        raw_score = base_score + multi_cat_penalty + repeat_boost
+        final_score = min(100.0, max(0.0, raw_score))
+
+        # 5. Classify level
+        level = self._classify_level(final_score)
+
+        result = RiskScore(
+            score=round(final_score, 1),
+            level=level,
+            components={
+                "base_score": round(base_score, 1),
+                "multi_category_penalty": round(multi_cat_penalty, 1),
+                "repeat_offender_boost": round(repeat_boost, 1),
+            },
+            repeat_offender=is_repeat,
+        )
+
+        logger.info(
+            "risk_scored",
+            score=result.score,
+            level=result.level,
+            components=result.components,
+            repeat_offender=is_repeat,
+        )
+        return result
+
+    def _compute_base_score(self, result: FilterResult) -> float:
+        """
+        Compute base score from model predictions.
+
+        Uses weighted sum of flagged category scores.
+        """
+        if not result.is_flagged:
+            # Even unflagged content gets a small score based on max prediction
+            return result.max_score * 20  # Scale 0-1 → 0-20
+
+        # Weighted sum of flagged category scores
+        weighted_sum = 0.0
+        weight_total = 0.0
+
+        for category, score in result.scores.items():
+            weight = self.CATEGORY_WEIGHTS.get(category.lower(), 0.5)
+            weighted_sum += score * weight * 100
+            weight_total += weight
+
+        if weight_total > 0:
+            return weighted_sum / weight_total
+        return result.max_score * 60
+
+    def _multi_category_penalty(self, result: FilterResult) -> float:
+        """Add penalty when multiple harmful categories are detected."""
+        num_categories = len(result.categories)
+        if num_categories <= 1:
+            return 0.0
+        # Each additional category adds 5 points
+        return (num_categories - 1) * 5.0
+
+    def _repeat_offender_boost(self, user_history: dict | None) -> tuple[float, bool]:
+        """Boost score for users with violation history."""
+        if not user_history:
+            return 0.0, False
+
+        total_violations = user_history.get("total_violations", 0)
+        is_repeat = total_violations >= self.REPEAT_OFFENDER_VIOLATIONS
+
+        if is_repeat:
+            return self.REPEAT_OFFENDER_BOOST, True
+        elif total_violations > 0:
+            # Smaller boost for users with some history
+            return total_violations * 3.0, False
+        return 0.0, False
+
+    def _classify_level(self, score: float) -> str:
+        """Map numeric score to risk level."""
+        if score <= self.settings.risk_low_max:
+            return "LOW"
+        elif score <= self.settings.risk_medium_max:
+            return "MEDIUM"
+        else:
+            return "HIGH"

app/pipeline/workflow.py

@@ -0,0 +1,327 @@
+# app/pipeline/workflow.py
+# LangGraph state machine — orchestrates the full moderation pipeline
+
+from __future__ import annotations
+from typing import Any, TypedDict, Literal
+from dataclasses import asdict
+
+from langgraph.graph import StateGraph, END
+
+from app.pipeline.preprocessor import (
+    Preprocessor,
+    ProcessedText,
+    ProcessedImage,
+    ProcessedVideo,
+)
+from app.pipeline.fast_filter import FastFilter, FilterResult
+from app.pipeline.risk_scorer import RiskScorer, RiskScore
+from app.pipeline.deep_analyzer import DeepAnalyzer, DeepAnalysisResult
+from app.pipeline.decision_engine import DecisionEngine, Decision
+from app.services.mongo_service import mongo_service
+from app.services.redis_service import redis_service
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+# ──────────────────────────────────────────────
+# Pipeline State Schema
+# ──────────────────────────────────────────────
+
+class PipelineState(TypedDict, total=False):
+    """State that flows through the LangGraph pipeline."""
+    # Input
+    input_type: str  # "text", "image", "video"
+    raw_content: Any  # str for text, bytes for image/video
+    user_id: str | None
+
+    # Preprocessed
+    processed_text: ProcessedText | None
+    processed_image: ProcessedImage | None
+    processed_video: ProcessedVideo | None
+
+    # Pipeline stages
+    filter_result: FilterResult | None
+    filter_results: list[FilterResult]  # For video (multiple frames)
+    risk_score: RiskScore | None
+    deep_result: DeepAnalysisResult | None
+    decision: Decision | None
+
+    # Context
+    user_history: dict | None
+
+    # Metadata
+    error: str | None
+
+
+# ──────────────────────────────────────────────
+# Pipeline Node Functions
+# ──────────────────────────────────────────────
+
+preprocessor = Preprocessor()
+fast_filter = FastFilter()
+risk_scorer = RiskScorer()
+deep_analyzer = DeepAnalyzer()
+decision_engine = DecisionEngine()
+
+
+async def preprocess_node(state: PipelineState) -> dict:
+    """Node 1: Preprocess the raw input."""
+    input_type = state["input_type"]
+    raw = state["raw_content"]
+
+    try:
+        if input_type == "text":
+            processed = preprocessor.process_text(raw)
+            return {"processed_text": processed}
+
+        elif input_type == "image":
+            processed = preprocessor.process_image(raw)
+            return {"processed_image": processed}
+
+        elif input_type == "video":
+            processed = preprocessor.process_video(raw)
+            return {"processed_video": processed}
+
+        else:
+            return {"error": f"Unknown input type: {input_type}"}
+
+    except Exception as e:
+        logger.error("preprocess_failed", error=str(e))
+        return {"error": f"Preprocessing failed: {str(e)}"}
+
+
+async def fetch_user_history_node(state: PipelineState) -> dict:
+    """Node 1b: Fetch user moderation history (parallel with preprocess)."""
+    user_id = state.get("user_id")
+    if not user_id:
+        return {"user_history": None}
+
+    # Try Redis cache first
+    cached = await redis_service.get_user_history(user_id)
+    if cached:
+        return {"user_history": cached}
+
+    # Fall back to MongoDB
+    history = await mongo_service.get_user_history(user_id)
+    if history:
+        await redis_service.cache_user_history(user_id, history)
+    return {"user_history": history}
+
+
+async def fast_filter_node(state: PipelineState) -> dict:
+    """Node 2: Run fast AI filter."""
+    input_type = state["input_type"]
+
+    try:
+        if input_type == "text" and state.get("processed_text"):
+            result = fast_filter.filter_text(state["processed_text"])
+            return {"filter_result": result}
+
+        elif input_type == "image" and state.get("processed_image"):
+            result = fast_filter.filter_image(state["processed_image"])
+            return {"filter_result": result}
+
+        elif input_type == "video" and state.get("processed_video"):
+            # Analyze each frame, take the worst result
+            video = state["processed_video"]
+            frame_results = []
+            for frame in video.frames:
+                result = fast_filter.filter_image(frame)
+                frame_results.append(result)
+
+            # Use the highest-risk frame as the representative result
+            if frame_results:
+                worst = max(frame_results, key=lambda r: r.max_score)
+                return {
+                    "filter_result": worst,
+                    "filter_results": frame_results,
+                }
+            else:
+                return {
+                    "filter_result": FilterResult(
+                        input_type="video",
+                        is_flagged=False,
+                        max_score=0.0,
+                    )
+                }
+
+        return {"error": "No processed content available for filtering"}
+
+    except Exception as e:
+        logger.error("fast_filter_failed", error=str(e))
+        return {"error": f"Fast filter failed: {str(e)}"}
+
+
+async def risk_score_node(state: PipelineState) -> dict:
+    """Node 3: Compute composite risk score."""
+    filter_result = state.get("filter_result")
+    if not filter_result:
+        return {"error": "No filter result to score"}
+
+    try:
+        user_history = state.get("user_history")
+        score = risk_scorer.score(filter_result, user_history)
+        return {"risk_score": score}
+    except Exception as e:
+        logger.error("risk_score_failed", error=str(e))
+        return {"error": f"Risk scoring failed: {str(e)}"}
+
+
+def route_by_risk(state: PipelineState) -> str:
+    """
+    Conditional router: decides whether to do deep analysis or skip to decision.
+
+    - LOW / MEDIUM → skip directly to decision
+    - HIGH → go to deep analysis
+    """
+    risk = state.get("risk_score")
+    if risk and risk.level == "HIGH":
+        return "deep_analysis"
+    return "decide"
+
+
+async def deep_analysis_node(state: PipelineState) -> dict:
+    """Node 4 (conditional): Deep analysis with CLIP + Gemini."""
+    input_type = state["input_type"]
+    filter_result = state.get("filter_result")
+
+    try:
+        if input_type == "text" and state.get("processed_text"):
+            result = await deep_analyzer.analyze_text(
+                state["processed_text"].cleaned,
+                filter_result,
+            )
+            return {"deep_result": result}
+
+        elif input_type in ("image", "video") and state.get("processed_image"):
+            result = await deep_analyzer.analyze_image(
+                state["processed_image"].image,
+                filter_result,
+            )
+            return {"deep_result": result}
+
+        elif input_type == "video" and state.get("processed_video"):
+            # Use the worst frame for deep analysis
+            video = state["processed_video"]
+            if video.frames:
+                # Find the worst frame based on filter_results
+                worst_frame = video.frames[0]
+                filter_results = state.get("filter_results", [])
+                if filter_results:
+                    worst_idx = max(
+                        range(len(filter_results)),
+                        key=lambda i: filter_results[i].max_score,
+                    )
+                    if worst_idx < len(video.frames):
+                        worst_frame = video.frames[worst_idx]
+
+                result = await deep_analyzer.analyze_image(
+                    worst_frame.image,
+                    filter_result,
+                )
+                return {"deep_result": result}
+
+        return {"deep_result": None}
+
+    except Exception as e:
+        logger.error("deep_analysis_failed", error=str(e))
+        return {"deep_result": None}
+
+
+async def decision_node(state: PipelineState) -> dict:
+    """Node 5: Final decision."""
+    risk = state.get("risk_score")
+    if not risk:
+        # Emergency fallback
+        return {
+            "decision": Decision(
+                action="WARNING",
+                reason="Pipeline error: no risk score available",
+                severity="medium",
+            )
+        }
+
+    try:
+        deep_result = state.get("deep_result")
+        user_history = state.get("user_history")
+        decision = decision_engine.decide(risk, deep_result, user_history)
+        return {"decision": decision}
+    except Exception as e:
+        logger.error("decision_failed", error=str(e))
+        return {
+            "decision": Decision(
+                action="WARNING",
+                reason=f"Decision engine error: {str(e)}",
+                severity="medium",
+            )
+        }
+
+
+# ──────────────────────────────────────────────
+# Build the LangGraph Workflow
+# ──────────────────────────────────────────────
+
+def build_moderation_workflow():
+    """
+    Construct and compile the LangGraph moderation pipeline.
+
+    Flow:
+        preprocess → fast_filter → risk_score
+            ├─ LOW/MEDIUM → decide
+            └─ HIGH → deep_analysis → decide
+
+    Returns:
+        Compiled LangGraph workflow.
+    """
+    graph = StateGraph(PipelineState)
+
+    # Add nodes
+    graph.add_node("preprocess", preprocess_node)
+    graph.add_node("fetch_history", fetch_user_history_node)
+    graph.add_node("fast_filter", fast_filter_node)
+    graph.add_node("risk_score", risk_score_node)
+    graph.add_node("deep_analysis", deep_analysis_node)
+    graph.add_node("decide", decision_node)
+
+    # Define edges
+    graph.set_entry_point("preprocess")
+
+    # After preprocess, run fast filter
+    graph.add_edge("preprocess", "fast_filter")
+
+    # After fast filter, compute risk score
+    graph.add_edge("fast_filter", "risk_score")
+
+    # Conditional routing based on risk level
+    graph.add_conditional_edges(
+        "risk_score",
+        route_by_risk,
+        {
+            "deep_analysis": "deep_analysis",
+            "decide": "decide",
+        },
+    )
+
+    # Deep analysis flows to decision
+    graph.add_edge("deep_analysis", "decide")
+
+    # Decision is the terminal node
+    graph.add_edge("decide", END)
+
+    # Compile
+    workflow = graph.compile()
+    logger.info("moderation_workflow_compiled")
+    return workflow
+
+
+# Global compiled workflow (initialized at startup)
+moderation_workflow = None
+
+
+def get_workflow():
+    """Get or create the compiled moderation workflow."""
+    global moderation_workflow
+    if moderation_workflow is None:
+        moderation_workflow = build_moderation_workflow()
+    return moderation_workflow

app/services/__init__.py

@@ -0,0 +1,2 @@
+# app/services/__init__.py
+"""External service integrations: MongoDB, Redis, Gemini."""

app/services/gemini_service.py

@@ -0,0 +1,247 @@
+# app/services/gemini_service.py
+# Gemini API integration via LangChain for deep analysis reasoning
+
+from app.config import get_settings
+from app.observability.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class GeminiService:
+    """
+    Google Gemini API client powered by LangChain.
+
+    Used in the deep analysis path for contextual reasoning
+    about flagged content. Replaces the old raw REST calls
+    with structured LangChain invocations for:
+    - Reliable structured output (JSON)
+    - Automatic retry and fallback
+    - LangSmith trace integration
+    """
+
+    def __init__(self):
+        self.settings = get_settings()
+        self.llm = None
+        self._current_key_idx = 0
+        self._initialized = False
+
+    def initialize(self) -> None:
+        """Initialize the LangChain Gemini client."""
+        keys = self.settings.gemini_keys_list
+        if not keys:
+            logger.warning("gemini_no_keys", reason="No API keys configured")
+            return
+
+        try:
+            from langchain_google_genai import ChatGoogleGenerativeAI
+
+            self.llm = ChatGoogleGenerativeAI(
+                model=self.settings.gemini_model,
+                google_api_key=keys[self._current_key_idx],
+                temperature=0.1,
+                max_output_tokens=1024,
+                convert_system_message_to_human=True,
+            )
+            self._initialized = True
+            logger.info("gemini_initialized", model=self.settings.gemini_model)
+        except Exception as e:
+            logger.error("gemini_init_failed", error=str(e))
+
+    def _rotate_key(self) -> bool:
+        """Rotate to the next API key. Returns False if all keys exhausted."""
+        keys = self.settings.gemini_keys_list
+        if not keys:
+            return False
+
+        self._current_key_idx = (self._current_key_idx + 1) % len(keys)
+        try:
+            from langchain_google_genai import ChatGoogleGenerativeAI
+
+            self.llm = ChatGoogleGenerativeAI(
+                model=self.settings.gemini_model,
+                google_api_key=keys[self._current_key_idx],
+                temperature=0.1,
+                max_output_tokens=1024,
+                convert_system_message_to_human=True,
+            )
+            logger.info("gemini_key_rotated", key_index=self._current_key_idx)
+            return True
+        except Exception:
+            return False
+
+    async def analyze_text(self, text: str, context: dict | None = None) -> dict:
+        """
+        Perform deep contextual analysis of flagged text.
+
+        Args:
+            text: The flagged text content.
+            context: Additional context (fast filter results, categories, etc.).
+
+        Returns:
+            Structured analysis with verdict, reasoning, and severity.
+        """
+        if not self._initialized:
+            return {"error": "Gemini not initialized", "is_confirmed": False}
+
+        from langchain_core.messages import SystemMessage, HumanMessage
+
+        system_prompt = """You are an expert content moderator specializing in cyberbullying detection.
+You are analyzing content that has been flagged as potentially harmful by automated filters.
+
+Your task: Provide a detailed, contextual analysis. Consider:
+- Intent and context (sarcasm, jokes, genuine threats)
+- Severity level (mild rudeness vs. serious threats)
+- Whether this constitutes cyberbullying
+- Impact on minors (under 18)
+
+Respond in this exact JSON format:
+{
+    "is_confirmed": true/false,
+    "severity": "low" | "medium" | "high" | "critical",
+    "categories": ["category1", "category2"],
+    "reasoning": "Detailed explanation of your analysis",
+    "recommended_action": "allow" | "warn" | "block" | "escalate",
+    "confidence": 0.0-1.0
+}"""
+
+        context_str = ""
+        if context:
+            context_str = f"\n\nPre-filter context: {context}"
+
+        human_message = f"""Analyze this flagged content:{context_str}
+
+Content: "{text}"
+
+Provide your JSON analysis:"""
+
+        return await self._invoke(system_prompt, human_message)
+
+    async def analyze_image(self, image_base64: str, context: dict | None = None) -> dict:
+        """
+        Perform deep analysis of a flagged image.
+
+        Args:
+            image_base64: Base64-encoded image.
+            context: Additional context from fast filter.
+
+        Returns:
+            Structured analysis dict.
+        """
+        if not self._initialized:
+            return {"error": "Gemini not initialized", "is_confirmed": False}
+
+        from langchain_core.messages import SystemMessage, HumanMessage
+
+        system_prompt = """You are an expert content moderator analyzing images for content harmful to minors.
+
+Analyze the image for:
+- Violence, gore, weapons
+- Nudity, sexual content
+- Drug/alcohol imagery
+- Self-harm or suicide content
+- Hate symbols, extremist content
+- Cyberbullying imagery (humiliating photos, etc.)
+
+Respond in this exact JSON format:
+{
+    "is_confirmed": true/false,
+    "severity": "low" | "medium" | "high" | "critical",
+    "categories": ["category1", "category2"],
+    "reasoning": "Description of what was found",
+    "recommended_action": "allow" | "warn" | "block" | "escalate",
+    "confidence": 0.0-1.0
+}"""
+
+        context_str = f"\nPre-filter flags: {context}" if context else ""
+
+        human_content = [
+            {"type": "text", "text": f"Analyze this flagged image:{context_str}\n\nProvide your JSON analysis:"},
+            {
+                "type": "image_url",
+                "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"},
+            },
+        ]
+
+        return await self._invoke_multimodal(system_prompt, human_content)
+
+    async def _invoke(self, system_prompt: str, human_message: str) -> dict:
+        """Invoke Gemini with retry on rate limits."""
+        from langchain_core.messages import SystemMessage, HumanMessage
+        import json
+
+        keys = self.settings.gemini_keys_list
+        attempts = max(len(keys), 1)
+
+        for attempt in range(attempts):
+            try:
+                messages = [
+                    SystemMessage(content=system_prompt),
+                    HumanMessage(content=human_message),
+                ]
+                response = await self.llm.ainvoke(messages)
+                return self._parse_response(response.content)
+
+            except Exception as e:
+                error_str = str(e)
+                if "429" in error_str or "quota" in error_str.lower():
+                    logger.warning("gemini_rate_limited", attempt=attempt)
+                    if not self._rotate_key():
+                        break
+                else:
+                    logger.error("gemini_invoke_failed", error=error_str)
+                    return {"error": error_str, "is_confirmed": False}
+
+        return {"error": "All Gemini API keys exhausted", "is_confirmed": False}
+
+    async def _invoke_multimodal(self, system_prompt: str, human_content: list) -> dict:
+        """Invoke Gemini with multimodal content."""
+        from langchain_core.messages import SystemMessage, HumanMessage
+
+        keys = self.settings.gemini_keys_list
+        attempts = max(len(keys), 1)
+
+        for attempt in range(attempts):
+            try:
+                messages = [
+                    SystemMessage(content=system_prompt),
+                    HumanMessage(content=human_content),
+                ]
+                response = await self.llm.ainvoke(messages)
+                return self._parse_response(response.content)
+
+            except Exception as e:
+                error_str = str(e)
+                if "429" in error_str or "quota" in error_str.lower():
+                    if not self._rotate_key():
+                        break
+                else:
+                    return {"error": error_str, "is_confirmed": False}
+
+        return {"error": "All Gemini API keys exhausted", "is_confirmed": False}
+
+    def _parse_response(self, text: str) -> dict:
+        """Parse JSON from Gemini response text."""
+        import json, re
+
+        try:
+            # Extract JSON block (handle markdown code fences)
+            json_match = re.search(r"\{[\s\S]*\}", text)
+            if json_match:
+                return json.loads(json_match.group())
+        except json.JSONDecodeError:
+            pass
+
+        logger.warning("gemini_parse_failed", raw_text=text[:200])
+        return {
+            "error": "Failed to parse Gemini response",
+            "is_confirmed": False,
+            "raw_response": text[:500],
+        }
+
+    @property
+    def is_initialized(self) -> bool:
+        return self._initialized
+
+
+# Global singleton
+gemini_service = GeminiService()