deploy: backend from 694f8ea

main.py

@@ -1160,23 +1160,10 @@ class RequestMiddleware(BaseHTTPMiddleware):
                 },
                 headers={"X-Request-ID": request_id},
             )
-        except Exception as exc:
-            duration = round(time.time() - start, 3)
-            logger.error(f"[{request_id}] Unhandled error after {duration}s: {exc}")
-            return JSONResponse(
-                status_code=500,
-                content={
-                    "detail": "Internal server error",
-                    "error": type(exc).__name__,
-                    "message": str(exc),
-                    "requestId": request_id,
-                },
-                headers={"X-Request-ID": request_id},
-            )
 
 
-app.add_middleware(RequestMiddleware)
 app.add_middleware(AuthMiddleware)
+app.add_middleware(RequestMiddleware)
 
 # Set up rate limiting with slowapi
 if HAS_RATE_LIMITING and setup_rate_limiting:  # type: ignore[truthy-function]

middleware/rate_limiter.py

@@ -8,6 +8,8 @@ from fastapi import Request
 from slowapi import Limiter
 from slowapi.errors import RateLimitExceeded as SlowAPIRateLimitExceeded
 
+RateLimitExceeded = SlowAPIRateLimitExceeded
+
 logger = logging.getLogger("mathpulse.ratelimit")
 
 # Environment-based configuration with defaults

routes/admin_routes.py

@@ -1,7 +1,12 @@
-from typing import Optional
+import os
+import json
+import logging
+import urllib.request
+import urllib.error
+from datetime import datetime, timezone
+from typing import Optional, Dict, Any
 from fastapi import APIRouter, Depends, HTTPException, Request, UploadFile, File, Form, BackgroundTasks
 from pydantic import BaseModel
-import logging
 
 from rag.firebase_storage_loader import _init_firebase_storage, PDF_METADATA
 from scripts.ingest_from_storage import ingest_from_firebase_storage
@@ -21,6 +26,13 @@ logger = logging.getLogger("mathpulse.admin")
 
 router = APIRouter(prefix="/api/admin", tags=["admin"])
 
+REINGESTION_STATUS: Dict[str, Any] = {
+    "status": "idle",
+    "last_run": None,
+    "message": None,
+    "mode": None
+}
+
 def require_admin(request: Request):
     user = getattr(request.state, "user", None)
     if user is None:
@@ -33,8 +45,161 @@ class ReingestRequest(BaseModel):
     subjectId: Optional[str] = None
     storagePath: Optional[str] = None
 
