🚀 Auto-deploy backend from GitHub (8e7094e)

main.py

@@ -104,6 +104,9 @@ from routes.risk_router import router as risk_router
 from routes.tutor_checkin import router as tutor_checkin_router
 from routes.practice import router as practice_router
 from routes.ai_monitoring import router as ai_monitoring_router
+from routes.class_analytics_routes import router as class_analytics_router
+from routes.intervention_routes import router as intervention_router
+from routes.pipeline_routes import router as pipeline_router
 
 # Rate limiting (slowapi)
 try:
@@ -379,6 +382,19 @@ ROLE_POLICIES: Dict[str, Set[str]] = {
     "/api/predict-risk/batch": TEACHER_OR_ADMIN,
     "/api/learning-path": ALL_APP_ROLES,
     "/api/analytics/daily-insight": TEACHER_OR_ADMIN,
+    "/api/analytics/class/{class_id}": TEACHER_OR_ADMIN,
+    "/api/analytics/class/{class_id}/students": TEACHER_OR_ADMIN,
+    "/api/analytics/class/{class_id}/topics": TEACHER_OR_ADMIN,
+    "/api/analytics/class/{class_id}/refresh-insights": TEACHER_OR_ADMIN,
+    "/api/analytics/class/{class_id}/invalidate-cache": ALL_APP_ROLES,
+    "/api/intervention/generate": TEACHER_OR_ADMIN,
+    "/api/intervention/{student_id}": TEACHER_OR_ADMIN,
+    "/api/intervention/{student_id}/step/{step_number}/complete": ALL_APP_ROLES,
+    "/api/intervention/{student_id}/export-pdf": TEACHER_OR_ADMIN,
+    "/api/pipeline/event": ALL_APP_ROLES,
+    "/api/pipeline/profile/{student_id}": ALL_APP_ROLES,
+    "/api/pipeline/profile/{student_id}/recompute": TEACHER_OR_ADMIN,
+    "/api/pipeline/admin/backfill": ADMIN_ONLY,
     "/api/upload/class-records": TEACHER_OR_ADMIN,
     "/api/class-section/{class_section_id}": TEACHER_OR_ADMIN,
     "/api/upload/class-records/risk-refresh/recent": TEACHER_OR_ADMIN,
@@ -386,6 +402,7 @@ ROLE_POLICIES: Dict[str, Set[str]] = {
     "/api/import/student-accounts/commit": TEACHER_OR_ADMIN,
     "/api/admin/users": ADMIN_ONLY,
     "/api/admin/users/bulk-action": ADMIN_ONLY,
+    "/api/admin/school-analytics": ADMIN_ONLY,
     "/api/teacher/create-student-account": TEACHER_OR_ADMIN,
     "/api/upload/course-materials": TEACHER_OR_ADMIN,
     "/api/upload/course-materials/recent": TEACHER_OR_ADMIN,
@@ -1149,6 +1166,9 @@ app.include_router(risk_router)
 app.include_router(tutor_checkin_router)
 app.include_router(practice_router)
 app.include_router(ai_monitoring_router)
+app.include_router(class_analytics_router)
+app.include_router(intervention_router)
+app.include_router(pipeline_router)
 
 
 # ─── Global Exception Handler ─────────────────────────────────

routes/admin_routes.py

@@ -204,3 +204,112 @@ async def delete_uploaded_file(
     except Exception as e:
         logger.error(f"Failed to delete file: {e}")
         raise HTTPException(status_code=500, detail=f"Failed to delete file: {e}")
+
+
+
+# ─── School-Wide Analytics ─────────────────────────────────────────────────
+
+@router.get("/school-analytics")
+def get_school_analytics(request: Request):
+    """School-wide WRI aggregation for admin dashboard."""
+    require_admin(request)
+
+    try:
+        import firebase_admin
+        from firebase_admin import firestore as fs
+        db = fs.client()
+    except Exception:
+        raise HTTPException(status_code=503, detail="Firestore unavailable")
+
+    from datetime import datetime, timezone, timedelta
+    from collections import defaultdict
+
+    # Read all student_profiles
+    profiles = list(db.collection("student_profiles").stream())
+
+    total = len(profiles)
+    if total == 0:
+        # Fallback: read from managedStudents
+        profiles = list(db.collection("managedStudents").stream())
+        total = len(profiles)
+
+    wri_dist = {"safe": 0, "watch": 0, "intervene": 0, "critical": 0, "at_risk": 0, "pending_assessment": 0}
+    wri_values = []
+    grade_wri: dict = defaultdict(list)
+    class_wri: dict = defaultdict(list)
+    weak_topics_counter: dict = defaultdict(int)
+    recent_escalations = []
+    pending_count = 0
+    now = datetime.now(timezone.utc)
+
+    for doc in profiles:
+        data = doc.to_dict()
+        status = data.get("risk_status") or data.get("riskStatus") or "pending_assessment"
+
+        # Normalize status
+        if status in wri_dist:
+            wri_dist[status] += 1
+        else:
+            wri_dist["pending_assessment"] += 1
+
+        wri = data.get("wri")
+        if wri is not None:
+            wri_values.append(wri)
+            grade = str(data.get("grade_level") or data.get("gradeLevel") or data.get("grade", "?"))
+            grade_wri[grade].append(wri)
+            class_id = data.get("class_id") or data.get("classroomId") or ""
+            if class_id:
+                class_wri[class_id].append(wri)
+
+        if status == "pending_assessment" or data.get("diagnosticScore") is None:
+            pending_count += 1
+
+        # Weak topics
+        weak = data.get("quiz_performance", {}).get("lowest_accuracy_topics") or []
+        if not weak:
+            weak = [data.get("weakestTopic")] if data.get("weakestTopic") and data.get("weakestTopic") != "N/A" else []
+        for t in weak[:2]:
+            if t:
+                weak_topics_counter[t] += 1
+
+        # Recent escalations (last 24h)
+        updated = data.get("wri_updated_at") or data.get("riskUpdatedAt")
+        if updated and status in ("critical", "at_risk"):
+            try:
+                if hasattr(updated, "seconds"):
+                    dt = datetime.fromtimestamp(updated.seconds, tz=timezone.utc)
+                elif isinstance(updated, str):
+                    dt = datetime.fromisoformat(updated.replace("Z", "+00:00"))
+                else:
+                    dt = None
+                if dt and (now - dt).total_seconds() < 86400:
+                    recent_escalations.append({
+                        "student_id": doc.id,
+                        "student_name": data.get("display_name") or data.get("name", "Unknown"),
+                        "risk_status": status,
+                        "wri": wri,
+                        "teacher_id": data.get("teacher_id") or data.get("teacherId", ""),
+                        "escalated_at": dt.isoformat(),
+                    })
+            except Exception:
+                pass
+
+    school_avg = round(sum(wri_values) / len(wri_values), 1) if wri_values else 0.0
+    avg_by_grade = {g: round(sum(v) / len(v), 1) for g, v in grade_wri.items() if v}
+    classes_ranked = sorted(
+        [{"class_id": c, "avg_wri": round(sum(v) / len(v), 1), "student_count": len(v)} for c, v in class_wri.items() if v],
+        key=lambda x: x["avg_wri"]
+    )
+    top_weak = sorted(weak_topics_counter.items(), key=lambda x: -x[1])[:10]
+
+    return {
+        "total_students": total,
+        "wri_distribution": wri_dist,
+        "school_avg_wri": school_avg,
+        "avg_wri_by_grade": avg_by_grade,
+        "classes_ranked": classes_ranked[:20],
+        "top_weak_topics_school": [t for t, _ in top_weak],
+        "recent_escalations": recent_escalations[:20],
+        "pending_assessment_count": pending_count,
+        "generated_at": now.isoformat(),
+    }

routes/class_analytics_routes.py

@@ -0,0 +1,110 @@
+"""Class Analytics API routes."""
+
+import time
+import logging
+from typing import Optional
+from fastapi import APIRouter, HTTPException, Request
+
+logger = logging.getLogger("mathpulse.class_analytics_routes")
+
+router = APIRouter(prefix="/api/analytics/class", tags=["class-analytics"])
+
+# Rate limit: 1 refresh per 5 min per class
+_refresh_timestamps: dict[str, float] = {}
+
+
+def _require_teacher(request: Request):
+    user = getattr(request.state, "user", None)
+    if not user:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    if user.role not in ("teacher", "admin"):
+        raise HTTPException(status_code=403, detail="Teacher or admin access required")
+    return user
+
+
+@router.get("/{class_id}")
+async def get_class_analytics(class_id: str, request: Request, refresh: bool = False):
+    """Get full class analytics report. Cached for 30 min unless refresh=true."""
+    user = _require_teacher(request)
+
+    from services.class_analytics_engine import get_class_analytics_engine
+    engine = get_class_analytics_engine()
+    report = await engine.get_class_analytics(class_id, user.uid, force_refresh=refresh)
+    return report.model_dump()
+
+
+@router.get("/{class_id}/students")
+async def get_class_students(
+    class_id: str, request: Request, filter: Optional[str] = "all"
+):
+    """Get student summaries for a class with optional filtering."""
+    user = _require_teacher(request)
+
+    from services.class_analytics_engine import get_class_analytics_engine
+    engine = get_class_analytics_engine()
+    report = await engine.get_class_analytics(class_id, user.uid)
+
+    students = report.students
+    if filter == "top_performers":
+        students = sorted(
+            [s for s in students if s.quiz_attempt_count > 0],
+            key=lambda s: s.avg_score,
+            reverse=True,
+        )[:10]
+    elif filter == "needs_attention":
+        students = sorted(
+            [s for s in students if s.risk_level in ("High Risk", "Critical", "Unassessed")],
+            key=lambda s: s.avg_score,
+        )
+
+    return [s.model_dump() for s in students]
+
+
+@router.get("/{class_id}/topics")
+async def get_class_topics(class_id: str, request: Request):
+    """Get topic performance sorted by accuracy (worst first)."""
+    user = _require_teacher(request)
+
+    from services.class_analytics_engine import get_class_analytics_engine
+    engine = get_class_analytics_engine()
+    report = await engine.get_class_analytics(class_id, user.uid)
+
+    topics = report.insights.topic_performance if report.insights else []
+    return [t.model_dump() for t in topics]
+
+
+@router.post("/{class_id}/refresh-insights")
+async def refresh_class_insights(class_id: str, request: Request):
+    """Force regeneration of AI insights. Rate limited: 1 per 5 min per class."""
+    user = _require_teacher(request)
+
+    last_refresh = _refresh_timestamps.get(class_id, 0)
+    if time.time() - last_refresh < 300:
+        raise HTTPException(
+            status_code=429,
+            detail="Insights can only be refreshed once every 5 minutes.",
+        )
+
+    from services.class_analytics_engine import get_class_analytics_engine
+    engine = get_class_analytics_engine()
+    engine.invalidate_cache(class_id)
+    report = await engine.get_class_analytics(class_id, user.uid, force_refresh=True)
+    _refresh_timestamps[class_id] = time.time()
+
+    if report.insights:
+        return report.insights.model_dump()
+    return {"error": "Failed to generate insights"}
+
+
+@router.post("/{class_id}/invalidate-cache")
+async def invalidate_class_cache(class_id: str, request: Request):
+    """Invalidate cached analytics for a class (called after quiz completion)."""
+    # Allow any authenticated user (student completing quiz triggers this)
+    user = getattr(request.state, "user", None)
+    if not user:
+        raise HTTPException(status_code=401, detail="Authentication required")
+
+    from services.class_analytics_engine import get_class_analytics_engine
+    engine = get_class_analytics_engine()
+    engine.invalidate_cache(class_id)
+    return {"status": "cache_invalidated", "class_id": class_id}

routes/intervention_routes.py

@@ -0,0 +1,84 @@
+"""Intervention API routes."""
+
+import logging
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+from typing import Optional
+
+logger = logging.getLogger("mathpulse.intervention_routes")
+
+router = APIRouter(prefix="/api/intervention", tags=["intervention"])
+
+
+class GenerateRequest(BaseModel):
+    student_id: str
+
+
+class CompleteStepRequest(BaseModel):
+    score: float = 0.0
+    time_spent_minutes: int = 0
+
+
+def _require_auth(request: Request):
+    user = getattr(request.state, "user", None)
+    if not user:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return user
+
+
+@router.post("/generate")
+async def generate_intervention(body: GenerateRequest, request: Request):
+    """Generate a full intervention plan for a student."""
+    _require_auth(request)
+
+    from services.intervention_engine import get_intervention_engine
+    engine = get_intervention_engine()
+    plan = await engine.generate_full_intervention(body.student_id, force=True)
+    return plan.model_dump()
+
+
+@router.get("/{student_id}")
+async def get_intervention(student_id: str, request: Request):
+    """Get latest intervention plan. Returns persisted doc if exists, otherwise generates."""
+    user = _require_auth(request)
+
+    # Try to return persisted plan first (fast, no AI cost)
+    try:
+        from firebase_admin import firestore as fs
+        db = fs.client()
+        doc = db.collection("intervention_plans").document(student_id).get()
+        if doc.exists:
+            return doc.to_dict()
+    except Exception:
+        pass
+
+    # No persisted plan — generate one
+    from services.intervention_engine import get_intervention_engine
+    engine = get_intervention_engine()
+    plan = await engine.generate_full_intervention(student_id, force=False)
+    return plan.model_dump()
+
+
+@router.post("/{student_id}/step/{step_number}/complete")
+async def complete_step(student_id: str, step_number: int, body: CompleteStepRequest, request: Request):
+    """Mark a learning step as completed."""
+    _require_auth(request)
+
+    from services.intervention_engine import get_intervention_engine
+    engine = get_intervention_engine()
+    result = await engine.complete_step(student_id, step_number, body.score, body.time_spent_minutes)
+
+    if "error" in result:
+        raise HTTPException(status_code=400, detail=result["error"])
+    return result
+
+
+@router.get("/{student_id}/export-pdf")
+async def export_pdf_data(student_id: str, request: Request):
+    """Get all data needed for PDF export."""
+    _require_auth(request)
+
+    from services.intervention_engine import get_intervention_engine
+    engine = get_intervention_engine()
+    plan = await engine.generate_full_intervention(student_id, force=False)
+    return plan.model_dump()

routes/pipeline_routes.py

@@ -0,0 +1,131 @@
+"""Student Intelligence Pipeline API routes."""
+
+import logging
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request
+from pydantic import BaseModel
+from typing import Any, Dict, Literal
+
+logger = logging.getLogger("mathpulse.pipeline_routes")
+
+router = APIRouter(prefix="/api/pipeline", tags=["pipeline"])
+
+
+class PipelineEventPayload(BaseModel):
+    student_id: str
+    event_type: Literal["diagnostic", "quiz", "battle", "lesson", "module", "session"]
+    event_data: Dict[str, Any] = {}
+    occurred_at: str
+    class_id: str = ""
+    teacher_id: str = ""
+
+
+def _require_auth(request: Request):
+    user = getattr(request.state, "user", None)
+    if not user:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return user
+
+
+async def _run_pipeline(payload: PipelineEventPayload):
+    """Background task to run the pipeline."""
+    try:
+        from services.student_intelligence_pipeline import get_pipeline, StudentActivityEvent
+        pipeline = get_pipeline()
+        event = StudentActivityEvent(**payload.model_dump())
+        await pipeline.process_event(event)
+    except Exception as e:
+        logger.error(f"Pipeline background task failed: {e}", exc_info=True)
+
+
+@router.post("/event", status_code=202)
+async def receive_event(payload: PipelineEventPayload, background_tasks: BackgroundTasks, request: Request):
+    """Universal intake endpoint. Returns 202 immediately, processes async."""
+    user = _require_auth(request)
+    # Students can only emit events for themselves
+    if user.role == "student" and payload.student_id != user.uid:
+        raise HTTPException(status_code=403, detail="Students can only emit events for their own ID")
+    background_tasks.add_task(_run_pipeline, payload)
+    return {"status": "accepted", "student_id": payload.student_id}
+
+
+@router.get("/profile/{student_id}")
+async def get_profile(student_id: str, request: Request):
+    """Get full student profile."""
+    _require_auth(request)
+
+    _firebase_firestore = None
+    try:
+        from firebase_admin import firestore as ff
+        _firebase_firestore = ff
+    except Exception:
+        raise HTTPException(status_code=503, detail="Firestore unavailable")
+
+    db = _firebase_firestore.client()
+    doc = db.collection("student_profiles").document(student_id).get()
+    if not doc.exists:
+        raise HTTPException(status_code=404, detail="Profile not found")
+    return doc.to_dict()
+
+
+@router.post("/profile/{student_id}/recompute")
+async def recompute_profile(student_id: str, background_tasks: BackgroundTasks, request: Request):
+    """Force full profile rebuild from raw Firestore data."""
+    user = _require_auth(request)
+    if user.role not in ("teacher", "admin"):
+        raise HTTPException(status_code=403, detail="Teacher or admin required")
+
+    async def _recompute():
+        from services.student_intelligence_pipeline import get_pipeline, StudentActivityEvent
+        pipeline = get_pipeline()
+        # Trigger a synthetic diagnostic event to force full recompute
+        event = StudentActivityEvent(
+            student_id=student_id,
+            event_type="session",
+            event_data={"event": "force_recompute"},
+            occurred_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc).isoformat(),
+            class_id="",
+            teacher_id=user.uid,
+        )
+        await pipeline.process_event(event)
+
+    background_tasks.add_task(_recompute)
+    return {"status": "recompute_queued", "student_id": student_id}
+
+
+@router.post("/admin/backfill", status_code=202)
+async def backfill_all(background_tasks: BackgroundTasks, request: Request):
+    """One-time migration: rebuild student_profiles for ALL existing students."""
+    user = _require_auth(request)
+    if user.role != "admin":
+        raise HTTPException(status_code=403, detail="Admin only")
+
+    async def _backfill():
+        try:
+            from services.student_intelligence_pipeline import get_pipeline, StudentActivityEvent
+            from firebase_admin import firestore as ff
+            db = ff.client()
+            pipeline = get_pipeline()
+
+            students = db.collection("managedStudents").stream()
+            count = 0
+            for student_doc in students:
+                sid = student_doc.id
+                data = student_doc.to_dict()
+                event = StudentActivityEvent(
+                    student_id=sid,
+                    event_type="session",
+                    event_data={"event": "backfill"},
+                    occurred_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc).isoformat(),
+                    class_id=data.get("classroomId", ""),
+                    teacher_id=data.get("teacherId", ""),
+                )
+                await pipeline.process_event(event)
+                count += 1
+                if count % 10 == 0:
+                    logger.info(f"Backfilled {count} students")
+            logger.info(f"Backfill complete: {count} students processed")
+        except Exception as e:
+            logger.error(f"Backfill failed: {e}", exc_info=True)
+
+    background_tasks.add_task(_backfill)
+    return {"status": "backfill_started"}

scripts/backfill_student_profiles.py

@@ -0,0 +1,111 @@
+"""
+MathPulse AI — Backfill Student Profiles
+
+One-time migration script to build student_profiles for ALL existing students
+who already have data in Firestore but no unified profile.
+
+Usage:
+  cd backend
+  python -m scripts.backfill_student_profiles
+"""
+
+import asyncio
+import logging
+import os
+import sys
+
+# Add backend to path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+logger = logging.getLogger("backfill")
+
+
+async def backfill_all_profiles():
+    """Rebuild student_profiles for all existing managedStudents."""
+    # Initialize Firebase
+    import firebase_admin
+    from firebase_admin import firestore
+
+    if not firebase_admin._apps:
+        sa_path = os.environ.get("FIREBASE_SERVICE_ACCOUNT_FILE")
+        sa_json = os.environ.get("FIREBASE_SERVICE_ACCOUNT_JSON")
+        if sa_path and os.path.exists(sa_path):
+            from firebase_admin import credentials
+            cred = credentials.Certificate(sa_path)
+            firebase_admin.initialize_app(cred)
+        elif sa_json:
+            import json
+            from firebase_admin import credentials
+            cred = credentials.Certificate(json.loads(sa_json))
+            firebase_admin.initialize_app(cred)
+        else:
+            firebase_admin.initialize_app()
+
+    db = firestore.client()
+
+    from services.student_intelligence_pipeline import get_pipeline, StudentActivityEvent
+    from datetime import datetime, timezone
+
+    pipeline = get_pipeline()
+
+    # Fetch all managed students
+    logger.info("Fetching all managedStudents...")
+    students = list(db.collection("managedStudents").stream())
+    total = len(students)
+    logger.info(f"Found {total} students to backfill")
+
+    success = 0
+    errors = 0
+
+    for i, student_doc in enumerate(students):
+        sid = student_doc.id
+        data = student_doc.to_dict()
+
+        try:
+            # Create a synthetic event to trigger full profile build
+            event = StudentActivityEvent(
+                student_id=sid,
+                event_type="session",
+                event_data={"event": "backfill", "source": "migration_script"},
+                occurred_at=datetime.now(timezone.utc).isoformat(),
+                class_id=data.get("classroomId", ""),
+                teacher_id=data.get("teacherId", ""),
+            )
+
+            # Set basic profile fields from managed student data
+            profile_ref = db.collection("student_profiles").document(sid)
+            profile_doc = profile_ref.get()
+            if not profile_doc.exists:
+                # Pre-seed with identity data
+                profile_ref.set({
+                    "student_id": sid,
+                    "display_name": data.get("name", ""),
+                    "grade_level": data.get("gradeLevel", data.get("grade", "")),
+                    "section": data.get("section", ""),
+                    "class_id": data.get("classroomId", ""),
+                    "teacher_id": data.get("teacherId", ""),
+                    "diagnostic_score": data.get("diagnosticScore"),
+                    "external_grades_avg": data.get("externalGradesAvg"),
+                    "wri": data.get("wri"),
+                    "risk_status": data.get("riskStatus", "pending_assessment"),
+                    "wri_weights": data.get("weights", {"w1": 0.30, "w2": 0.40, "w3": 0.30}),
+                    "profile_version": 0,
+                }, merge=True)
+
+            # Run pipeline to compute P and update everything
+            await pipeline.process_event(event)
+            success += 1
+
+        except Exception as e:
+            logger.error(f"Error backfilling {sid}: {e}")
+            errors += 1
+
+        if (i + 1) % 10 == 0:
+            logger.info(f"Progress: {i + 1}/{total} (success={success}, errors={errors})")
+
+    logger.info(f"Backfill complete: {success} success, {errors} errors out of {total} total")
+
+
+if __name__ == "__main__":
+    asyncio.run(backfill_all_profiles())

services/class_analytics_engine.py

@@ -0,0 +1,555 @@
+"""
+MathPulse AI — Class Analytics Engine
+
+Fetches real quiz data from Firestore, computes per-student and class-level
+metrics, generates AI insights via DeepSeek, and caches results.
+"""
+
+import asyncio
+import json
+import logging
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger("mathpulse.class_analytics")
+
+# ─── Firestore helper ──────────────────────────────────────────────────────
+
+_firebase_firestore = None
+
+
+def _get_firestore_client():
+    global _firebase_firestore
+    if _firebase_firestore is None:
+        try:
+            from firebase_admin import firestore as ff
+            _firebase_firestore = ff
+        except Exception:
+            return None
+    try:
+        return _firebase_firestore.client()
+    except Exception:
+        return None
+
+
+# ─── Models ────────────────────────────────────────────────────────────────
+
+class StudentAnalyticsSummary(BaseModel):
+    student_id: str
+    student_name: str
+    avatar_url: Optional[str] = None
+    grade_level: str = ""
+    section: str = ""
+    avg_score: float = 0.0
+    quiz_attempt_count: int = 0
+    last_active: Optional[str] = None
+    risk_level: Literal["Low Risk", "Medium Risk", "High Risk", "Critical", "Unassessed"] = "Unassessed"
+    engagement_level: Literal["Low", "Medium", "High"] = "Low"
+    weakest_topic: Optional[str] = None
+    accuracy_by_topic: Dict[str, float] = Field(default_factory=dict)
+    completion_rate: float = 0.0
+
+
+class TopicPerformance(BaseModel):
+    topic: str
+    class_accuracy: float = 0.0
+    struggling_count: int = 0
+    mastered_count: int = 0
+
+
+class ClassInsights(BaseModel):
+    class_id: str
+    generated_at: str = ""
+    class_summary: str = ""
+    top_weak_topics: List[str] = Field(default_factory=list)
+    recommended_actions: List[str] = Field(default_factory=list)
+    class_strengths: str = ""
+    risk_distribution: Dict[str, int] = Field(default_factory=dict)
+    topic_performance: List[TopicPerformance] = Field(default_factory=list)
+
+
+class ClassAnalyticsReport(BaseModel):
+    class_id: str
+    class_name: str = ""
+    grade_level: str = ""
+    section: str = ""
+    teacher_id: str = ""
+    student_count: int = 0
+    class_average: float = 0.0
+    completion_rate: float = 0.0
+    participation_rate: float = 0.0
+    attention_count: int = 0
+    students: List[StudentAnalyticsSummary] = Field(default_factory=list)
+    insights: Optional[ClassInsights] = None
+    generated_at: str = ""
+
+
+# ─── Risk & Engagement Classification ─────────────────────────────────────
+
+def classify_risk(avg_score: float, quiz_count: int, days_since_active: Optional[int]) -> str:
+    """Map to WRI status bands. When no WRI is available, estimate from avg_score."""
+    if quiz_count == 0:
+        return "pending_assessment"
+    # Approximate WRI bands from avg_score (actual WRI uses D, G, P weights)
+    if avg_score >= 88:
+        return "safe"
+    if avg_score >= 80:
+        return "watch"
+    if avg_score >= 75:
+        return "intervene"
+    if avg_score >= 68:
+        return "critical"
+    return "at_risk"
+
+
+def classify_engagement(days_since_active: Optional[int], recent_quiz_count: int) -> str:
+    if days_since_active is not None and days_since_active <= 2 and recent_quiz_count >= 5:
+        return "High"
+    if days_since_active is not None and days_since_active <= 7:
+        return "Medium"
+    return "Low"
+
+
+# ─── Engine ────────────────────────────────────────────────────────────────
+
+class ClassAnalyticsEngine:
+    """Computes class analytics from Firestore data."""
+
+    def __init__(self):
+        self._cache: Dict[str, tuple] = {}  # class_id -> (report, timestamp)
+        self._cache_ttl = 1800  # 30 min
+
+    async def get_class_analytics(
+        self, class_id: str, teacher_id: str, force_refresh: bool = False
+    ) -> ClassAnalyticsReport:
+        # Check cache
+        if not force_refresh and class_id in self._cache:
+            report, cached_at = self._cache[class_id]
+            if time.time() - cached_at < self._cache_ttl:
+                return report
+
+        db = _get_firestore_client()
+        if not db:
+            logger.error("Firestore client unavailable")
+            return ClassAnalyticsReport(class_id=class_id, teacher_id=teacher_id, generated_at=_now_iso())
+
+        # Fast path: try reading from denormalized student_summaries (written by pipeline)
+        summaries_snap = list(db.collection("classes").document(class_id).collection("student_summaries").stream())
+        use_fast_path = len(summaries_snap) > 0 and not force_refresh
+
+        # Fetch class info
+        class_doc = db.collection("classrooms").document(class_id).get()
+        class_data = class_doc.to_dict() if class_doc.exists else {}
+
+        class_name = class_data.get("name", "")
+        grade_level = class_data.get("gradeLevel", class_data.get("grade", ""))
+        section = class_data.get("section", "")
+
+        # Fetch students in this class from managedStudents
+        students_query = db.collection("managedStudents").where("classroomId", "==", class_id).stream()
+        managed_students = []
+        for doc_snap in students_query:
+            managed_students.append({"id": doc_snap.id, **doc_snap.to_dict()})
+
+        # Also try classSectionId match
+        if not managed_students:
+            class_section_id = class_data.get("classSectionId", class_id)
+            students_query2 = db.collection("managedStudents").where("classSectionId", "==", class_section_id).stream()
+            for doc_snap in students_query2:
+                managed_students.append({"id": doc_snap.id, **doc_snap.to_dict()})
+
+        # Fetch quiz data for each student from progress collection
+        student_summaries = await self._build_student_summaries(db, managed_students)
+
+        # Compute class-level metrics
+        assessed_students = [s for s in student_summaries if s.quiz_attempt_count > 0]
+        student_count = len(student_summaries)
+
+        class_average = 0.0
+        if assessed_students:
+            class_average = sum(s.avg_score for s in assessed_students) / len(assessed_students)
+
+        completion_rate = 0.0
+        if student_count > 0:
+            completion_rate = (len(assessed_students) / student_count) * 100
+
+        # Participation = active in last 7 days
+        now = datetime.now(timezone.utc)
+        active_count = sum(1 for s in student_summaries if s.last_active and _days_since(s.last_active) <= 7)
+        participation_rate = (active_count / student_count * 100) if student_count > 0 else 0.0
+
+        # Attention = intervene + critical + at_risk
+        attention_count = sum(1 for s in student_summaries if s.risk_level in ("intervene", "critical", "at_risk"))
+
+        # Topic performance
+        topic_perf = self._compute_topic_performance(student_summaries)
+
+        # Risk distribution
+        risk_dist = {"safe": 0, "watch": 0, "intervene": 0, "critical": 0, "at_risk": 0, "pending_assessment": 0}
+        for s in student_summaries:
+            # Prefer stored WRI status from managedStudents if available
+            stored_status = None
+            try:
+                ms_doc = db.collection("managedStudents").document(s.student_id).get()
+                if ms_doc.exists:
+                    stored_status = ms_doc.to_dict().get("riskStatus")
+            except Exception:
+                pass
+            status = stored_status if stored_status in risk_dist else s.risk_level
+            risk_dist[status] = risk_dist.get(status, 0) + 1
+
+        # Generate AI insights
+        insights = await self._generate_insights(
+            class_id=class_id,
+            class_name=class_name,
+            grade_level=grade_level,
+            section=section,
+            student_count=student_count,
+            class_average=class_average,
+            completion_rate=completion_rate,
+            participation_rate=participation_rate,
+            risk_dist=risk_dist,
+            topic_perf=topic_perf,
+        )
+
+        report = ClassAnalyticsReport(
+            class_id=class_id,
+            class_name=class_name,
+            grade_level=grade_level,
+            section=section,
+            teacher_id=teacher_id,
+            student_count=student_count,
+            class_average=round(class_average, 1),
+            completion_rate=round(completion_rate, 1),
+            participation_rate=round(participation_rate, 1),
+            attention_count=attention_count,
+            students=student_summaries,
+            insights=insights,
+            generated_at=_now_iso(),
+        )
+
+        # Cache
+        self._cache[class_id] = (report, time.time())
+
+        # Persist to Firestore
+        try:
+            db.collection("class_analytics").document(class_id).set(
+                {**report.model_dump(), "cached_at": _now_iso()},
+                merge=True,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to persist analytics cache: {e}")
+
+        return report
+
+    async def _build_student_summaries(
+        self, db: Any, managed_students: List[Dict]
+    ) -> List[StudentAnalyticsSummary]:
+        summaries = []
+        now = datetime.now(timezone.utc)
+
+        for student in managed_students:
+            student_id = student.get("id", "")
+            student_name = student.get("name", "Unknown")
+            avatar = student.get("avatar", "")
+
+            # Try to fetch quiz data from progress/{student_id}
+            quiz_attempts = []
+            try:
+                # Try by student ID first
+                progress_doc = db.collection("progress").document(student_id).get()
+                if progress_doc.exists:
+                    pdata = progress_doc.to_dict()
+                    quiz_attempts = pdata.get("quizAttempts", [])
+
+                # Also try by LRN if available
+                if not quiz_attempts and student.get("lrn"):
+                    progress_doc2 = db.collection("progress").document(student["lrn"]).get()
+                    if progress_doc2.exists:
+                        pdata2 = progress_doc2.to_dict()
+                        quiz_attempts = pdata2.get("quizAttempts", [])
+
+                # Also try accountUid
+                if not quiz_attempts and student.get("accountUid"):
+                    progress_doc3 = db.collection("progress").document(student["accountUid"]).get()
+                    if progress_doc3.exists:
+                        pdata3 = progress_doc3.to_dict()
+                        quiz_attempts = pdata3.get("quizAttempts", [])
+
+                # Also check practice_results subcollection
+                if not quiz_attempts and student.get("accountUid"):
+                    practice_sessions = (
+                        db.collection("practice_results")
+                        .document(student["accountUid"])
+                        .collection("sessions")
+                        .order_by("submitted_at", direction="DESCENDING")
+                        .limit(20)
+                        .stream()
+                    )
+                    for sess in practice_sessions:
+                        sd = sess.to_dict()
+                        quiz_attempts.append({
+                            "quizId": sd.get("session_id", ""),
+                            "score": sd.get("score_percent", 0),
+                            "completedAt": sd.get("submitted_at"),
+                            "answers": sd.get("per_question_feedback", []),
+                        })
+            except Exception as e:
+                logger.debug(f"Error fetching progress for {student_id}: {e}")
+
+            # Compute metrics from quiz attempts
+            quiz_count = len(quiz_attempts)
+            avg_score = 0.0
+            accuracy_by_topic: Dict[str, List[float]] = {}
+
+            if quiz_count > 0:
+                scores = [float(q.get("score", 0)) for q in quiz_attempts]
+                avg_score = sum(scores) / len(scores)
+
+                # Extract per-question topic accuracy if available
+                for attempt in quiz_attempts:
+                    answers = attempt.get("answers", [])
+                    quiz_id = attempt.get("quizId", "")
+                    # Use quizId as topic proxy if no per-question topic
+                    topic = _extract_topic_from_quiz_id(quiz_id)
+                    if topic:
+                        if topic not in accuracy_by_topic:
+                            accuracy_by_topic[topic] = []
+                        accuracy_by_topic[topic].append(float(attempt.get("score", 0)))
+
+            # Compute topic averages
+            topic_avgs = {t: sum(scores) / len(scores) for t, scores in accuracy_by_topic.items() if scores}
+            weakest_topic = min(topic_avgs, key=topic_avgs.get) if topic_avgs else student.get("weakestTopic")
+            if weakest_topic == "N/A":
+                weakest_topic = None
+
+            # Last active
+            last_active_ts = student.get("lastActive")
+            last_active_str = None
+            days_since_active = None
+            if last_active_ts:
+                try:
+                    if hasattr(last_active_ts, "seconds"):
+                        last_dt = datetime.fromtimestamp(last_active_ts.seconds, tz=timezone.utc)
+                    else:
+                        last_dt = last_active_ts
+                    last_active_str = last_dt.isoformat()
+                    days_since_active = (now - last_dt).days
+                except Exception:
+                    pass
+
+            # Recent quiz count (last 14 days)
+            recent_quiz_count = 0
+            for q in quiz_attempts:
+                completed = q.get("completedAt")
+                if completed:
+                    try:
+                        if hasattr(completed, "seconds"):
+                            q_dt = datetime.fromtimestamp(completed.seconds, tz=timezone.utc)
+                        else:
+                            q_dt = completed if isinstance(completed, datetime) else datetime.now(timezone.utc)
+                        if (now - q_dt).days <= 14:
+                            recent_quiz_count += 1
+                    except Exception:
+                        pass
+
+            risk_level = classify_risk(avg_score, quiz_count, days_since_active)
+            engagement = classify_engagement(days_since_active, recent_quiz_count)
+
+            summaries.append(StudentAnalyticsSummary(
+                student_id=student_id,
+                student_name=student_name,
+                avatar_url=avatar or None,
+                grade_level=student.get("gradeLevel", student.get("grade", "")),
+                section=student.get("section", ""),
+                avg_score=round(avg_score, 1),
+                quiz_attempt_count=quiz_count,
+                last_active=last_active_str,
+                risk_level=risk_level,
+                engagement_level=engagement,
+                weakest_topic=weakest_topic,
+                accuracy_by_topic=topic_avgs,
+                completion_rate=min(quiz_count / 5 * 100, 100) if quiz_count > 0 else 0.0,
+            ))
+
+        return summaries
+
+    def _compute_topic_performance(self, students: List[StudentAnalyticsSummary]) -> List[TopicPerformance]:
+        topic_data: Dict[str, Dict] = {}
+
+        for s in students:
+            for topic, accuracy in s.accuracy_by_topic.items():
+                if topic not in topic_data:
+                    topic_data[topic] = {"scores": [], "struggling": 0, "mastered": 0}
+                topic_data[topic]["scores"].append(accuracy)
+                if accuracy < 60:
+                    topic_data[topic]["struggling"] += 1
+                if accuracy >= 80:
+                    topic_data[topic]["mastered"] += 1
+
+        # Also include weakest_topic from students without per-topic data
+        for s in students:
+            if s.weakest_topic and s.weakest_topic not in topic_data and s.quiz_attempt_count > 0:
+                topic_data[s.weakest_topic] = {
+                    "scores": [s.avg_score],
+                    "struggling": 1 if s.avg_score < 60 else 0,
+                    "mastered": 0,
+                }
+
+        result = []
+        for topic, data in topic_data.items():
+            if not data["scores"]:
+                continue
+            result.append(TopicPerformance(
+                topic=topic,
+                class_accuracy=round(sum(data["scores"]) / len(data["scores"]), 1),
+                struggling_count=data["struggling"],
+                mastered_count=data["mastered"],
+            ))
+
+        return sorted(result, key=lambda t: t.class_accuracy)[:8]
+
+    async def _generate_insights(
+        self,
+        class_id: str,
+        class_name: str,
+        grade_level: str,
+        section: str,
+        student_count: int,
+        class_average: float,
+        completion_rate: float,
+        participation_rate: float,
+        risk_dist: Dict[str, int],
+        topic_perf: List[TopicPerformance],
+    ) -> ClassInsights:
+        # Format topic performance for prompt
+        topic_lines = "\n".join(
+            f"  - {t.topic}: {t.class_accuracy}% (struggling: {t.struggling_count})"
+            for t in topic_perf[:6]
+        ) or "  No topic data available yet."
+
+        weak_topics = [t.topic for t in topic_perf[:3]] if topic_perf else []
+
+        prompt = f"""You are MathPulse AI analyzing a class's performance data for a Filipino K-12 teacher.
+
+Class: Grade {grade_level} - {section} ({class_name})
+Student Count: {student_count}
+Class Average Score: {class_average:.1f}%
+Completion Rate: {completion_rate:.1f}%
+Participation Rate: {participation_rate:.1f}%
+
+Risk Distribution:
+- Critical: {risk_dist.get('Critical', 0)} students
+- High Risk: {risk_dist.get('High Risk', 0)} students
+- Medium Risk: {risk_dist.get('Medium Risk', 0)} students
+- Low Risk: {risk_dist.get('Low Risk', 0)} students
+- Unassessed: {risk_dist.get('Unassessed', 0)} students
+
+Topic Performance (class accuracy):
+{topic_lines}
+
+Top Weakest Topics: {', '.join(weak_topics) if weak_topics else 'None identified yet'}
+
+Generate a JSON response with these exact keys:
+{{
+  "class_summary": "2-3 sentence overview of class performance. Be honest but constructive.",
+  "class_strengths": "1 sentence on what the class is doing well.",
+  "top_weak_topics": ["topic1", "topic2", "topic3"],
+  "recommended_actions": [
+    "Specific action 1 (max 20 words)",
+    "Specific action 2",
+    "Specific action 3"
+  ]
+}}
+
+Be specific to Filipino K-12 DepEd context. If data is limited, acknowledge it and suggest next steps."""
+
+        try:
+            from services.ai_client import get_deepseek_client, CHAT_MODEL
+
+            client = get_deepseek_client()
+            response = client.chat.completions.create(
+                model=CHAT_MODEL,
+                messages=[
+                    {"role": "system", "content": "You are MathPulse AI, a class analytics assistant for Filipino K-12 math teachers. Respond only with valid JSON."},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=0.3,
+                max_tokens=500,
+                response_format={"type": "json_object"},
+            )
+
+            content = response.choices[0].message.content or "{}"
+            parsed = json.loads(content)
+
+            return ClassInsights(
+                class_id=class_id,
+                generated_at=_now_iso(),
+                class_summary=parsed.get("class_summary", "Analytics data is being collected."),
+                top_weak_topics=parsed.get("top_weak_topics", weak_topics),
+                recommended_actions=parsed.get("recommended_actions", ["Encourage students to complete more quizzes."]),
+                class_strengths=parsed.get("class_strengths", "Class is actively using the platform."),
+                risk_distribution=risk_dist,
+                topic_performance=topic_perf,
+            )
+        except Exception as e:
+            logger.warning(f"DeepSeek insights generation failed: {e}")
+            # Return fallback insights
+            return ClassInsights(
+                class_id=class_id,
+                generated_at=_now_iso(),
+                class_summary=f"Class has {student_count} students with an average score of {class_average:.0f}%. {risk_dist.get('Unassessed', 0)} students have not yet taken any quizzes.",
+                top_weak_topics=weak_topics,
+                recommended_actions=[
+                    "Encourage unassessed students to complete their first quiz.",
+                    "Review struggling topics in the next class session.",
+                    "Schedule one-on-one check-ins with Critical risk students.",
+                ],
+                class_strengths="Students are enrolled and the platform is ready for use." if class_average < 50 else f"Class maintains a {class_average:.0f}% average.",
+                risk_distribution=risk_dist,
+                topic_performance=topic_perf,
+            )
+
+    def invalidate_cache(self, class_id: str) -> None:
+        self._cache.pop(class_id, None)
+
+
+# ─── Helpers ───────────────────────────────────────────────────────────────
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _days_since(iso_str: str) -> int:
+    try:
+        dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
+        return (datetime.now(timezone.utc) - dt).days
+    except Exception:
+        return 999
+
+
+def _extract_topic_from_quiz_id(quiz_id: str) -> Optional[str]:
+    """Extract topic name from quiz ID patterns like 'algebra-1', 'geometry-basics'."""
+    if not quiz_id:
+        return None
+    # Common patterns: subject-topic, module_quiz, etc.
+    parts = quiz_id.replace("_", "-").replace(".", "-").split("-")
+    if len(parts) >= 2:
+        # Capitalize and join meaningful parts
+        topic = " ".join(p.capitalize() for p in parts[:2] if p and not p.isdigit())
+        return topic if topic else None
+    return quiz_id.capitalize() if quiz_id else None
+
+
+# Singleton
+_engine_instance: Optional[ClassAnalyticsEngine] = None
+
+
+def get_class_analytics_engine() -> ClassAnalyticsEngine:
+    global _engine_instance
+    if _engine_instance is None:
+        _engine_instance = ClassAnalyticsEngine()
+    return _engine_instance

services/intervention_engine.py

@@ -0,0 +1,587 @@
+"""
+MathPulse AI — Intervention Engine
+
+Generates personalized intervention plans for at-risk students using
+real quiz data from Firestore and DeepSeek AI for learning path generation.
+"""
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger("mathpulse.intervention_engine")
+
+# ─── Firestore helper ──────────────────────────────────────────────────────
+
+_firebase_firestore = None
+
+
+def _get_firestore_client():
+    global _firebase_firestore
+    if _firebase_firestore is None:
+        try:
+            from firebase_admin import firestore as ff
+            _firebase_firestore = ff
+        except Exception:
+            return None
+    try:
+        return _firebase_firestore.client()
+    except Exception:
+        return None
+
+
+# ─── Models ────────────────────────────────────────────────────────────────
+
+class LearningStep(BaseModel):
+    step_number: int
+    type: Literal["video_lesson", "practice", "assessment", "chat_session", "review"] = "practice"
+    title: str
+    description: str = ""
+    duration_minutes: int = 10
+    num_items: Optional[int] = None
+    topic: str = ""
+    competency_tag: str = ""
+    difficulty: Literal["easy", "medium", "hard"] = "easy"
+    is_completed: bool = False
+    completion_score: Optional[float] = None
+
+
+class LearningPath(BaseModel):
+    student_id: str
+    generated_at: str = ""
+    methodology_tags: List[str] = Field(default_factory=lambda: ["Interactive", "Video", "Practice", "Quiz"])
+    steps: List[LearningStep] = Field(default_factory=list)
+    estimated_duration_days: int = 7
+    primary_weak_topic: str = ""
+    all_weak_topics: List[str] = Field(default_factory=list)
+    ai_rationale: str = ""
+
+
+class InterventionPlan(BaseModel):
+    student_id: str
+    student_name: str = ""
+    grade_level: str = ""
+    section: str = ""
+    risk_level: Literal["Low Risk", "Medium Risk", "High Risk", "Critical", "Unassessed"] = "Unassessed"
+    avg_score: float = 0.0
+    engagement_level: Literal["Low", "Medium", "High"] = "Low"
+    last_active: Optional[str] = None
+    weakest_topic: str = ""
+    weak_topics: List[str] = Field(default_factory=list)
+    accuracy_by_topic: Dict[str, float] = Field(default_factory=dict)
+    learning_strengths: str = ""
+    next_steps_summary: str = ""
+    learning_path: Optional[LearningPath] = None
+    generated_at: str = ""
+    teacher_recommendations: List[str] = Field(default_factory=list)
+
+
+# ─── Risk & Engagement Classification ─────────────────────────────────────
+
+def _classify_risk(avg_score: float, quiz_count: int, days_since_active: Optional[int]) -> str:
+    if quiz_count == 0:
+        return "Unassessed"
+    engagement_low = (days_since_active is None or days_since_active > 7) or quiz_count < 3
+    if avg_score < 50 and engagement_low:
+        return "Critical"
+    if avg_score < 60 or (avg_score < 75 and engagement_low):
+        return "High Risk"
+    if avg_score < 75:
+        return "Medium Risk"
+    return "Low Risk"
+
+
+def _classify_engagement(days_since_active: Optional[int], recent_quiz_count: int) -> str:
+    if days_since_active is not None and days_since_active <= 2 and recent_quiz_count >= 5:
+        return "High"
+    if days_since_active is not None and days_since_active <= 7:
+        return "Medium"
+    return "Low"
+
+
+# ─── Engine ────────────────────────────────────────────────────────────────
+
+class InterventionEngine:
+    """Generates full intervention plans for at-risk students."""
+
+    def __init__(self):
+        self._cache: Dict[str, tuple] = {}  # student_id -> (plan, timestamp)
+        self._cache_ttl = 3600  # 1 hour
+
+    async def generate_full_intervention(self, student_id: str, force: bool = False) -> InterventionPlan:
+        # Check cache
+        if not force and student_id in self._cache:
+            plan, cached_at = self._cache[student_id]
+            if time.time() - cached_at < self._cache_ttl:
+                return plan
+
+        db = _get_firestore_client()
+        if not db:
+            logger.error("Firestore client unavailable")
+            return InterventionPlan(student_id=student_id, generated_at=_now_iso())
+
+        # Fetch student data from managedStudents
+        student_data = await self._fetch_student_data(db, student_id)
+        if not student_data:
+            return InterventionPlan(
+                student_id=student_id,
+                student_name="Unknown",
+                generated_at=_now_iso(),
+                learning_strengths="No assessment data available yet.",
+                next_steps_summary="Assign a diagnostic quiz to begin intervention planning.",
+            )
+
+        # Fetch quiz attempts
+        quiz_attempts = await self._fetch_quiz_attempts(db, student_id, student_data)
+
+        # Compute metrics
+        now = datetime.now(timezone.utc)
+        quiz_count = len(quiz_attempts)
+        avg_score = 0.0
+        accuracy_by_topic: Dict[str, List[float]] = {}
+
+        if quiz_count > 0:
+            scores = [float(q.get("score", 0)) for q in quiz_attempts]
+            avg_score = sum(scores) / len(scores)
+
+            for attempt in quiz_attempts:
+                topic = self._extract_topic(attempt)
+                if topic:
+                    if topic not in accuracy_by_topic:
+                        accuracy_by_topic[topic] = []
+                    accuracy_by_topic[topic].append(float(attempt.get("score", 0)))
+
+        topic_avgs = {t: round(sum(s) / len(s), 1) for t, s in accuracy_by_topic.items() if s}
+        weak_topics = [t for t, s in sorted(topic_avgs.items(), key=lambda x: x[1]) if s < 70][:5]
+        strong_topics = [t for t, s in topic_avgs.items() if s >= 70]
+        weakest_topic = weak_topics[0] if weak_topics else student_data.get("weakestTopic", "Foundational Skills")
+        if weakest_topic == "N/A":
+            weakest_topic = "Foundational Skills"
+
+        # Last active
+        days_since_active = None
+        last_active_str = None
+        last_active_ts = student_data.get("lastActive")
+        if last_active_ts:
+            try:
+                if hasattr(last_active_ts, "seconds"):
+                    last_dt = datetime.fromtimestamp(last_active_ts.seconds, tz=timezone.utc)
+                else:
+                    last_dt = last_active_ts
+                last_active_str = last_dt.isoformat()
+                days_since_active = (now - last_dt).days
+            except Exception:
+                pass
+
+        # Recent quiz count (last 14 days)
+        recent_count = sum(1 for q in quiz_attempts if self._is_recent(q, now, 14))
+
+        risk_level = _classify_risk(avg_score, quiz_count, days_since_active)
+        engagement = _classify_engagement(days_since_active, recent_count)
+
+        # Generate AI insights
+        insights = await self._generate_insights(
+            grade_level=student_data.get("gradeLevel", student_data.get("grade", "11")),
+            section=student_data.get("section", ""),
+            risk_level=risk_level,
+            avg_score=avg_score,
+            engagement=engagement,
+            strong_topics=strong_topics,
+            weak_topics=weak_topics,
+            quiz_count=quiz_count,
+        )
+
+        # Generate learning path
+        learning_path = await self._generate_learning_path(
+            student_id=student_id,
+            grade_level=student_data.get("gradeLevel", student_data.get("grade", "11")),
+            risk_level=risk_level,
+            engagement=engagement,
+            weak_topics=weak_topics,
+            weakest_topic=weakest_topic,
+            topic_avgs=topic_avgs,
+        )
+
+        # Generate teacher recommendations
+        recommendations = await self._generate_recommendations(
+            grade_level=student_data.get("gradeLevel", student_data.get("grade", "11")),
+            risk_level=risk_level,
+            weak_topics=weak_topics,
+            avg_score=avg_score,
+        )
+
+        plan = InterventionPlan(
+            student_id=student_id,
+            student_name=student_data.get("name", "Unknown"),
+            grade_level=student_data.get("gradeLevel", student_data.get("grade", "")),
+            section=student_data.get("section", ""),
+            risk_level=risk_level,
+            avg_score=round(avg_score, 1),
+            engagement_level=engagement,
+            last_active=last_active_str,
+            weakest_topic=weakest_topic,
+            weak_topics=weak_topics,
+            accuracy_by_topic=topic_avgs,
+            learning_strengths=insights.get("learning_strengths", "Shows potential for improvement with guided support."),
+            next_steps_summary=insights.get("next_steps_summary", f"Focus on {weakest_topic} with structured practice."),
+            learning_path=learning_path,
+            generated_at=_now_iso(),
+            teacher_recommendations=recommendations,
+        )
+
+        # Cache
+        self._cache[student_id] = (plan, time.time())
+
+        # Persist to Firestore
+        try:
+            db.collection("intervention_plans").document(student_id).set(
+                plan.model_dump(), merge=True
+            )
+        except Exception as e:
+            logger.warning(f"Failed to persist intervention plan: {e}")
+
+        return plan
+
+    async def _fetch_student_data(self, db: Any, student_id: str) -> Optional[Dict]:
+        """Fetch student from managedStudents or users collection."""
+        try:
+            doc = db.collection("managedStudents").document(student_id).get()
+            if doc.exists:
+                return {"id": doc.id, **doc.to_dict()}
+
+            # Try users collection
+            doc2 = db.collection("users").document(student_id).get()
+            if doc2.exists:
+                return {"id": doc2.id, **doc2.to_dict()}
+        except Exception as e:
+            logger.debug(f"Error fetching student {student_id}: {e}")
+        return None
+
+    async def _fetch_quiz_attempts(self, db: Any, student_id: str, student_data: Dict) -> List[Dict]:
+        """Fetch quiz attempts from progress collection and practice_results."""
+        attempts = []
+
+        # Try progress/{student_id}
+        for lookup_id in [student_id, student_data.get("lrn"), student_data.get("accountUid")]:
+            if not lookup_id:
+                continue
+            try:
+                progress_doc = db.collection("progress").document(lookup_id).get()
+                if progress_doc.exists:
+                    pdata = progress_doc.to_dict()
+                    quiz_data = pdata.get("quizAttempts", [])
+                    if quiz_data:
+                        attempts = quiz_data
+                        break
+            except Exception:
+                pass
+
+        # Also check practice_results
+        account_uid = student_data.get("accountUid") or student_id
+        try:
+            sessions = (
+                db.collection("practice_results")
+                .document(account_uid)
+                .collection("sessions")
+                .order_by("submitted_at", direction="DESCENDING")
+                .limit(30)
+                .stream()
+            )
+            for sess in sessions:
+                sd = sess.to_dict()
+                attempts.append({
+                    "quizId": sd.get("session_id", ""),
+                    "score": sd.get("score_percent", 0),
+                    "completedAt": sd.get("submitted_at"),
+                    "answers": sd.get("per_question_feedback", []),
+                    "subject": sd.get("subject", ""),
+                })
+        except Exception:
+            pass
+
+        return attempts[:30]
+
+    def _extract_topic(self, attempt: Dict) -> Optional[str]:
+        """Extract topic from quiz attempt."""
+        # Check subject field first
+        if attempt.get("subject"):
+            return attempt["subject"]
+        # Try to extract from quizId
+        quiz_id = attempt.get("quizId", "")
+        if not quiz_id:
+            return None
+        parts = quiz_id.replace("_", "-").replace(".", "-").split("-")
+        if len(parts) >= 2:
+            topic = " ".join(p.capitalize() for p in parts[:2] if p and not p.isdigit())
+            return topic if topic else None
+        return quiz_id.capitalize() if quiz_id else None
+
+    def _is_recent(self, attempt: Dict, now: datetime, days: int) -> bool:
+        completed = attempt.get("completedAt")
+        if not completed:
+            return False
+        try:
+            if hasattr(completed, "seconds"):
+                q_dt = datetime.fromtimestamp(completed.seconds, tz=timezone.utc)
+            elif isinstance(completed, datetime):
+                q_dt = completed
+            else:
+                return False
+            return (now - q_dt).days <= days
+        except Exception:
+            return False
+
+    async def _generate_insights(self, **kwargs) -> Dict[str, str]:
+        """Generate learning_strengths and next_steps_summary via DeepSeek."""
+        prompt = f"""You are MathPulse AI analyzing a Filipino K-12 student's performance data.
+
+Student: Grade {kwargs['grade_level']}, Section {kwargs['section']}
+Risk Level: {kwargs['risk_level']}
+Average Score: {kwargs['avg_score']:.1f}%
+Engagement: {kwargs['engagement']}
+Strong Topics (accuracy > 70%): {', '.join(kwargs['strong_topics'][:3]) or 'None identified yet'}
+Weak Topics (accuracy < 60%): {', '.join(kwargs['weak_topics'][:3]) or 'None identified yet'}
+Quiz Attempt Count (last 30 days): {kwargs['quiz_count']}
+
+Generate two SHORT insights (max 20 words each):
+1. LEARNING STRENGTHS: What the student excels at or shows potential in. Be specific and encouraging.
+2. NEXT STEPS: The single most important action for the teacher/student right now.
+
+Return as JSON:
+{{"learning_strengths": "...", "next_steps_summary": "..."}}"""
+
+        try:
+            from services.ai_client import get_deepseek_client, CHAT_MODEL
+            client = get_deepseek_client()
+            response = client.chat.completions.create(
+                model=CHAT_MODEL,
+                messages=[
+                    {"role": "system", "content": "You are MathPulse AI. Respond only with valid JSON."},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=0.3,
+                max_tokens=200,
+                response_format={"type": "json_object"},
+            )
+            content = response.choices[0].message.content or "{}"
+            return json.loads(content)
+        except Exception as e:
+            logger.warning(f"DeepSeek insights failed: {e}")
+            return {
+                "learning_strengths": "Shows willingness to engage with the platform." if kwargs['quiz_count'] > 0 else "No assessment data yet — potential to be discovered.",
+                "next_steps_summary": f"Begin with foundational practice in {kwargs['weak_topics'][0] if kwargs['weak_topics'] else 'core topics'}.",
+            }
+
+    async def _generate_learning_path(self, **kwargs) -> LearningPath:
+        """Generate a structured learning path via DeepSeek."""
+        student_id = kwargs["student_id"]
+        weak_topics = kwargs["weak_topics"]
+        weakest_topic = kwargs["weakest_topic"]
+        engagement = kwargs["engagement"]
+        risk_level = kwargs["risk_level"]
+        grade_level = kwargs["grade_level"]
+        topic_avgs = kwargs["topic_avgs"]
+
+        style_hint = "shorter steps (5-8 min), gamified" if engagement == "Low" else "standard pacing (10-15 min)"
+        estimated_days = 5 if risk_level == "Critical" else 7
+
+        prompt = f"""Create a personalized intervention learning path for a Filipino K-12 math student.
+
+Student Profile:
+- Grade Level: {grade_level}
+- Risk Level: {risk_level}
+- Primary Weak Topic: {weakest_topic}
+- All Weak Topics: {', '.join(weak_topics[:4]) or weakest_topic}
+- Accuracy by Topic: {json.dumps(topic_avgs)}
+- Engagement: {engagement} → {style_hint}
+
+Create a 4-6 step learning path that:
+1. Starts with the MOST CRITICAL weak topic (lowest accuracy)
+2. Uses varied methodology: video → practice → assessment → review cycle
+3. Scales difficulty: start easy, progress to grade-level
+4. Total estimated time: {estimated_days} days
+
+Return ONLY valid JSON:
+{{
+  "methodology_tags": ["Interactive", "Video", "Practice", "Quiz"],
+  "estimated_duration_days": {estimated_days},
+  "ai_rationale": "1 sentence explaining why this path was chosen",
+  "steps": [
+    {{
+      "step_number": 1,
+      "type": "video_lesson",
+      "title": "Topic - Concept Name",
+      "description": "Brief description of what student will learn",
+      "duration_minutes": 8,
+      "num_items": null,
+      "topic": "Topic Name",
+      "competency_tag": "M11GM-Ia-1",
+      "difficulty": "easy"
+    }}
+  ]
+}}"""
+
+        try:
+            from services.ai_client import get_deepseek_client, CHAT_MODEL
+            client = get_deepseek_client()
+            response = client.chat.completions.create(
+                model=CHAT_MODEL,
+                messages=[
+                    {"role": "system", "content": "You are a curriculum designer for Filipino K-12 DepEd math. Respond only with valid JSON."},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=0.4,
+                max_tokens=800,
+                response_format={"type": "json_object"},
+            )
+            content = response.choices[0].message.content or "{}"
+            parsed = json.loads(content)
+
+            steps = []
+            for s in parsed.get("steps", []):
+                steps.append(LearningStep(
+                    step_number=s.get("step_number", len(steps) + 1),
+                    type=s.get("type", "practice"),
+                    title=s.get("title", "Practice Step"),
+                    description=s.get("description", ""),
+                    duration_minutes=s.get("duration_minutes", 10),
+                    num_items=s.get("num_items"),
+                    topic=s.get("topic", weakest_topic),
+                    competency_tag=s.get("competency_tag", ""),
+                    difficulty=s.get("difficulty", "easy"),
+                ))
+
+            return LearningPath(
+                student_id=student_id,
+                generated_at=_now_iso(),
+                methodology_tags=parsed.get("methodology_tags", ["Interactive", "Video", "Practice", "Quiz"]),
+                steps=steps,
+                estimated_duration_days=parsed.get("estimated_duration_days", estimated_days),
+                primary_weak_topic=weakest_topic,
+                all_weak_topics=weak_topics,
+                ai_rationale=parsed.get("ai_rationale", f"Focused on {weakest_topic} as the primary area needing improvement."),
+            )
+        except Exception as e:
+            logger.warning(f"DeepSeek learning path generation failed: {e}")
+            # Fallback: generate a basic path
+            return self._fallback_learning_path(student_id, weakest_topic, weak_topics)
+
+    def _fallback_learning_path(self, student_id: str, weakest_topic: str, weak_topics: List[str]) -> LearningPath:
+        """Generate a basic learning path without AI."""
+        steps = [
+            LearningStep(step_number=1, type="video_lesson", title=f"{weakest_topic} - Fundamentals",
+                         description="Review core concepts", duration_minutes=8, topic=weakest_topic, difficulty="easy"),
+            LearningStep(step_number=2, type="practice", title=f"{weakest_topic} - Guided Practice",
+                         description="Work through examples", duration_minutes=12, num_items=10, topic=weakest_topic, difficulty="easy"),
+            LearningStep(step_number=3, type="practice", title=f"{weakest_topic} - Independent Practice",
+                         description="Solve problems independently", duration_minutes=15, num_items=10, topic=weakest_topic, difficulty="medium"),
+            LearningStep(step_number=4, type="assessment", title=f"{weakest_topic} - Mastery Check",
+                         description="Demonstrate understanding", duration_minutes=10, num_items=5, topic=weakest_topic, difficulty="medium"),
+        ]
+        if len(weak_topics) > 1:
+            steps.append(LearningStep(step_number=5, type="review", title=f"{weak_topics[1]} - Review",
+                                      description="Brief review of secondary weak area", duration_minutes=10, topic=weak_topics[1], difficulty="easy"))
+
+        return LearningPath(
+            student_id=student_id,
+            generated_at=_now_iso(),
+            steps=steps,
+            estimated_duration_days=7,
+            primary_weak_topic=weakest_topic,
+            all_weak_topics=weak_topics,
+            ai_rationale=f"Structured path focusing on {weakest_topic} with progressive difficulty.",
+        )
+
+    async def _generate_recommendations(self, **kwargs) -> List[str]:
+        """Generate teacher recommendations via DeepSeek."""
+        prompt = f"""Generate 3-5 concise, actionable recommendations for a teacher working with this at-risk student.
+
+Student: Grade {kwargs['grade_level']}, Risk: {kwargs['risk_level']}
+Weak Topics: {', '.join(kwargs['weak_topics'][:3]) or 'Foundational Skills'}
+Avg Score: {kwargs['avg_score']:.0f}%
+
+Return as a JSON array of strings. Each recommendation max 25 words. Be specific to the weak topics."""
+
+        try:
+            from services.ai_client import get_deepseek_client, CHAT_MODEL
+            client = get_deepseek_client()
+            response = client.chat.completions.create(
+                model=CHAT_MODEL,
+                messages=[
+                    {"role": "system", "content": "You are a K-12 math education advisor. Respond only with a JSON array of strings."},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=0.3,
+                max_tokens=300,
+                response_format={"type": "json_object"},
+            )
+            content = response.choices[0].message.content or "[]"
+            parsed = json.loads(content)
+            if isinstance(parsed, list):
+                return parsed[:5]
+            if isinstance(parsed, dict):
+                return parsed.get("recommendations", parsed.get("actions", []))[:5]
+        except Exception as e:
+            logger.warning(f"DeepSeek recommendations failed: {e}")
+
+        return [
+            f"Schedule 1-on-1 review session for {kwargs['weak_topics'][0] if kwargs['weak_topics'] else 'foundational skills'}.",
+            "Assign additional practice problems at reduced difficulty level.",
+            "Monitor quiz completion and provide immediate feedback.",
+        ]
+
+    async def complete_step(self, student_id: str, step_number: int, score: float, time_spent: int) -> Dict:
+        """Mark a learning step as completed."""
+        db = _get_firestore_client()
+        if not db:
+            return {"error": "Firestore unavailable"}
+
+        try:
+            plan_ref = db.collection("intervention_plans").document(student_id)
+            plan_doc = plan_ref.get()
+            if not plan_doc.exists:
+                return {"error": "No intervention plan found"}
+
+            plan_data = plan_doc.to_dict()
+            learning_path = plan_data.get("learning_path", {})
+            steps = learning_path.get("steps", [])
+
+            for step in steps:
+                if step.get("step_number") == step_number:
+                    step["is_completed"] = True
+                    step["completion_score"] = score
+                    break
+
+            plan_ref.update({"learning_path.steps": steps})
+
+            # Invalidate cache
+            self._cache.pop(student_id, None)
+
+            return {"status": "completed", "step_number": step_number, "score": score}
+        except Exception as e:
+            logger.error(f"Failed to complete step: {e}")
+            return {"error": str(e)}
+
+    def invalidate_cache(self, student_id: str) -> None:
+        self._cache.pop(student_id, None)
+
+
+# ─── Helpers ───────────────────────────────────────────────────────────────
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# Singleton
+_engine_instance: Optional[InterventionEngine] = None
+
+
+def get_intervention_engine() -> InterventionEngine:
+    global _engine_instance
+    if _engine_instance is None:
+        _engine_instance = InterventionEngine()
+    return _engine_instance

services/student_intelligence_pipeline.py

@@ -0,0 +1,546 @@
+"""
+MathPulse AI — Student Intelligence Pipeline
+
+Central event processor that:
+- Intercepts every student activity completion
+- Recomputes P (systemPerformanceAvg) from all accumulated scores
+- Calls existing compute_wri() with updated D, G, P
+- Writes denormalized student_profiles and class summaries
+- Triggers DeepSeek AI context generation when meaningful
+"""
+
+import json
+import logging
+import os
+import time
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger("mathpulse.pipeline")
+
+# ─── Firestore ─────────────────────────────────────────────────────────────
+
+_firebase_firestore = None
+
+
+def _get_db():
+    global _firebase_firestore
+    if _firebase_firestore is None:
+        try:
+            from firebase_admin import firestore as ff
+            _firebase_firestore = ff
+        except Exception:
+            return None
+    try:
+        return _firebase_firestore.client()
+    except Exception:
+        return None
+
+
+# ─── Models ────────────────────────────────────────────────────────────────
+
+class StudentActivityEvent(BaseModel):
+    student_id: str
+    event_type: Literal["diagnostic", "quiz", "battle", "lesson", "module", "session"]
+    event_data: Dict[str, Any] = Field(default_factory=dict)
+    occurred_at: str  # ISO string
+    class_id: str = ""
+    teacher_id: str = ""
+
+
+class ProfileUpdateResult(BaseModel):
+    student_id: str
+    profile_updated: bool = False
+    p_updated: bool = False
+    new_p: Optional[float] = None
+    wri_recomputed: bool = False
+    new_wri: Optional[float] = None
+    new_risk_status: str = "pending_assessment"
+    risk_status_changed: bool = False
+    previous_risk_status: Optional[str] = None
+    ai_regenerated: bool = False
+
+
+# ─── Source weights and recency multipliers for P computation ──────────────
+
+SOURCE_WEIGHTS = {
+    "practice": 1.0,
+    "lesson_quiz": 1.0,
+    "module_quiz": 1.2,
+    "assessment": 1.2,
+    "battle": 0.8,
+    "intervention_quiz": 1.3,
+    "diagnostic": 1.0,
+}
+
+
+def _recency_multiplier(occurred_at: datetime) -> float:
+    now = datetime.now(timezone.utc)
+    days_ago = (now - occurred_at).days
+    if days_ago <= 7:
+        return 1.5
+    if days_ago <= 30:
+        return 1.0
+    return 0.6
+
+
+# ─── Pipeline ──────────────────────────────────────────────────────────────
+
+class StudentIntelligencePipeline:
+
+    async def process_event(self, event: StudentActivityEvent) -> ProfileUpdateResult:
+        """Master entry point. Called after every student activity."""
+        db = _get_db()
+        if not db:
+            logger.error("Firestore unavailable")
+            return ProfileUpdateResult(student_id=event.student_id)
+
+        result = ProfileUpdateResult(student_id=event.student_id)
+
+        try:
+            # 1. Load or create profile
+            profile = self._load_profile(db, event.student_id)
+
+            # 2. Load managed student data (source of D, G, weights)
+            managed = self._load_managed_student(db, event.student_id)
+
+            # 3. Update profile section from event
+            self._update_profile_section(profile, event)
+
+            # 4. Compute P from all activity (skip for session events)
+            if event.event_type != "session":
+                d_score = managed.get("diagnosticScore") or profile.get("diagnostic", {}).get("overall_score")
+                new_p = self._compute_system_performance_avg(db, event.student_id, event, diagnostic_score=d_score)
+                profile["system_performance_avg"] = new_p
+                result.p_updated = True
+                result.new_p = new_p
+            else:
+                new_p = profile.get("system_performance_avg")
+
+            # 5. Recompute WRI using existing compute_wri function
+            d = managed.get("diagnosticScore") or profile.get("diagnostic", {}).get("overall_score")
+            g = managed.get("externalGradesAvg") or profile.get("external_grades_avg")
+            weights = managed.get("weights") or profile.get("wri_weights") or {"w1": 0.30, "w2": 0.40, "w3": 0.30}
+
+            if d is not None and event.event_type != "session":
+                from services.wri_service import compute_wri
+                wri_result = compute_wri(d=d, g=g, p=new_p, weights=weights)
+
+                previous_status = profile.get("risk_status", "pending_assessment")
+                new_status = wri_result["risk_status"]
+
+                result.wri_recomputed = True
+                result.new_wri = wri_result["wri"]
+                result.new_risk_status = new_status
+                result.previous_risk_status = previous_status
+                result.risk_status_changed = previous_status != new_status
+
+                # Update profile
+                profile["wri"] = wri_result["wri"]
+                profile["risk_status"] = new_status
+                profile["previous_risk_status"] = previous_status
+                profile["system_performance_avg"] = new_p
+                profile["external_grades_avg"] = g
+                profile["diagnostic_score"] = d
+                profile["wri_weights"] = weights
+                profile["g_fallback"] = wri_result["g_fallback"]
+                profile["p_fallback"] = wri_result["p_fallback"]
+                profile["wri_updated_at"] = _now_iso()
+
+                # Compute risk trend
+                profile["risk_trend"] = self._compute_risk_trend(profile)
+
+                # 6. Write to managedStudents (update P, WRI, riskStatus)
+                self._update_managed_student(db, event.student_id, wri_result, new_p)
+
+            # 7. AI context generation (cost-controlled)
+            if self._should_regenerate_ai(event, profile, result):
+                ai_ctx = await self._generate_ai_context(profile, event)
+                if ai_ctx:
+                    profile["ai_context"] = ai_ctx
+                    result.ai_regenerated = True
+
+            # 8. Update metadata
+            profile["last_updated_at"] = _now_iso()
+            profile["last_event_type"] = event.event_type
+            profile["last_event_at"] = event.occurred_at
+            profile["profile_version"] = profile.get("profile_version", 0) + 1
+
+            # 9. Write student_profiles/{id}
+            db.collection("student_profiles").document(event.student_id).set(profile, merge=True)
+            result.profile_updated = True
+
+            # 10. Write denormalized summary
+            if event.class_id:
+                self._write_summary(db, event.class_id, event.student_id, profile)
+
+            # 11. Invalidate class analytics cache
+            if event.class_id:
+                self._invalidate_class_cache(db, event.class_id)
+
+        except Exception as e:
+            logger.error(f"Pipeline error for {event.student_id}: {e}", exc_info=True)
+
+        return result
+
+    # ─── Profile loading ───────────────────────────────────────────────────
+
+    def _load_profile(self, db, student_id: str) -> Dict:
+        doc = db.collection("student_profiles").document(student_id).get()
+        if doc.exists:
+            return doc.to_dict()
+        return {"student_id": student_id, "profile_version": 0, "risk_status": "pending_assessment"}
+
+    def _load_managed_student(self, db, student_id: str) -> Dict:
+        doc = db.collection("managedStudents").document(student_id).get()
+        if doc.exists:
+            return doc.to_dict()
+        # Try users collection
+        doc2 = db.collection("users").document(student_id).get()
+        return doc2.to_dict() if doc2.exists else {}
+
+    # ─── Profile section updates ───────────────────────────────────────────
+
+    def _update_profile_section(self, profile: Dict, event: StudentActivityEvent):
+        ed = event.event_data
+
+        if event.event_type == "diagnostic":
+            profile.setdefault("diagnostic", {})
+            profile["diagnostic"]["completed"] = True
+            profile["diagnostic"]["completed_at"] = event.occurred_at
+            profile["diagnostic"]["overall_score"] = ed.get("overall_score", 0)
+            profile["diagnostic"]["per_topic_scores"] = ed.get("per_topic_scores", {})
+            profile["diagnostic"]["weak_topics"] = [
+                t for t, s in ed.get("per_topic_scores", {}).items() if s < 60
+            ]
+            profile["diagnostic"]["strong_topics"] = [
+                t for t, s in ed.get("per_topic_scores", {}).items() if s >= 75
+            ]
+            profile["diagnostic_score"] = ed.get("overall_score", 0)
+
+        elif event.event_type in ("quiz", "battle"):
+            qp = profile.setdefault("quiz_performance", {
+                "total_attempts": 0, "avg_score_all_time": None,
+                "recent_attempts": [], "accuracy_by_topic": {},
+            })
+            qp["total_attempts"] = qp.get("total_attempts", 0) + 1
+
+            score = ed.get("score", 0)
+            topic = ed.get("topic", "")
+
+            # Update rolling average
+            prev_avg = qp.get("avg_score_all_time") or 0
+            prev_count = qp["total_attempts"] - 1
+            if prev_count > 0:
+                qp["avg_score_all_time"] = round(
+                    (prev_avg * prev_count + score) / qp["total_attempts"], 1
+                )
+            else:
+                qp["avg_score_all_time"] = score
+
+            # Update per-topic accuracy (exponential moving average)
+            if topic:
+                acc = qp.setdefault("accuracy_by_topic", {})
+                prev = acc.get(topic, score)
+                acc[topic] = round(prev * 0.7 + score * 0.3, 1)
+
+            # Add to recent attempts (keep last 10)
+            recent = qp.setdefault("recent_attempts", [])
+            recent.insert(0, {
+                "quiz_id": ed.get("quiz_id", ""),
+                "topic": topic,
+                "competency_tag": ed.get("competency_tag", ""),
+                "score": score,
+                "source": ed.get("source", event.event_type),
+                "attempted_at": event.occurred_at,
+            })
+            qp["recent_attempts"] = recent[:10]
+
+            # Compute lowest/highest topics
+            if qp.get("accuracy_by_topic"):
+                sorted_topics = sorted(qp["accuracy_by_topic"].items(), key=lambda x: x[1])
+                qp["lowest_accuracy_topics"] = [t for t, _ in sorted_topics[:5]]
+                qp["highest_accuracy_topics"] = [t for t, _ in sorted_topics[-5:]]
+
+            # Battle-specific
+            if event.event_type == "battle":
+                bp = profile.setdefault("battle_performance", {"total_battles": 0, "battles_won": 0})
+                bp["total_battles"] = bp.get("total_battles", 0) + 1
+                if ed.get("won"):
+                    bp["battles_won"] = bp.get("battles_won", 0) + 1
+                bp["win_rate"] = round(bp["battles_won"] / bp["total_battles"] * 100, 1) if bp["total_battles"] > 0 else 0
+                bp["avg_battle_score"] = score
+                bp["last_battle_at"] = event.occurred_at
+
+        elif event.event_type == "lesson":
+            ce = profile.setdefault("content_engagement", {"lessons_completed": 0, "modules_completed": 0, "topics_studied": []})
+            if ed.get("is_completed"):
+                ce["lessons_completed"] = ce.get("lessons_completed", 0) + 1
+            topic = ed.get("topic", "")
+            if topic and topic not in ce.get("topics_studied", []):
+                ce.setdefault("topics_studied", []).append(topic)
+            ce["last_content_at"] = event.occurred_at
+
+        elif event.event_type == "module":
+            ce = profile.setdefault("content_engagement", {"lessons_completed": 0, "modules_completed": 0, "topics_studied": []})
+            if ed.get("is_completed"):
+                ce["modules_completed"] = ce.get("modules_completed", 0) + 1
+            ce["last_content_at"] = event.occurred_at
+
+        elif event.event_type == "session":
+            eng = profile.setdefault("engagement", {"total_sessions": 0, "login_streak": 0})
+            eng["total_sessions"] = eng.get("total_sessions", 0) + 1
+            eng["last_active_at"] = event.occurred_at
+            eng["days_since_last_active"] = 0
+
+    # ─── P computation ─────────────────────────────────────────────────────
+
+    def _compute_system_performance_avg(
+        self, db, student_id: str, current_event: StudentActivityEvent, diagnostic_score: Optional[float] = None
+    ) -> float:
+        """Compute P from ALL in-platform scores with source weights and recency."""
+        scores = []
+
+        # Fetch from quizSubmissions
+        try:
+            subs = db.collection("quizSubmissions").where("lrn", "==", student_id).order_by(
+                "submittedAt", direction="DESCENDING"
+            ).limit(50).stream()
+            for s in subs:
+                d = s.to_dict()
+                score = d.get("score", 0)
+                source = d.get("source", "practice")
+                submitted = d.get("submittedAt")
+                if submitted and hasattr(submitted, "seconds"):
+                    dt = datetime.fromtimestamp(submitted.seconds, tz=timezone.utc)
+                else:
+                    dt = datetime.now(timezone.utc) - timedelta(days=15)
+                scores.append((score, source, dt))
+        except Exception as e:
+            logger.debug(f"quizSubmissions fetch failed for {student_id}: {e}")
+
+        # Fetch from progress/{id}.quizAttempts
+        try:
+            for lookup_id in [student_id]:
+                prog = db.collection("progress").document(lookup_id).get()
+                if prog.exists:
+                    attempts = prog.to_dict().get("quizAttempts", [])
+                    for a in attempts:
+                        score = a.get("score", 0)
+                        completed = a.get("completedAt")
+                        if completed and hasattr(completed, "seconds"):
+                            dt = datetime.fromtimestamp(completed.seconds, tz=timezone.utc)
+                        else:
+                            dt = datetime.now(timezone.utc) - timedelta(days=15)
+                        scores.append((score, "practice", dt))
+                    break
+        except Exception as e:
+            logger.debug(f"progress fetch failed for {student_id}: {e}")
+
+        # Include current event score
+        if current_event.event_type in ("quiz", "battle", "diagnostic"):
+            event_score = current_event.event_data.get("score") or current_event.event_data.get("overall_score", 0)
+            source = current_event.event_data.get("source", current_event.event_type)
+            try:
+                dt = datetime.fromisoformat(current_event.occurred_at.replace("Z", "+00:00"))
+            except Exception:
+                dt = datetime.now(timezone.utc)
+            scores.append((event_score, source, dt))
+
+        if not scores:
+            # Fallback to D (diagnostic score)
+            return float(diagnostic_score) if diagnostic_score is not None else 0.0
+
+        # Weighted average
+        weighted_sum = 0.0
+        weight_sum = 0.0
+        for score, source, dt in scores:
+            sw = SOURCE_WEIGHTS.get(source, 1.0)
+            rm = _recency_multiplier(dt)
+            weighted_sum += score * sw * rm
+            weight_sum += sw * rm
+
+        return round(weighted_sum / weight_sum, 2) if weight_sum > 0 else 0.0
+
+    # ─── WRI update to managedStudents ─────────────────────────────────────
+
+    def _update_managed_student(self, db, student_id: str, wri_result: Dict, new_p: float):
+        """Write updated WRI data to managedStudents/{id}."""
+        try:
+            ref = db.collection("managedStudents").document(student_id)
+            if not ref.get().exists:
+                return
+            ref.update({
+                "wri": wri_result["wri"],
+                "riskStatus": wri_result["risk_status"],
+                "systemPerformanceAvg": new_p,
+                "riskUpdatedAt": datetime.now(timezone.utc),
+            })
+        except Exception as e:
+            logger.warning(f"Failed to update managedStudents/{student_id}: {e}")
+
+    # ─── Risk trend ────────────────────────────────────────────────────────
+
+    def _compute_risk_trend(self, profile: Dict) -> str:
+        qp = profile.get("quiz_performance", {})
+        avg_all = qp.get("avg_score_all_time")
+        if avg_all is None:
+            return "insufficient_data"
+        # Compute 7-day avg from recent_attempts
+        recent = qp.get("recent_attempts", [])
+        now = datetime.now(timezone.utc)
+        scores_7d = []
+        for a in recent:
+            try:
+                at = a.get("attempted_at", "")
+                if isinstance(at, str):
+                    dt = datetime.fromisoformat(at.replace("Z", "+00:00"))
+                elif hasattr(at, "seconds"):
+                    dt = datetime.fromtimestamp(at.seconds, tz=timezone.utc)
+                else:
+                    continue
+                if (now - dt).days <= 7:
+                    scores_7d.append(a.get("score", 0))
+            except Exception:
+                continue
+        if len(scores_7d) < 2:
+            return "insufficient_data"
+        avg_7d = sum(scores_7d) / len(scores_7d)
+        if avg_7d > avg_all + 5:
+            return "improving"
+        if avg_7d < avg_all - 5:
+            return "worsening"
+        return "stable"
+
+    # ─── AI context generation ─────────────────────────────────────────────
+
+    def _should_regenerate_ai(self, event: StudentActivityEvent, profile: Dict, result: ProfileUpdateResult) -> bool:
+        if event.event_type == "session":
+            return False
+        if event.event_type == "diagnostic":
+            return True
+        if result.risk_status_changed:
+            return True
+        ai_ctx = profile.get("ai_context", {})
+        if not ai_ctx.get("generated_at"):
+            return True
+        # Rate limit: max once per 6 hours
+        try:
+            last_gen = datetime.fromisoformat(ai_ctx["generated_at"].replace("Z", "+00:00"))
+            if (datetime.now(timezone.utc) - last_gen).total_seconds() < 21600:
+                return False
+        except Exception:
+            pass
+        return event.event_type in ("quiz", "battle") and result.new_risk_status in ("critical", "at_risk")
+
+    async def _generate_ai_context(self, profile: Dict, event: StudentActivityEvent) -> Optional[Dict]:
+        """Call DeepSeek for AI context generation."""
+        try:
+            from services.ai_client import get_deepseek_client, CHAT_MODEL
+
+            weak_topics = profile.get("quiz_performance", {}).get("lowest_accuracy_topics", [])[:3]
+            diag = profile.get("diagnostic", {})
+            qp = profile.get("quiz_performance", {})
+
+            prompt = f"""Analyze this student's FULL learning history and generate insights.
+
+Student: Grade {profile.get('grade_level', '?')}, Section {profile.get('section', '?')}
+WRI: {profile.get('wri', 'N/A')} | Status: {profile.get('risk_status', 'pending_assessment')}
+Previous Status: {profile.get('previous_risk_status', 'N/A')}
+Risk Trend: {profile.get('risk_trend', 'insufficient_data')}
+
+Diagnostic Score: {diag.get('overall_score', 'N/A')}%
+System Performance (P): {profile.get('system_performance_avg', 'N/A')}%
+External Grades (G): {profile.get('external_grades_avg', 'N/A')}%
+
+Quiz Performance: {qp.get('total_attempts', 0)} attempts, avg {qp.get('avg_score_all_time', 'N/A')}%
+Weakest Topics: {', '.join(weak_topics) or 'None identified'}
+Strongest Topics: {', '.join(qp.get('highest_accuracy_topics', [])[:3]) or 'None identified'}
+
+Latest Event: {event.event_type} — score {event.event_data.get('score', event.event_data.get('overall_score', 'N/A'))}%
+
+Generate JSON:
+{{"ai_summary": "1 sentence: current status with WRI context",
+"ai_strengths": "specific topics/skills they excel at",
+"ai_concerns": "specific gaps needing attention",
+"ai_recommendation": "top 1 action for teacher this week (max 20 words)",
+"notable_change": "what changed most since last update",
+"rag_topics_used": []}}"""
+
+            client = get_deepseek_client()
+            response = client.chat.completions.create(
+                model=CHAT_MODEL,
+                messages=[
+                    {"role": "system", "content": "You are MathPulse AI analyzing Filipino K-12 math student data. Respond only with valid JSON."},
+                    {"role": "user", "content": prompt},
+                ],
+                temperature=0.3,
+                max_tokens=400,
+                response_format={"type": "json_object"},
+            )
+            content = response.choices[0].message.content or "{}"
+            parsed = json.loads(content)
+            parsed["generated_at"] = _now_iso()
+            parsed["based_on_wri_status"] = profile.get("risk_status", "pending_assessment")
+            return parsed
+        except Exception as e:
+            logger.warning(f"DeepSeek AI context generation failed: {e}")
+            return None
+
+    # ─── Denormalized summary ──────────────────────────────────────────────
+
+    def _write_summary(self, db, class_id: str, student_id: str, profile: Dict):
+        """Write lightweight summary for Teacher Dashboard reads."""
+        qp = profile.get("quiz_performance", {})
+        ai = profile.get("ai_context", {})
+        eng = profile.get("engagement", {})
+
+        summary = {
+            "student_id": student_id,
+            "display_name": profile.get("display_name", ""),
+            "wri": profile.get("wri"),
+            "risk_status": profile.get("risk_status", "pending_assessment"),
+            "previous_risk_status": profile.get("previous_risk_status"),
+            "risk_trend": profile.get("risk_trend", "insufficient_data"),
+            "avg_score_all_time": qp.get("avg_score_all_time"),
+            "avg_score_last_7_days": qp.get("avg_score_last_7_days"),
+            "score_trend": qp.get("score_trend", "insufficient_data"),
+            "last_active_at": eng.get("last_active_at"),
+            "days_since_last_active": eng.get("days_since_last_active", 999),
+            "weakest_topic": (qp.get("lowest_accuracy_topics") or [None])[0],
+            "ai_summary": ai.get("ai_summary", ""),
+            "has_intervention_plan": profile.get("intervention", {}).get("has_active_plan", False),
+            "lessons_completed": profile.get("content_engagement", {}).get("lessons_completed", 0),
+            "modules_completed": profile.get("content_engagement", {}).get("modules_completed", 0),
+            "updated_at": _now_iso(),
+        }
+
+        try:
+            db.collection("classes").document(class_id).collection("student_summaries").document(student_id).set(summary, merge=True)
+        except Exception as e:
+            logger.warning(f"Failed to write summary for {student_id} in class {class_id}: {e}")
+
+    def _invalidate_class_cache(self, db, class_id: str):
+        try:
+            db.collection("class_analytics").document(class_id).update({"cache_valid": False})
+        except Exception:
+            pass
+
+
+# ─── Helpers ───────────────────────────────────────────────────────────────
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# Singleton
+_pipeline: Optional[StudentIntelligencePipeline] = None
+
+
+def get_pipeline() -> StudentIntelligencePipeline:
+    global _pipeline
+    if _pipeline is None:
+        _pipeline = StudentIntelligencePipeline()
+    return _pipeline