+def trigger_github_curriculum_workflow(token: str, ref: str = "main", force: bool = True) -> bool:
+    """Uses standard library urllib.request to POST to GitHub Actions workflow dispatch endpoint."""
+    url = "https://api.github.com/repos/Deign86/MATHPULSE-AI/actions/workflows/ingest-curriculum.yml/dispatches"
+    payload = {
+        "ref": ref,
+        "inputs": {
+            "force_reindex": force,
+            "upload_to_firebase": True,
+        }
+    }
+    data = json.dumps(payload).encode("utf-8")
+    headers = {
+        "Accept": "application/vnd.github+json",
+        "Authorization": f"Bearer {token}",
+        "X-GitHub-Api-Version": "2022-11-28",
+        "User-Agent": "MathPulseAI-Admin",
+        "Content-Type": "application/json",
+    }
+    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, timeout=10) as response:
+            if response.status == 204:
+                logger.info("Successfully triggered GitHub Actions curriculum ingestion workflow.")
+                return True
+            logger.warning(f"Unexpected status from GitHub Actions dispatch: {response.status}")
+            return False
+    except urllib.error.HTTPError as exc:
+        logger.error(f"GitHub Actions dispatch HTTP error {exc.code}: {exc.reason} - {exc.read().decode('utf-8', errors='ignore')}")
+        return False
+    except Exception as exc:
+        logger.error(f"Failed to trigger GitHub Actions workflow: {exc}")
+        return False
+
+def run_cloud_ingestion_and_upload():
+    """
+    Runs ingest_from_firebase_storage(force_reindex=True),
+    uploads vectorstore files to Firebase Storage via upload_directory and upload_vectorstore,
+    and updates REINGESTION_STATUS.
+    """
+    global REINGESTION_STATUS
+    REINGESTION_STATUS["status"] = "running"
+    REINGESTION_STATUS["last_run"] = datetime.now(timezone.utc).isoformat()
+    REINGESTION_STATUS["mode"] = "background_tasks"
+    REINGESTION_STATUS["message"] = "Remote re-ingestion from Firebase Storage in progress..."
+    logger.info("Starting background cloud curriculum reingestion and upload...")
+
+    try:
+        # Step 1: Run ingestion
+        ingest_from_firebase_storage(force_reindex=True)
+        logger.info("Curriculum ingestion from Firebase Storage completed successfully.")
+
+        upload_errors = []
+
+        # Step 2: Upload vectorstore directory via scripts.upload_vectorstore_to_firebase.upload_directory
+        try:
+            upload_dir_fn = None
+            init_storage_fn = None
+            vec_source_dir = None
+            remote_pfx = "vectorstore/"
+
+            try:
+                from scripts.upload_vectorstore_to_firebase import (
+                    upload_directory as _upload_dir,
+                    _init_firebase_storage as _init_storage,
+                    VECTORSTORE_SOURCE_DIR as _source_dir,
+                    REMOTE_PREFIX as _pfx,
+                )
+                upload_dir_fn = _upload_dir
+                init_storage_fn = _init_storage
+                vec_source_dir = _source_dir
+                remote_pfx = _pfx
+            except ImportError:
+                try:
+                    from backend.scripts.upload_vectorstore_to_firebase import (
+                        upload_directory as _upload_dir,
+                        _init_firebase_storage as _init_storage,
+                        VECTORSTORE_SOURCE_DIR as _source_dir,
+                        REMOTE_PREFIX as _pfx,
+                    )
+                    upload_dir_fn = _upload_dir
+                    init_storage_fn = _init_storage
+                    vec_source_dir = _source_dir
+                    remote_pfx = _pfx
+                except ImportError as imp_err:
+                    logger.warning(f"Could not import upload_vectorstore_to_firebase: {imp_err}")
+
+            if upload_dir_fn and init_storage_fn:
+                _, bucket = init_storage_fn()
+                if bucket is not None:
+                    if vec_source_dir is None:
+                        from pathlib import Path
+                        vec_source_dir = Path("datasets/vectorstore")
+                    uploaded, skipped = upload_dir_fn(vec_source_dir, bucket, remote_pfx)
+                    logger.info(f"Vectorstore directory uploaded: {uploaded} files uploaded, {skipped} skipped.")
+                else:
+                    logger.warning("Firebase Storage bucket not initialized; skipping upload_directory.")
+        except Exception as e:
+            logger.error(f"Error during vectorstore directory upload: {e}")
+            upload_errors.append(f"directory_upload: {e}")
+
+        # Step 3: Upload vectorstore archive via scripts.upload_vectorstore.upload_vectorstore
+        try:
+            upload_vec_fn = None
+            try:
+                from scripts.upload_vectorstore import upload_vectorstore as _upload_vec
+                upload_vec_fn = _upload_vec
+            except ImportError:
+                try:
+                    from backend.scripts.upload_vectorstore import upload_vectorstore as _upload_vec
+                    upload_vec_fn = _upload_vec
+                except ImportError:
+                    import sys
+                    from pathlib import Path
+                    repo_root = str(Path(__file__).resolve().parents[2])
+                    if repo_root not in sys.path:
+                        sys.path.insert(0, repo_root)
+                    try:
+                        from scripts.upload_vectorstore import upload_vectorstore as _upload_vec
+                        upload_vec_fn = _upload_vec
+                    except ImportError as imp_err:
+                        logger.warning(f"Could not import upload_vectorstore: {imp_err}")
+
+            if upload_vec_fn:
+                success = upload_vec_fn()
+                if success:
+                    logger.info("Vectorstore archive upload completed successfully.")
+                else:
+                    logger.warning("Vectorstore archive upload returned False.")
+                    upload_errors.append("zip_upload_failed")
+        except Exception as e:
+            logger.error(f"Error during vectorstore archive upload: {e}")
+            upload_errors.append(f"zip_upload: {e}")
+
+        if upload_errors:
+            REINGESTION_STATUS["status"] = "completed_with_warnings"
+            REINGESTION_STATUS["message"] = f"Ingestion finished, but upload had warnings: {'; '.join(upload_errors)}"
+        else:
+            REINGESTION_STATUS["status"] = "completed"
+            REINGESTION_STATUS["message"] = "Remote re-ingestion and vectorstore upload completed successfully."
+        logger.info("Cloud curriculum reingestion process finished.")
+
+    except Exception as exc:
+        logger.error(f"Background cloud reingestion failed: {exc}", exc_info=True)
+        REINGESTION_STATUS["status"] = "failed"
+        REINGESTION_STATUS["message"] = f"Re-ingestion failed: {str(exc)}"
+
+_run_reingestion_task = run_cloud_ingestion_and_upload
+
+@router.get("/reingest-status")
+async def get_reingest_status(_admin=Depends(require_admin)):
+    return REINGESTION_STATUS
+
 @router.post("/upload-pdf")
 async def upload_pdf(
+    background_tasks: BackgroundTasks,
     subjectId: str = Form(...),
     subjectName: str = Form(...),
     semester: int = Form(...),
@@ -71,9 +236,9 @@ async def upload_pdf(
         "quarter": quarter
     }
     
-    # Reingest
+    # Reingest in background
     try:
-        ingest_from_firebase_storage(force_reindex=True)
+        background_tasks.add_task(run_cloud_ingestion_and_upload)
     except Exception as e:
         logger.error(f"Failed to trigger reingestion: {e}")
         
@@ -102,19 +267,6 @@ async def upload_pdf(
         "storageUrl": storage_url
     }
 
-def _run_reingestion_task():
-    try:
-        logger.info("Starting background curriculum reingestion from Firebase Storage...")
-        ingest_from_firebase_storage(force_reindex=True)
-        from scripts.upload_vectorstore_to_firebase import upload_directory, _init_firebase_storage, VECTORSTORE_SOURCE_DIR, REMOTE_PREFIX
-        _, bucket = _init_firebase_storage()
-        if bucket is not None:
-            upload_directory(VECTORSTORE_SOURCE_DIR, bucket, REMOTE_PREFIX)
-        logger.info("Background curriculum reingestion complete.")
-    except Exception as exc:
-        logger.error(f"Background reingestion failed: {exc}")
-
-
 @router.post("/reingest-pdf")
 async def reingest_pdf(
     background_tasks: BackgroundTasks,
@@ -122,10 +274,46 @@ async def reingest_pdf(
     _admin=Depends(require_admin)
 ):
     try:
-        background_tasks.add_task(_run_reingestion_task)
-        import asyncio
+        github_token = os.getenv("GITHUB_PAT") or os.getenv("GITHUB_TOKEN")
+        dispatched_gh = False
+        if github_token:
+            dispatched_gh = trigger_github_curriculum_workflow(token=github_token)
+
         audit_fn = _get_audit_logger()
+
+        if dispatched_gh:
+            REINGESTION_STATUS["status"] = "running"
+            REINGESTION_STATUS["mode"] = "github_actions"
+            REINGESTION_STATUS["last_run"] = datetime.now(timezone.utc).isoformat()
+            REINGESTION_STATUS["message"] = "Remote re-ingestion dispatched to GitHub Actions runner."
+
+            if audit_fn:
+                import asyncio
+                asyncio.create_task(audit_fn(
+                    action="REINGEST_RAG_KNOWLEDGE",
+                    actor_uid=_admin.uid,
+                    actor_name=_admin.name if hasattr(_admin, "name") else "Unknown",
+                    actor_email=_admin.email if hasattr(_admin, "email") else "",
+                    actor_role=_admin.role,
+                    description="Triggered remote cloud reingestion of the RAG knowledge base via GitHub Actions",
+                    route="/api/admin/reingest-pdf",
+                    module="admin"
+                ))
+
+            return {
+                "success": True,
+                "message": "Remote re-ingestion dispatched to GitHub Actions runner.",
+                "execution_mode": "github_actions"
+            }
+
+        REINGESTION_STATUS["status"] = "running"
+        REINGESTION_STATUS["mode"] = "background_tasks"
+        REINGESTION_STATUS["last_run"] = datetime.now(timezone.utc).isoformat()
+        REINGESTION_STATUS["message"] = "Remote re-ingestion started in the cloud."
+        background_tasks.add_task(run_cloud_ingestion_and_upload)
+
         if audit_fn:
+            import asyncio
             asyncio.create_task(audit_fn(
                 action="REINGEST_RAG_KNOWLEDGE",
                 actor_uid=_admin.uid,
@@ -136,7 +324,12 @@ async def reingest_pdf(
                 route="/api/admin/reingest-pdf",
                 module="admin"
             ))
-        return {"success": True, "message": "Remote re-ingestion started in background. Check RAG Manager in a few minutes."}
+
+        return {
+            "success": True,
+            "message": "Remote re-ingestion started in the cloud.",
+            "execution_mode": "background_tasks"
+        }
     except Exception as e:
         logger.error(f"Failed to trigger reingestion: {e}")
         raise HTTPException(status_code=500, detail=f"Failed to trigger reingestion: {e}")

routes/at_risk_resolution.py

@@ -13,7 +13,7 @@ from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
 from services.ai_client import CHAT_MODEL
-from services.deepseek_client import is_enabled, rag_grounded_completion, parse_json_response
+from services.inference_client import is_enabled, rag_grounded_completion, parse_json_response
 from rag.curriculum_rag import (
     retrieve_curriculum_context,
     format_retrieved_chunks,

routes/class_analytics_routes.py

@@ -53,7 +53,7 @@ async def get_class_students(
         )[:10]
     elif filter == "needs_attention":
         students = sorted(
-            [s for s in students if s.risk_level in ("High Risk", "Critical", "Unassessed")],
+            [s for s in students if s.risk_level in ("intervene", "critical", "at_risk")],
             key=lambda s: s.avg_score,
         )
 

routes/class_records_router.py

@@ -692,14 +692,23 @@ async def get_upload_students(
     upload_data = upload_snap.to_dict()
     section_id = (upload_data.get("section") or "unknown").replace(" ", "_").lower()
 
-    students_ref = client.collection("classRecords").document(teacher_uid).collection("sections").document(section_id).collection("students")
+    students_ref = upload_ref.collection("students")
     query_ref = students_ref.limit(limit + 1)
     if after:
         after_doc = students_ref.document(after).get()
         if after_doc.exists:
             query_ref = query_ref.start_after(after_doc)
 
-    docs = query_ref.stream()
+    docs = list(query_ref.stream())
+    if not docs:
+        legacy_ref = client.collection("classRecords").document(teacher_uid).collection("sections").document(section_id).collection("students")
+        query_ref = legacy_ref.limit(limit + 1)
+        if after:
+            after_doc = legacy_ref.document(after).get()
+            if after_doc.exists:
+                query_ref = query_ref.start_after(after_doc)
+        docs = list(query_ref.stream())
+
     students: List[Dict[str, Any]] = []
     for d in docs:
         data = d.to_dict()
@@ -739,8 +748,10 @@ async def generate_ai_class_report(request: Request, uploadId: str):
     section_id = (upload_data.get("section") or "unknown").replace(" ", "_").lower()
     metadata = upload_data.get("metadata", {})
 
-    students_ref = client.collection("classRecords").document(teacher_uid).collection("sections").document(section_id).collection("students")
-    docs = students_ref.stream()
+    students_ref = upload_ref.collection("students")
+    docs = list(students_ref.stream())
+    if not docs:
+        docs = list(client.collection("classRecords").document(teacher_uid).collection("sections").document(section_id).collection("students").stream())
     all_students = [d.to_dict() for d in docs if d.to_dict()]
 
     total = len(all_students)
@@ -991,24 +1002,28 @@ def _trigger_wri_recompute(
 
         try:
             student_ref = section_ref.collection("students").document(lrn)
-            student_ref.set({
-                "wriScore": wri_value,
+            student_data: Dict[str, Any] = {
                 "wriRiskBand": risk_status,
                 "wriComputedAt": datetime.utcnow().isoformat(),
-            }, merge=True)
+            }
+            if wri_value is not None:
+                student_data["wriScore"] = wri_value
+            student_ref.set(student_data, merge=True)
         except Exception as e:
             logger.error(f"Failed to write WRI to classRecords for LRN {lrn}: {e}")
 
         try:
             matched = users_ref.where("lrn", "==", lrn).limit(1).stream()
             for user_doc in matched:
-                user_doc.reference.set({
+                user_payload: Dict[str, Any] = {
                     "latestGrade": transmuted,
                     "wriExternalGrade": transmuted,
-                    "wriScore": wri_value,
-                    "wriRiskBand": risk_status,
                     "wriUpdatedAt": datetime.utcnow().isoformat(),
-                }, merge=True)
+                }
+                if wri_value is not None:
+                    user_payload["wriScore"] = wri_value
+                    user_payload["wriRiskBand"] = risk_status
+                user_doc.reference.set(user_payload, merge=True)
                 logger.info(f"LRN {lrn} matched to user {user_doc.id}: grade={transmuted}, WRI={wri_value}")
                 break
         except Exception as e:
@@ -1026,7 +1041,7 @@ def _persist_students(
         return False
 
     batch = client.batch()
-    section_ref = client.collection("classRecords").document(teacher_uid).collection("sections").document(section_id)
+    upload_ref = client.collection("classRecords").document(teacher_uid).collection("uploads").document(upload_id)
     count = 0
 
     try:
@@ -1057,7 +1072,7 @@ def _persist_students(
                 "updatedAt": datetime.utcnow().isoformat(),
             }
 
-            student_ref = section_ref.collection("students").document(lrn)
+            student_ref = upload_ref.collection("students").document(lrn)
             batch.set(student_ref, student_doc, merge=True)
             count += 1
 
@@ -1076,7 +1091,7 @@ def _persist_students(
                 logger.error(f"Final batch commit failed: {e}")
                 return False
 
-        logger.info(f"Persisted {count} students to classRecords/{teacher_uid}/sections/{section_id}")
+        logger.info(f"Persisted {count} students to classRecords/{teacher_uid}/uploads/{upload_id}/students")
         return True
     except Exception as e:
         logger.error(f"Failed to persist students: {e}")

routes/deepseek_rag_routes.py

@@ -13,7 +13,7 @@ from pydantic import BaseModel, Field
 from fastapi import APIRouter
 
 from services.ai_client import REASONER_MODEL, CHAT_MODEL
-from services.deepseek_client import is_enabled, rag_grounded_completion, parse_json_response
+from services.inference_client import is_enabled, rag_grounded_completion, parse_json_response
 from rag.curriculum_rag import (
     retrieve_curriculum_context,
     build_analysis_curriculum_context,

routes/pipeline_routes.py

@@ -100,7 +100,7 @@ async def recompute_profile(student_id: str, background_tasks: BackgroundTasks,
         # Trigger a synthetic diagnostic event to force full recompute
         event = StudentActivityEvent(
             student_id=student_id,
-            event_type="session",
+            event_type="force_recompute",
             event_data={"event": "force_recompute"},
             occurred_at=__import__("datetime").datetime.now(__import__("datetime").timezone.utc).isoformat(),
             class_id="",

services/class_analytics_engine.py

@@ -46,7 +46,7 @@ class StudentAnalyticsSummary(BaseModel):
     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"
+    risk_level: Literal["safe", "watch", "intervene", "critical", "at_risk", "pending_assessment"] = "pending_assessment"
     engagement_level: Literal["Low", "Medium", "High"] = "Low"
     weakest_topic: Optional[str] = None
     accuracy_by_topic: Dict[str, float] = Field(default_factory=dict)
@@ -188,6 +188,7 @@ class ClassAnalyticsEngine:
         topic_perf = self._compute_topic_performance(student_summaries)
 
         # Risk distribution
+        from services.wri_service import normalize_risk_band
         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
@@ -198,7 +199,7 @@ class ClassAnalyticsEngine:
                     stored_status = ms_doc.to_dict().get("riskStatus")
             except Exception:
                 pass
-            status = stored_status if stored_status in risk_dist else s.risk_level
+            status = normalize_risk_band(stored_status) if stored_status else s.risk_level
             risk_dist[status] = risk_dist.get(status, 0) + 1
 
         # Generate AI insights
@@ -442,11 +443,12 @@ 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
+- Safe: {risk_dist.get('safe', 0)} students
+- Watch: {risk_dist.get('watch', 0)} students
+- Intervene: {risk_dist.get('intervene', 0)} students
+- Critical: {risk_dist.get('critical', 0)} students
+- At Risk: {risk_dist.get('at_risk', 0)} students
+- Pending Assessment: {risk_dist.get('pending_assessment', 0)} students
 
 Topic Performance (class accuracy):
 {topic_lines}
@@ -501,12 +503,12 @@ Be specific to Filipino K-12 DepEd context. If data is limited, acknowledge it a
             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.",
+                class_summary=f"Class has {student_count} students with an average score of {class_average:.0f}%. {risk_dist.get('pending_assessment', 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.",
+                    "Schedule one-on-one check-ins with critical / at-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,

services/deepseek_client.py

@@ -1,87 +0,0 @@
-"""
-RAG-grounded DeepSeek client wrapper.
-
-All calls go through `rag_grounded_completion()` which enforces:
-- DEEPSEEK_ENABLED feature flag check
-- Retry with exponential backoff on 429
-- Token usage logging
-"""
-
-import os
-import time
-import json
-import logging
-from typing import Optional
-
-from services.ai_client import get_deepseek_client, CHAT_MODEL, REASONER_MODEL, RateLimitError
-
-logger = logging.getLogger(__name__)
-
-DEEPSEEK_ENABLED = os.getenv("DEEPSEEK_ENABLED", "true").lower() in ("true", "1", "yes")
-MAX_RETRIES = 3
-BACKOFF_DELAYS = [2, 4, 8]
-
-
-def is_enabled() -> bool:
-    return DEEPSEEK_ENABLED
-
-
-def rag_grounded_completion(
-    model: str,
-    system_prompt: str,
-    user_prompt: str,
-    temperature: float = 0.2,
-) -> Optional[str]:
-    """
-    Call DeepSeek with retry on 429. Returns response text or None if disabled/failed.
-    Logs token usage per call.
-    """
-    if not DEEPSEEK_ENABLED:
-        logger.info("[DEEPSEEK] Disabled via DEEPSEEK_ENABLED flag, skipping.")
-        return None
-
-    client = get_deepseek_client()
-
-    for attempt in range(MAX_RETRIES):
-        try:
-            response = client.chat.completions.create(
-                model=model,
-                messages=[
-                    {"role": "system", "content": system_prompt},
-                    {"role": "user", "content": user_prompt},
-                ],
-                temperature=temperature,
-            )
-            usage = response.usage
-            if usage:
-                logger.info(
-                    "[DEEPSEEK] model=%s prompt_tokens=%d completion_tokens=%d total=%d",
-                    model, usage.prompt_tokens, usage.completion_tokens, usage.total_tokens,
-                )
-            return response.choices[0].message.content or ""
-        except RateLimitError:
-            delay = BACKOFF_DELAYS[attempt] if attempt < len(BACKOFF_DELAYS) else 8
-            logger.warning("[DEEPSEEK] 429 rate limited, retry %d/%d in %ds", attempt + 1, MAX_RETRIES, delay)
-            time.sleep(delay)
-        except Exception as e:
-            logger.error("[DEEPSEEK] Call failed: %s", e)
-            return None
-
-    logger.error("[DEEPSEEK] All %d retries exhausted.", MAX_RETRIES)
-    return None
-
-
-def parse_json_response(text: Optional[str]) -> Optional[dict]:
-    """Attempt to parse JSON from DeepSeek response, handling markdown fences."""
-    if not text:
-        return None
-    cleaned = text.strip()
-    if cleaned.startswith("```"):
-        lines = cleaned.split("\n")
-        lines = [l for l in lines if not l.strip().startswith("```")]
-        cleaned = "\n".join(lines)
-    try:
-        return json.loads(cleaned)
-    except json.JSONDecodeError:
-        logger.warning("[DEEPSEEK] Failed to parse JSON response")
-        return None

services/inference_client.py

@@ -675,30 +675,6 @@ class InferenceClient:
             return self.interactive_timeout_sec
         return self.background_timeout_sec
 
-    def _messages_to_prompt(self, messages: List[Dict[str, str]]) -> str:
-        parts: List[str] = []
-        for msg in messages:
-            role = (msg.get("role") or "user").strip().lower()
-            content = (msg.get("content") or "").strip()
-            if not content or role in {"tool", "function"}:
-                continue
-            prefix = "USER"
-            if role == "system":
-                prefix = "SYSTEM"
-            elif role == "assistant":
-                prefix = "ASSISTANT"
-            parts.append(f"{prefix}:\n{content}")
-        parts.append("ASSISTANT:")
-        return "\n\n".join(parts)
-
-    def _latest_user_message(self, messages: List[Dict[str, str]]) -> str:
-        for msg in reversed(messages):
-            role = (msg.get("role") or "").strip().lower()
-            content = (msg.get("content") or "").strip()
-            if role == "user" and content:
-                return content
-        return self._messages_to_prompt(messages)
-
     def _call_deepseek(self, req: InferenceRequest, fallback_depth: int) -> str:
         """Call DeepSeek API with OpenAI-compatible chat completions."""
         if not self.ds_api_key:
@@ -769,12 +745,6 @@ class InferenceClient:
                     fallback_depth=fallback_depth,
                     route=route,
                 )
-                self._record_attempt(
-                    task_type=task_type,
-                    provider="deepseek",
-                    route=route,
-                    fallback_depth=fallback_depth,
-                )
                 self._record_completion(latency_ms=latency_ms)
                 self._bump_metric("requests_ok", 1)
                 return text
@@ -881,155 +851,75 @@ class InferenceClient:
 
         raise RuntimeError(f"DeepSeek call failed after {max_retries} attempts")
 
-    def _call_local_space(self, req: InferenceRequest, *, provider: str, route: str, fallback_depth: int) -> str:
-        target_model = req.model or self.default_model
-        url = f"{self.local_space_url.rstrip('/')}{self.local_generate_path}"
-
-        prompt = self._messages_to_prompt(req.messages)
-        payload: Dict[str, object] = {
-            "data": [
-                prompt,
-                [],
-                req.temperature,
-                req.top_p,
-                req.max_new_tokens,
-            ]
-        }
-        headers = {"Content-Type": "application/json"}
 
-        timeout = self._timeout_for(req, provider)
+DEEPSEEK_ENABLED = os.getenv("DEEPSEEK_ENABLED", "true").lower() in ("true", "1", "yes")
+_MAX_RETRIES = 3
+_BACKOFF_DELAYS = [2, 4, 8]
 
-        self._record_attempt(
-            task_type=req.task_type,
-            provider=provider,
-            route=route,
-            fallback_depth=fallback_depth,
-        )
-        start = time.perf_counter()
 
-        try:
-            resp = requests.post(url, headers=headers, json=payload, timeout=timeout)
-        except Exception as exc:
-            latency_ms = (time.perf_counter() - start) * 1000
-            log_model_call(
-                LOGGER,
-                provider=provider,
-                model=target_model,
-                endpoint=url,
-                latency_ms=latency_ms,
-                input_tokens=None,
-                output_tokens=None,
-                status="error",
-                error_class=exc.__class__.__name__,
-                error_message=str(exc),
-                task_type=req.task_type,
-                request_tag=req.request_tag,
-                retry_attempt=1,
-                fallback_depth=fallback_depth,
-                route=route,
-            )
-            self._bump_metric("requests_error", 1)
-            raise
-
-        latency_ms = (time.perf_counter() - start) * 1000
-        self._bump_bucket("status_code_counts", str(resp.status_code), 1)
-
-        if resp.status_code != 200:
-            self._bump_metric("requests_error", 1)
-            raise RuntimeError(f"Local Space error {resp.status_code}: {resp.text}")
-
-        data = resp.json()
-        event_id = data.get("event_id")
-        if not event_id:
-            return self._extract_text(data)
-
-        result_url = f"{self.local_space_url.rstrip('/')}/gradio_api/call/generate/{event_id}"
-        result_resp = requests.get(result_url, timeout=req.timeout_sec or self.local_timeout_sec)
-        if result_resp.status_code != 200:
-            raise RuntimeError(f"Local Space result error {result_resp.status_code}: {result_resp.text}")
-
-        line_data = None
-        for line in result_resp.text.splitlines():
-            if line.startswith("data:"):
-                line_data = line.split("data:", 1)[1].strip()
-
-        if not line_data:
-            raise RuntimeError("Local Space result stream missing data")
-
-        parsed = json.loads(line_data)
-        output_payload = parsed if isinstance(parsed, dict) else {"data": parsed}
-        text = self._extract_text(output_payload)
-        log_model_call(
-            LOGGER,
-            provider=provider,
-            model=target_model,
-            endpoint=url,
-            latency_ms=latency_ms,
-            input_tokens=None,
-            output_tokens=None,
-            status="ok",
-            task_type=req.task_type,
-            request_tag=req.request_tag,
-            retry_attempt=1,
-            fallback_depth=fallback_depth,
-            route=route,
-        )
-        self._bump_metric("requests_ok", 1)
-        return text
-
-    def _extract_text(self, data: object) -> str:
-        """Extract clean text from inference response, stripping JSON artifacts."""
-        if isinstance(data, list) and data:
-            first = data[0]
-            if isinstance(first, dict):
-                val = (first.get("generated_text") or "").strip()
-                if val:
-                    return self._clean_response_text(val)
-
-        if isinstance(data, dict):
-            direct = (data.get("generated_text") or "").strip()
-            if direct:
-                return self._clean_response_text(direct)
-
-            choices = data.get("choices", [])
-            if choices:
-                message = choices[0].get("message", {})
-                msg = (message.get("content") or "").strip()
-                if msg:
-                    return self._clean_response_text(msg)
-                reasoning = (message.get("reasoning") or "").strip()
-                if reasoning:
-                    return self._clean_response_text(reasoning)
+def is_enabled() -> bool:
+    return DEEPSEEK_ENABLED
 
-            generic_data = data.get("data")
-            if isinstance(generic_data, list) and generic_data:
-                first = generic_data[0]
-                if isinstance(first, str) and first.strip():
-                    return self._clean_response_text(first.strip())
 
-        raise RuntimeError(f"Unexpected inference response format: {data}")
+def rag_grounded_completion(
+    model: str,
+    system_prompt: str,
+    user_prompt: str,
+    temperature: float = 0.2,
+) -> Optional[str]:
+    """Call DeepSeek with retry on 429. Returns response text or None if disabled/failed."""
+    if not is_enabled():
+        LOGGER.info("[DEEPSEEK] Disabled via DEEPSEEK_ENABLED flag, skipping.")
+        return None
 
-    def _clean_response_text(self, text: str) -> str:
-        """Strip JSON braces, template artifacts, and whitespace from response text."""
-        text = text.strip()
+    client = get_deepseek_client()
 
-        if text.startswith("{") and text.endswith("}"):
-            try:
-                parsed = json.loads(text)
-                if isinstance(parsed, dict):
-                    if "content" in parsed:
-                        text = str(parsed["content"]).strip()
-                    elif "text" in parsed:
-                        text = str(parsed["text"]).strip()
-            except json.JSONDecodeError:
-                text = text.strip("{}")
-
-        if text.startswith("```json") or text.startswith("```"):
-            text = re.sub(r"^```(?:json)?", "", text).strip()
-        if text.endswith("```"):
-            text = text[:-3].strip()
-
-        return text.strip()
+    for attempt in range(_MAX_RETRIES):
+        try:
+            response = client.chat.completions.create(
+                model=model,
+                messages=[
+                    {"role": "system", "content": system_prompt},
+                    {"role": "user", "content": user_prompt},
+                ],
+                temperature=temperature,
+            )
+            usage = response.usage
+            if usage:
+                LOGGER.info(
+                    "[DEEPSEEK] model=%s prompt_tokens=%d completion_tokens=%d total=%d",
+                    model,
+                    usage.prompt_tokens,
+                    usage.completion_tokens,
+                    usage.total_tokens,
+                )
+            return response.choices[0].message.content or ""
+        except RateLimitError:
+            delay = _BACKOFF_DELAYS[attempt] if attempt < len(_BACKOFF_DELAYS) else 8
+            LOGGER.warning("[DEEPSEEK] 429 rate limited, retry %d/%d in %ds", attempt + 1, _MAX_RETRIES, delay)
+            time.sleep(delay)
+        except Exception as e:
+            LOGGER.error("[DEEPSEEK] Call failed: %s", e)
+            return None
+
+    LOGGER.error("[DEEPSEEK] All %d retries exhausted.", _MAX_RETRIES)
+    return None
+
+
+def parse_json_response(text: Optional[str]) -> Optional[dict]:
+    """Attempt to parse JSON from DeepSeek response, handling markdown fences."""
+    if not text:
+        return None
+    cleaned = text.strip()
+    if cleaned.startswith("```"):
+        lines = cleaned.split("\n")
+        lines = [l for l in lines if not l.strip().startswith("```")]
+        cleaned = "\n".join(lines)
+    try:
+        return json.loads(cleaned)
+    except json.JSONDecodeError:
+        LOGGER.warning("[DEEPSEEK] Failed to parse JSON response")
+        return None
 
 
 def create_default_client(firestore_client: Optional[Any] = None) -> InferenceClient:

services/intervention_engine.py

@@ -67,7 +67,7 @@ class InterventionPlan(BaseModel):
     student_name: str = ""
     grade_level: str = ""
     section: str = ""
-    risk_level: Literal["Low Risk", "Medium Risk", "High Risk", "Critical", "Unassessed"] = "Unassessed"
+    risk_level: str = "pending_assessment"
     avg_score: float = 0.0
     engagement_level: Literal["Low", "Medium", "High"] = "Low"
     last_active: Optional[str] = None
@@ -83,17 +83,19 @@ class InterventionPlan(BaseModel):
 
 # ─── Risk & Engagement Classification ─────────────────────────────────────
 
-def _classify_risk(avg_score: float, quiz_count: int, days_since_active: Optional[int]) -> str:
+def _classify_risk(avg_score: float, quiz_count: int, days_since_active: Optional[int] = None) -> str:
+    """Canonical 5-band DepEd classification when WRI is not directly available."""
     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"
+        return "pending_assessment"
+    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, lessons_completed: int = 0) -> str:
@@ -127,9 +129,18 @@ class InterventionEngine:
             logger.error("Firestore client unavailable")
             return InterventionPlan(student_id=student_id, generated_at=_now_iso())
 
-        # Fetch student data from managedStudents
+        # Check student_profiles directly first to minimize collection roundtrips
+        profile_data = None
+        try:
+            pdoc = db.collection("student_profiles").document(student_id).get()
+            if pdoc.exists:
+                profile_data = pdoc.to_dict()
+        except Exception as e:
+            logger.debug(f"Error reading student_profiles/{student_id}: {e}")
+
+        # Fetch student data from managedStudents if needed
         student_data = await self._fetch_student_data(db, student_id)
-        if not student_data:
+        if not student_data and not profile_data:
             return InterventionPlan(
                 student_id=student_id,
                 student_name="Unknown",
@@ -138,67 +149,107 @@ class InterventionEngine:
                 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)
+        if not student_data:
+            student_data = {
+                "id": student_id,
+                "name": profile_data.get("display_name") or profile_data.get("name", "Unknown"),
+                "gradeLevel": profile_data.get("grade_level", profile_data.get("grade", "11")),
+                "section": profile_data.get("section", ""),
+                **profile_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))
-
-        # Fetch lessons completed from progress doc
-        lessons_completed = 0
-        for lookup_id in [student_id, student_data.get("accountUid")]:
-            if not lookup_id:
-                continue
-            try:
-                pdoc = db.collection("progress").document(lookup_id).get()
-                if pdoc.exists:
-                    lessons_completed = pdoc.to_dict().get("totalLessonsCompleted", 0)
-                    break
-            except Exception:
-                pass
+        qp = (profile_data or {}).get("quiz_performance", {})
+        diag = (profile_data or {}).get("diagnostic", {})
+        ce = (profile_data or {}).get("content_engagement", {})
+        eng = (profile_data or {}).get("engagement", {})
+
+        # If profile_data already has quiz performance, use it directly to save roundtrips
+        if qp.get("total_attempts", 0) > 0 or qp.get("accuracy_by_topic"):
+            quiz_count = qp.get("total_attempts", 0)
+            avg_score = float(qp.get("avg_score_all_time") or profile_data.get("system_performance_avg") or 0.0)
+            topic_avgs = {t: float(s) for t, s in qp.get("accuracy_by_topic", {}).items()}
+            weak_topics = [t for t, s in sorted(topic_avgs.items(), key=lambda x: x[1]) if s < 70][:5]
+            if not weak_topics and qp.get("lowest_accuracy_topics"):
+                weak_topics = qp.get("lowest_accuracy_topics")[:5]
+            strong_topics = [t for t, s in topic_avgs.items() if s >= 70]
+            if not strong_topics and qp.get("highest_accuracy_topics"):
+                strong_topics = qp.get("highest_accuracy_topics")[:3]
+            weakest_topic = weak_topics[0] if weak_topics else student_data.get("weakestTopic", "Foundational Skills")
+            lessons_completed = ce.get("lessons_completed", 0)
+            days_since_active = eng.get("days_since_last_active")
+            last_active_str = eng.get("last_active_at")
+            recent_count = len(qp.get("recent_attempts", []))
+        else:
+            # Fallback: fetch quiz attempts and progress doc
+            quiz_attempts = await self._fetch_quiz_attempts(db, student_id, student_data)
+            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_count = sum(1 for q in quiz_attempts if self._is_recent(q, now, 14))
+
+            lessons_completed = 0
+            for lookup_id in [student_id, student_data.get("accountUid")]:
+                if not lookup_id:
+                    continue
+                try:
+                    pdoc = db.collection("progress").document(lookup_id).get()
+                    if pdoc.exists:
+                        lessons_completed = pdoc.to_dict().get("totalLessonsCompleted", 0)
+                        break
+                except Exception:
+                    pass
+
+        # Canonical risk classification: read canonical risk_status or wriRiskBand
+        raw_risk = (
+            (profile_data or {}).get("risk_status")
+            or (profile_data or {}).get("wriRiskBand")
+            or student_data.get("riskStatus")
+            or student_data.get("wriRiskBand")
+        )
+        if raw_risk:
+            from services.wri_service import normalize_risk_band
+            risk_level = normalize_risk_band(raw_risk)
+        else:
+            risk_level = _classify_risk(avg_score, quiz_count, days_since_active)
 
-        risk_level = _classify_risk(avg_score, quiz_count, days_since_active)
         engagement = _classify_engagement(days_since_active, recent_count, lessons_completed)
 
+
         # Generate AI insights
         insights = await self._generate_insights(
             grade_level=student_data.get("gradeLevel", student_data.get("grade", "11")),
@@ -405,7 +456,7 @@ Return as JSON:
         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
+        estimated_days = 5 if risk_level == "critical" else 7
 
         prompt = f"""Create a personalized intervention learning path for a Filipino K-12 math student.
 

services/question_bank_service.py

@@ -30,10 +30,18 @@ async def get_questions_for_battle(
 
     Uses Firestore random_seed field for pseudo-random ordering.
     If fewer than `count` questions exist, returns all available.
+
+    Firestore path: question_bank/{grade_level}/topics/{topic}/questions/{docId}
+    (alternating collection/document segments — 5 segments total).
     """
     db = _get_db()
-    collection_path = f"question_bank/{grade_level}/{topic}/questions"
-    collection_ref = db.collection(collection_path)
+    collection_ref = (
+        db.collection("question_bank")
+        .document(str(grade_level))
+        .collection("topics")
+        .document(topic)
+        .collection("questions")
+    )
 
     # Pseudo-random query using random_seed >= random threshold
     threshold = random.random()
@@ -57,13 +65,8 @@ async def get_questions_for_battle(
         docs.extend(list(fallback_query.stream()))
 
     questions = [doc.to_dict() for doc in docs]
-    # Ensure all required fields are present
-    valid_questions = []
-    for q in questions:
-        if q and all(k in q for k in ("question", "choices", "correct_answer", "difficulty")):
-            valid_questions.append(q)
-
-    return valid_questions
+    required_fields = ("question", "choices", "correct_answer", "difficulty")
+    return [q for q in questions if q and all(k in q for k in required_fields)]
 
 
 async def cache_session_questions(
@@ -73,25 +76,22 @@ async def cache_session_questions(
     grade_level: int,
     topic: str,
 ) -> None:
-    """Cache varied questions for a battle session with 24-hour TTL."""
+    """Cache varied questions for a battle session with 24-hour TTL.
+
+    Questions are inlined as a list field in the root session document
+    (1 Firestore write vs. N subcollection batch writes).
+    """
     db = _get_db()
     session_ref = db.collection("quiz_battle_sessions").document(session_id)
-
     session_ref.set({
         "player_ids": player_ids,
         "grade_level": grade_level,
         "topic": topic,
+        "questions": questions,
         "created_at": firestore.SERVER_TIMESTAMP,
         "variance_cached_until": datetime.now(timezone.utc) + timedelta(hours=24),
     })
 
-    # Write questions to subcollection
-    batch = db.batch()
-    for idx, q in enumerate(questions):
-        q_ref = session_ref.collection("questions").document(str(idx))
-        batch.set(q_ref, q)
-    batch.commit()
-
 
 async def get_cached_session(session_id: str) -> Optional[List[Dict]]:
     """
@@ -115,9 +115,7 @@ async def get_cached_session(session_id: str) -> Optional[List[Dict]]:
             cached_until = datetime.fromtimestamp(cached_until.timestamp(), tz=timezone.utc)
 
         if cached_until > datetime.now(timezone.utc):
-            # Return cached questions
-            q_docs = db.collection("quiz_battle_sessions").document(session_id).collection("questions").stream()
-            questions = [doc.to_dict() for doc in q_docs]
+            questions: List[Dict] = data.get("questions") or []
             return questions if questions else None
 
     return None

services/student_intelligence_pipeline.py

@@ -43,7 +43,7 @@ def _get_db():
 
 class StudentActivityEvent(BaseModel):
     student_id: str
-    event_type: Literal["diagnostic", "quiz", "battle", "lesson", "module", "session"]
+    event_type: Literal["diagnostic", "quiz", "battle", "lesson", "module", "session", "force_recompute", "backfill"]
     event_data: Dict[str, Any] = Field(default_factory=dict)
     occurred_at: str  # ISO string
     class_id: str = ""
@@ -314,6 +314,17 @@ class StudentIntelligencePipeline:
             eng["last_active_at"] = event.occurred_at
             eng["days_since_last_active"] = 0
 
+        elif event.event_type in ("force_recompute", "backfill"):
+            if "diagnostic_score" in ed:
+                profile["diagnostic_score"] = ed["diagnostic_score"]
+                profile.setdefault("diagnostic", {})["overall_score"] = ed["diagnostic_score"]
+            if "external_grades_avg" in ed:
+                profile["external_grades_avg"] = ed["external_grades_avg"]
+            if "system_performance_avg" in ed:
+                profile["system_performance_avg"] = ed["system_performance_avg"]
+            if "wri_weights" in ed:
+                profile["wri_weights"] = ed["wri_weights"]
+
     # ─── P computation ─────────────────────────────────────────────────────
 
     def _compute_system_performance_avg(
@@ -438,8 +449,8 @@ class StudentIntelligencePipeline:
     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 event.event_type in ("diagnostic", "force_recompute", "backfill"):
+            return bool(result.risk_status_changed or not profile.get("ai_context", {}).get("generated_at"))
         if result.risk_status_changed:
             return True
         ai_ctx = profile.get("ai_context", {})

services/wri_service.py

@@ -23,6 +23,34 @@ from typing import Optional
 DEFAULT_WEIGHTS = {"w1": 0.30, "w2": 0.40, "w3": 0.30}
 WEIGHT_TOLERANCE = 0.001
 
+CANONICAL_RISK_BANDS = ("safe", "watch", "intervene", "critical", "at_risk")
+CANONICAL_RISK_STATUSES = ("safe", "watch", "intervene", "critical", "at_risk", "pending_assessment")
+
+
+def normalize_risk_band(band: Optional[str]) -> str:
+    """Normalize legacy or mixed-case risk status strings to canonical 5-band WRI format."""
+    if not band:
+        return "pending_assessment"
+    normalized = band.strip().lower().replace("-", "_").replace(" ", "_")
+    if normalized in CANONICAL_RISK_STATUSES:
+        return normalized
+    tier_map = {
+        "low": "safe",
+        "low_risk": "safe",
+        "on_track": "safe",
+        "moderate": "watch",
+        "medium": "watch",
+        "medium_risk": "watch",
+        "high": "intervene",
+        "high_risk": "intervene",
+        "urgent": "critical",
+        "failing": "at_risk",
+        "unassessed": "pending_assessment",
+        "pending": "pending_assessment",
+    }
+    return tier_map.get(normalized, "pending_assessment")
+
+
 
 def compute_wri(
     d: Optional[float],

tests/test_admin_reingest.py

@@ -0,0 +1,393 @@
+"""
+Tests for Admin Curriculum Reingest & Upload Endpoints.
+
+Validates:
+1. POST /api/admin/reingest-pdf non-blocking execution with BackgroundTasks
+   (returns 200 OK, success=True, execution_mode="background_tasks", schedules run_cloud_ingestion_and_upload).
+2. POST /api/admin/reingest-pdf with GitHub workflow dispatch
+   (mocking urllib.request.urlopen returning HTTP 204 when GITHUB_PAT or GITHUB_TOKEN is set;
+   returns execution_mode="github_actions").
+3. POST /api/admin/upload-pdf non-blocking behavior
+   (re-ingestion scheduled via BackgroundTasks rather than running synchronously).
+4. GET /api/admin/reingest-status returns current REINGESTION_STATUS.
+5. Role-based access control (RBAC) enforcement on all endpoints.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from fastapi import BackgroundTasks
+from fastapi.testclient import TestClient
+
+import main as main_module
+from main import app
+from routes.admin_routes import PDF_METADATA, REINGESTION_STATUS
+
+
+@pytest.fixture
+def admin_auth():
+    orig_ready = getattr(main_module, "_firebase_ready", False)
+    orig_init = getattr(main_module, "_init_firebase_admin", None)
+    orig_auth = main_module.firebase_auth
+    orig_verify = getattr(main_module.firebase_auth, "verify_id_token", None) if main_module.firebase_auth else None
+
+    main_module._firebase_ready = True
+    main_module._init_firebase_admin = lambda: None
+    if not main_module.firebase_auth:
+        main_module.firebase_auth = MagicMock()
+    main_module.firebase_auth.verify_id_token = MagicMock(return_value={
+        "uid": "admin-test-uid",
+        "email": "admin@test.mathpulse.ai",
+        "name": "Admin Tester",
+        "role": "admin",
+    })
+
+    yield
+
+    main_module._firebase_ready = orig_ready
+    main_module._init_firebase_admin = orig_init
+    main_module.firebase_auth = orig_auth
+    if orig_verify and main_module.firebase_auth:
+        main_module.firebase_auth.verify_id_token = orig_verify
+
+
+@pytest.fixture
+def client(admin_auth):
+    return TestClient(app, headers={"Authorization": "Bearer admin-token"})
+
+
+@pytest.fixture(autouse=True)
+def reset_reingestion_state():
+    REINGESTION_STATUS.clear()
+    REINGESTION_STATUS.update({
+        "status": "idle",
+        "last_run": None,
+        "message": None,
+        "mode": None,
+    })
+    yield
+    REINGESTION_STATUS.clear()
+    REINGESTION_STATUS.update({
+        "status": "idle",
+        "last_run": None,
+        "message": None,
+        "mode": None,
+    })
+
+
+@pytest.fixture(autouse=True)
+def mock_audit_logger():
+    mock_logger = AsyncMock()
+    with patch("routes.admin_routes._get_audit_logger", return_value=mock_logger):
+        yield mock_logger
+
+
+# ─────────────────────────────────────────────────────────────
+# 1. POST /api/admin/reingest-pdf tests
+# ─────────────────────────────────────────────────────────────
+
+
+class TestReingestPdf:
+    """Tests for POST /api/admin/reingest-pdf endpoint."""
+
+    def test_reingest_pdf_background_tasks_execution(self, client, monkeypatch):
+        """Verify non-blocking execution with BackgroundTasks when no GitHub token is present."""
+        monkeypatch.delenv("GITHUB_PAT", raising=False)
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+
+        with patch.object(BackgroundTasks, "add_task") as mock_add_task, \
+             patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            response = client.post("/api/admin/reingest-pdf", json={})
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["execution_mode"] == "background_tasks"
+            assert "Remote re-ingestion started in the cloud." in data["message"]
+
+            mock_add_task.assert_called_once()
+            assert mock_add_task.call_args[0][0] == mock_ingest
+
+            assert REINGESTION_STATUS["status"] == "running"
+            assert REINGESTION_STATUS["mode"] == "background_tasks"
+            assert REINGESTION_STATUS["last_run"] is not None
+
+    def test_reingest_pdf_background_tasks_invokes_task(self, client, monkeypatch):
+        """Verify that run_cloud_ingestion_and_upload is scheduled and invoked by BackgroundTasks."""
+        monkeypatch.delenv("GITHUB_PAT", raising=False)
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+
+        with patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            response = client.post("/api/admin/reingest-pdf", json={"subjectId": "general_mathematics"})
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["execution_mode"] == "background_tasks"
+            mock_ingest.assert_called_once()
+
+    def test_reingest_pdf_github_actions_dispatch_with_pat(self, client, monkeypatch):
+        """Verify GitHub Actions workflow dispatch when GITHUB_PAT is set and HTTP 204 returned."""
+        monkeypatch.setenv("GITHUB_PAT", "ghp_mock_pat_token_test_12345")
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+
+        mock_response = MagicMock()
+        mock_response.status = 204
+        mock_cm = MagicMock()
+        mock_cm.__enter__.return_value = mock_response
+        mock_cm.__exit__.return_value = None
+
+        with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen, \
+             patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            response = client.post("/api/admin/reingest-pdf", json={})
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["execution_mode"] == "github_actions"
+            assert "dispatched to GitHub Actions runner" in data["message"]
+
+            mock_ingest.assert_not_called()
+            mock_urlopen.assert_called_once()
+            req = mock_urlopen.call_args[0][0]
+            assert "workflows/ingest-curriculum.yml/dispatches" in req.full_url
+            assert req.headers.get("Authorization") == "Bearer ghp_mock_pat_token_test_12345"
+
+            assert REINGESTION_STATUS["status"] == "running"
+            assert REINGESTION_STATUS["mode"] == "github_actions"
+            assert REINGESTION_STATUS["last_run"] is not None
+
+    def test_reingest_pdf_github_actions_dispatch_with_token(self, client, monkeypatch):
+        """Verify GitHub Actions workflow dispatch when GITHUB_TOKEN is set."""
+        monkeypatch.delenv("GITHUB_PAT", raising=False)
+        monkeypatch.setenv("GITHUB_TOKEN", "ghs_mock_token_test_67890")
+
+        mock_response = MagicMock()
+        mock_response.status = 204
+        mock_cm = MagicMock()
+        mock_cm.__enter__.return_value = mock_response
+        mock_cm.__exit__.return_value = None
+
+        with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen, \
+             patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            response = client.post("/api/admin/reingest-pdf", json={})
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["execution_mode"] == "github_actions"
+            mock_ingest.assert_not_called()
+
+            req = mock_urlopen.call_args[0][0]
+            assert req.headers.get("Authorization") == "Bearer ghs_mock_token_test_67890"
+
+    def test_reingest_pdf_fallback_to_background_tasks_on_dispatch_failure(self, client, monkeypatch):
+        """Verify graceful fallback to background_tasks if GitHub Actions dispatch returns non-204."""
+        monkeypatch.setenv("GITHUB_PAT", "ghp_mock_pat_token_test_12345")
+
+        mock_response = MagicMock()
+        mock_response.status = 500
+        mock_cm = MagicMock()
+        mock_cm.__enter__.return_value = mock_response
+        mock_cm.__exit__.return_value = None
+
+        with patch("urllib.request.urlopen", return_value=mock_cm), \
+             patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            response = client.post("/api/admin/reingest-pdf", json={})
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["execution_mode"] == "background_tasks"
+            mock_ingest.assert_called_once()
+            assert REINGESTION_STATUS["mode"] == "background_tasks"
+
+
+# ─────────────────────────────────────────────────────────────
+# 2. POST /api/admin/upload-pdf tests
+# ─────────────────────────────────────────────────────────────
+
+
+class TestUploadPdf:
+    """Tests for POST /api/admin/upload-pdf endpoint non-blocking behavior."""
+
+    def test_upload_pdf_non_blocking_background_task(self, client):
+        """Verify PDF upload writes to Firebase Storage and schedules ingestion via BackgroundTasks."""
+        mock_blob = MagicMock()
+        mock_bucket = MagicMock()
+        mock_bucket.name = "mathpulse-ai-2026.firebasestorage.app"
+        mock_bucket.blob.return_value = mock_blob
+
+        with patch("routes.admin_routes._init_firebase_storage", return_value=(None, mock_bucket)), \
+             patch("routes.admin_routes.run_cloud_ingestion_and_upload") as mock_ingest:
+            form_data = {
+                "subjectId": "general_mathematics",
+                "subjectName": "General Mathematics",
+                "semester": 1,
+                "quarter": 1,
+            }
+            files = {
+                "file": ("SSHS_GM_Q1_Module1.pdf", b"%PDF-1.4 test module content binary stream", "application/pdf")
+            }
+
+            response = client.post("/api/admin/upload-pdf", data=form_data, files=files)
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["success"] is True
+            assert data["chunkCount"] == 0
+            assert data["subjectId"] == "general_mathematics"
+            expected_storage_path = "curriculum/general_mathematics/SSHS_GM_Q1_Module1.pdf"
+            assert data["storageUrl"] == f"gs://mathpulse-ai-2026.firebasestorage.app/{expected_storage_path}"
+
+            mock_bucket.blob.assert_called_once_with(expected_storage_path)
+            mock_blob.upload_from_string.assert_called_once_with(
+                b"%PDF-1.4 test module content binary stream",
+                content_type="application/pdf",
+            )
+            mock_ingest.assert_called_once()
+
+            assert expected_storage_path in PDF_METADATA
+            assert PDF_METADATA[expected_storage_path] == {
+                "subject": "General Mathematics",
+                "subjectId": "general_mathematics",
+                "type": "uploaded_module",
+                "semester": 1,
+                "quarter": 1,
+            }
+
+    def test_upload_pdf_rejects_non_pdf_file(self, client):
+        """Verify rejection of non-PDF files with 400 Bad Request."""
+        form_data = {
+            "subjectId": "general_mathematics",
+            "subjectName": "General Mathematics",
+            "semester": 1,
+            "quarter": 1,
+        }
+        files = {
+            "file": ("curriculum_guide.docx", b"PK\x03\x04 mock docx data", "application/vnd.openxmlformats")
+        }
+
+        response = client.post("/api/admin/upload-pdf", data=form_data, files=files)
+        assert response.status_code == 400
+        assert "Only PDF files are allowed." in response.json()["detail"]
+
+    def test_upload_pdf_storage_not_initialized(self, client):
+        """Verify 500 error when Firebase storage is unavailable."""
+        with patch("routes.admin_routes._init_firebase_storage", return_value=(None, None)):
+            form_data = {
+                "subjectId": "general_mathematics",
+                "subjectName": "General Mathematics",
+                "semester": 1,
+                "quarter": 1,
+            }
+            files = {
+                "file": ("test.pdf", b"%PDF-1.4 sample", "application/pdf")
+            }
+            response = client.post("/api/admin/upload-pdf", data=form_data, files=files)
+            assert response.status_code == 500
+            assert "Firebase storage is not initialized." in response.json()["detail"]
+
+
+# ─────────────────────────────────────────────────────────────
+# 3. GET /api/admin/reingest-status tests
+# ─────────────────────────────────────────────────────────────
+
+
+class TestReingestStatus:
+    """Tests for GET /api/admin/reingest-status endpoint."""
+
+    def test_get_reingest_status_idle(self, client):
+        """Verify initial/idle status is returned accurately."""
+        response = client.get("/api/admin/reingest-status")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "idle"
+        assert data["last_run"] is None
+        assert data["message"] is None
+        assert data["mode"] is None
+
+    def test_get_reingest_status_reflects_active_reingestion(self, client):
+        """Verify that updated reingestion status is reflected in GET response."""
+        REINGESTION_STATUS["status"] = "running"
+        REINGESTION_STATUS["mode"] = "github_actions"
+        REINGESTION_STATUS["message"] = "Dispatched workflow to GitHub Actions"
+        REINGESTION_STATUS["last_run"] = "2026-09-07T13:00:00+00:00"
+
+        response = client.get("/api/admin/reingest-status")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "running"
+        assert data["mode"] == "github_actions"
+        assert data["message"] == "Dispatched workflow to GitHub Actions"
+        assert data["last_run"] == "2026-09-07T13:00:00+00:00"
+
+
+# ─────────────────────────────────────────────────────────────
+# 4. Authentication & RBAC Enforcement tests
+# ─────────────────────────────────────────────────────────────
+
+
+class TestAdminReingestAuth:
+    """Tests role-based access control (RBAC) on admin reingestion endpoints."""
+
+    def test_reingest_pdf_unauthorized(self):
+        """Missing auth header returns 401."""
+        unauth_client = TestClient(app)
+        response = unauth_client.post("/api/admin/reingest-pdf", json={})
+        assert response.status_code in {401, 403}
+
+    def test_reingest_pdf_forbidden_for_student(self):
+        """Student role returns 403 Forbidden."""
+        main_module._firebase_ready = True
+        main_module._init_firebase_admin = lambda: None
+        if not main_module.firebase_auth:
+            main_module.firebase_auth = MagicMock()
+        mock_claims = {
+            "uid": "student-uid",
+            "email": "student@mathpulse.ai",
+            "role": "student",
+        }
+        with patch.object(main_module.firebase_auth, "verify_id_token", return_value=mock_claims):
+            student_client = TestClient(app, headers={"Authorization": "Bearer student-token"})
+            response = student_client.post("/api/admin/reingest-pdf", json={})
+            assert response.status_code == 403
+
+    def test_upload_pdf_unauthorized(self):
+        """Upload without token returns 401."""
+        unauth_client = TestClient(app)
+        response = unauth_client.post("/api/admin/upload-pdf")
+        assert response.status_code in {401, 403}
+
+    def test_upload_pdf_forbidden_for_student(self):
+        """Upload with student token returns 403."""
+        main_module._firebase_ready = True
+        main_module._init_firebase_admin = lambda: None
+        if not main_module.firebase_auth:
+            main_module.firebase_auth = MagicMock()
+        mock_claims = {
+            "uid": "student-uid",
+            "email": "student@mathpulse.ai",
+            "role": "student",
+        }
+        with patch.object(main_module.firebase_auth, "verify_id_token", return_value=mock_claims):
+            student_client = TestClient(app, headers={"Authorization": "Bearer student-token"})
+            response = student_client.post("/api/admin/upload-pdf")
+            assert response.status_code == 403
+
+    def test_get_reingest_status_forbidden_for_student(self):
+        """Status endpoint with student token returns 403."""
+        main_module._firebase_ready = True
+        main_module._init_firebase_admin = lambda: None
+        if not main_module.firebase_auth:
+            main_module.firebase_auth = MagicMock()
+        mock_claims = {
+            "uid": "student-uid",
+            "email": "student@mathpulse.ai",
+            "role": "student",
+        }
+        with patch.object(main_module.firebase_auth, "verify_id_token", return_value=mock_claims):
+            student_client = TestClient(app, headers={"Authorization": "Bearer student-token"})
+            response = student_client.get("/api/admin/reingest-status")
+            assert response.status_code == 403
+

tests/test_quiz_battle.py

@@ -125,16 +125,24 @@ class TestQuestionBankService:
                 "difficulty": "easy",
                 "random_seed": 0.5,
             }
-            mock_collection = MagicMock()
-            mock_collection.where.return_value.order_by.return_value.limit.return_value.stream.return_value = [mock_doc]
-            mock_collection.where.return_value.order_by.return_value.limit.return_value.stream.return_value = [mock_doc]
-            mock_db.return_value.collection.return_value = mock_collection
+            # New path: db.collection("question_bank").document(grade).collection("topics").document(topic).collection("questions")
+            # MagicMock auto-chains, so navigate to the questions collection via the chained call
+            mock_questions_col = (
+                mock_db.return_value
+                .collection.return_value   # "question_bank"
+                .document.return_value     # grade_level doc
+                .collection.return_value   # "topics"
+                .document.return_value     # topic doc
+                .collection.return_value   # "questions"
+            )
+            mock_questions_col.where.return_value.order_by.return_value.limit.return_value.stream.return_value = [mock_doc]
 
             from services.question_bank_service import get_questions_for_battle
             questions = await get_questions_for_battle(8, "linear_equations", 1)
             assert len(questions) == 1
             assert questions[0]["question"] == "What is 2+2?"
 
+
     @pytest.mark.asyncio
     async def test_cache_session_questions(self):
         """Cache questions for 24 hours."""