v1.2

.gitignore

@@ -8,3 +8,7 @@ uploads/
 .next
 dist
 .vercel
+inter
+Resume.pdf
+LANGGRAPH_AND_TOOLS.md
+WORKFLOW.md

backend/__init__.py

@@ -0,0 +1 @@
+# This file makes the backend directory a Python package.

backend/config.py

@@ -1,4 +1,5 @@
 from pydantic_settings import BaseSettings
+from pydantic import field_validator
 from functools import lru_cache
 import os
 from dotenv import load_dotenv
@@ -38,6 +39,26 @@ class Settings(BaseSettings):
         env_file = ".env"
         extra = "ignore"
 
+    @field_validator("MONGO_URI")
+    @classmethod
+    def validate_mongo_uri(cls, value: str) -> str:
+        v = (value or "").strip().lower()
+        if "localhost" in v or "127.0.0.1" in v:
+            raise ValueError("MONGO_URI must point to MongoDB Atlas, not localhost")
+        if not v.startswith("mongodb+srv://"):
+            raise ValueError("MONGO_URI must use mongodb+srv:// for cloud deployment")
+        return value
+
+    @field_validator("REDIS_URL")
+    @classmethod
+    def validate_redis_url(cls, value: str) -> str:
+        v = (value or "").strip().lower()
+        if "localhost" in v or "127.0.0.1" in v:
+            raise ValueError("REDIS_URL must point to a cloud Redis instance, not localhost")
+        if not (v.startswith("redis://") or v.startswith("rediss://")):
+            raise ValueError("REDIS_URL must start with redis:// or rediss://")
+        return value
+
 
 @lru_cache()
 def get_settings() -> Settings:

backend/database.py

@@ -36,9 +36,17 @@ async def connect_db():
     )
 
     # Test connections
-    await mongo_client.admin.command("ping")
-    await redis_client.ping()
-    print("✅ Connected to MongoDB Atlas and Redis")
+    try:
+        await mongo_client.admin.command("ping")
+        print("✅ Connected to MongoDB Atlas")
+    except Exception as e:
+        print(f"❌ Failed to connect to MongoDB: {e}")
+
+    try:
+        await redis_client.ping()
+        print("✅ Connected to Redis")
+    except Exception as e:
+        print(f"⚠️ Failed to connect to Redis (URL might be invalid or unreachable): {e}")
 
 
 async def close_db():

backend/main.py

@@ -4,6 +4,8 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
 import os
 
+
+
 from config import get_settings
 from database import connect_db, close_db
 

backend/models/collections.py

@@ -6,6 +6,8 @@ SKILLS = "skills"
 JOB_ROLES = "job_roles"
 ROLE_REQUIREMENTS = "role_requirements"
 QUESTIONS = "questions"
+TOPICS = "topics"
+TOPIC_QUESTIONS = "topic_questions"
 SESSIONS = "sessions"
 ANSWERS = "answers"
 RESULTS = "results"

backend/requirements.txt

@@ -1,6 +1,7 @@
 fastapi==0.115.0
 uvicorn[standard]==0.30.0
 motor==3.5.0
+pymongo<4.9
 redis[hiredis]==5.0.0
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
@@ -8,6 +9,10 @@ bcrypt==4.0.1
 python-multipart==0.0.9
 google-genai==1.5.0
 langgraph==0.2.0
+langchain-core==0.2.43
 pydantic-settings==2.5.0
+email-validator==2.2.0
 python-dotenv==1.0.1
 aiofiles==24.1.0
+pypdf==5.4.0
+python-docx==1.1.2

backend/routers/admin.py

@@ -1,14 +1,20 @@
-from fastapi import APIRouter, Depends, HTTPException, Query
+import json
+from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form
 from auth.jwt import require_role, get_current_user
 from schemas.admin import (
     JobRoleCreate, JobRoleUpdate,
     QuestionCreate, QuestionUpdate,
     RoleRequirementCreate,
+    TopicCreate, TopicUpdate, TopicPublishUpdate,
 )
 from services.admin_service import (
     create_role, update_role, delete_role, list_roles,
-    create_question, update_question, delete_question, list_questions,
+    create_question, update_question, delete_question, list_questions, get_question_by_id,
+    create_topic, list_topics, update_topic, delete_topic, set_topic_publish_status,
+    import_questions_from_pdf,
     create_requirement, list_requirements, delete_requirement,
+    list_quit_interviews, list_admin_reports, get_admin_report_detail,
+    list_admin_users, delete_admin_user,
 )
 from services.analytics_service import get_admin_analytics
 
@@ -69,10 +75,18 @@ async def delete_role_endpoint(
 @router.get("/questions")
 async def get_questions(
     role_id: str = Query(None),
+    topic_id: str = Query(None),
+    interview_type: str = Query(None),
+    difficulty: str = Query(None),
     current_user: dict = Depends(require_role("admin")),
 ):
     """List questions, optionally filtered by role."""
-    questions = await list_questions(role_id)
+    questions = await list_questions(
+        role_id=role_id,
+        topic_id=topic_id,
+        interview_type=interview_type,
+        difficulty=difficulty,
+    )
     return {"questions": questions}
 
 
@@ -84,6 +98,8 @@ async def create_question_endpoint(
     """Create a new question (admin only)."""
     result = await create_question(
         role_id=request.role_id,
+        topic_id=request.topic_id,
+        interview_type=request.interview_type,
         question=request.question,
         difficulty=request.difficulty,
         category=request.category,
@@ -92,6 +108,64 @@ async def create_question_endpoint(
     return result
 
 
+@router.get("/questions/{question_id}")
+async def get_question_by_id_endpoint(
+    question_id: str,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Get one question by id (admin only)."""
+    try:
+        question = await get_question_by_id(question_id)
+        return question
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
+@router.post("/questions/upload")
+async def upload_questions_pdf_endpoint(
+    interview_type: str = Form("resume"),
+    role_id: str | None = Form(None),
+    topic_id: str | None = Form(None),
+    subjects: str | None = Form(None),
+    file: UploadFile = File(...),
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Upload a PDF and extract interview questions (admin only)."""
+    if not file.filename:
+        raise HTTPException(status_code=400, detail="No file provided")
+
+    if not file.filename.lower().endswith(".pdf"):
+        raise HTTPException(status_code=400, detail="Only PDF files are supported for question import")
+
+    content = await file.read()
+    if len(content) > 10 * 1024 * 1024:
+        raise HTTPException(status_code=400, detail="File too large. Maximum 10MB")
+
+    parsed_subjects = []
+    if subjects:
+        try:
+            parsed_subjects = json.loads(subjects)
+            if not isinstance(parsed_subjects, list):
+                raise ValueError
+        except Exception:
+            parsed_subjects = [s.strip() for s in subjects.split(",") if s.strip()]
+
+    try:
+        result = await import_questions_from_pdf(
+            role_id=role_id,
+            topic_id=topic_id,
+            interview_type=interview_type,
+            subjects=parsed_subjects,
+            filename=file.filename,
+            file_content=content,
+        )
+        return result
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to import questions from PDF: {str(e)}")
+
+
 @router.put("/questions/{question_id}")
 async def update_question_endpoint(
     question_id: str,
@@ -130,6 +204,76 @@ async def get_requirements(
     return {"requirements": requirements}
 
 
+# ─── Topics ───
+
+@router.get("/topics")
+async def get_topics(current_user: dict = Depends(get_current_user)):
+    """List all topic categories (accessible by all authenticated users)."""
+    only_published = current_user.get("role") != "admin"
+    topics = await list_topics(only_published=only_published)
+    return {"topics": topics}
+
+
+@router.post("/topics")
+async def create_topic_endpoint(
+    request: TopicCreate,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Create a topic category (admin only)."""
+    try:
+        result = await create_topic(name=request.name, description=request.description)
+        return result
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.put("/topics/{topic_id}")
+async def update_topic_endpoint(
+    topic_id: str,
+    request: TopicUpdate,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Update a topic category (admin only)."""
+    try:
+        result = await update_topic(topic_id, request.model_dump())
+        return result
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.delete("/topics/{topic_id}")
+async def delete_topic_endpoint(
+    topic_id: str,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Delete a topic category and its topic questions (admin only)."""
+    success = await delete_topic(topic_id)
+    if not success:
+        raise HTTPException(status_code=404, detail="Topic not found")
+    return {"message": "Topic deleted"}
+
+
+@router.put("/topics/{topic_id}/publish")
+async def publish_topic_endpoint(
+    topic_id: str,
+    request: TopicPublishUpdate,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Publish/unpublish a topic for student interview selection (admin only)."""
+    try:
+        result = await set_topic_publish_status(
+            topic_id,
+            request.is_published,
+            timer_enabled=request.timer_enabled,
+            timer_seconds=request.timer_seconds,
+        )
+        return result
+    except ValueError as e:
+        detail = str(e)
+        status_code = 404 if "not found" in detail.lower() else 400
+        raise HTTPException(status_code=status_code, detail=detail)
+
+
 @router.post("/requirements")
 async def create_requirement_endpoint(
     request: RoleRequirementCreate,
@@ -165,3 +309,63 @@ async def get_analytics(
     """Get admin analytics dashboard data."""
     analytics = await get_admin_analytics()
     return analytics
+
+
+@router.get("/quit-interviews")
+async def get_quit_interviews(
+    limit: int = Query(100, ge=1, le=500),
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Get full details about interviews quit by users."""
+    items = await list_quit_interviews(limit=limit)
+    return {"items": items}
+
+
+@router.get("/reports")
+async def get_admin_reports(
+    limit: int = Query(100, ge=1, le=500),
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Get all interview report summaries for admin."""
+    items = await list_admin_reports(limit=limit)
+    return {"items": items}
+
+
+@router.get("/reports/{session_id}")
+async def get_admin_report_by_session(
+    session_id: str,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Get full report details for a specific interview session (admin only)."""
+    try:
+        item = await get_admin_report_detail(session_id=session_id)
+        return item
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+
+@router.get("/users")
+async def get_admin_users(
+    limit: int = Query(500, ge=1, le=1000),
+    current_user: dict = Depends(require_role("admin")),
+):
+    """List users for admin management."""
+    items = await list_admin_users(limit=limit)
+    return {"items": items}
+
+
+@router.delete("/users/{user_id}")
+async def delete_admin_user_endpoint(
+    user_id: str,
+    current_user: dict = Depends(require_role("admin")),
+):
+    """Delete a student user and related records (admin only)."""
+    try:
+        success = await delete_admin_user(user_id, current_user["user_id"])
+        if not success:
+            raise HTTPException(status_code=404, detail="User not found")
+        return {"message": "User deleted"}
+    except ValueError as e:
+        detail = str(e)
+        status_code = 404 if "not found" in detail.lower() else 400
+        raise HTTPException(status_code=status_code, detail=detail)

backend/routers/interview.py

@@ -3,10 +3,11 @@ from auth.jwt import get_current_user
 from schemas.interview import (
     StartInterviewRequest,
     SubmitAnswerRequest,
+    QuitInterviewRequest,
     InterviewStartResponse,
     AnswerResponse,
 )
-from services.interview_service import start_interview, submit_answer
+from services.interview_service import start_interview, submit_answer, quit_interview
 from services.evaluation_service import generate_report
 
 router = APIRouter()
@@ -22,6 +23,9 @@ async def start_interview_endpoint(
         result = await start_interview(
             user_id=current_user["user_id"],
             role_id=request.role_id,
+            custom_role=request.custom_role,
+            interview_type=request.interview_type,
+            topic_id=request.topic_id,
         )
         return result
     except Exception as e:
@@ -47,6 +51,37 @@ async def submit_answer_endpoint(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+@router.post("/quit")
+async def quit_interview_endpoint(
+    request: QuitInterviewRequest,
+    current_user: dict = Depends(get_current_user),
+):
+    """Quit an in-progress interview and generate a partial report if answers exist."""
+    try:
+        quit_result = await quit_interview(
+            session_id=request.session_id,
+            user_id=current_user["user_id"],
+        )
+
+        report = None
+        if quit_result.get("report_generated"):
+            report = await generate_report(
+                session_id=request.session_id,
+                user_id=current_user["user_id"],
+            )
+
+        return {
+            "session_id": request.session_id,
+            "report_generated": bool(report),
+            "report": report,
+            "message": quit_result.get("message", "Interview quit"),
+        }
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
 @router.get("/report")
 async def get_interview_report(
     session_id: str,

backend/routers/profile.py

@@ -1,8 +1,9 @@
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from auth.jwt import get_current_user
 from database import get_db
 from models.collections import USERS, RESUMES, SKILLS
 from utils.helpers import str_objectid
+from utils.skills import normalize_skill_list, cluster_skills
 from bson import ObjectId
 
 router = APIRouter()
@@ -32,6 +33,7 @@ async def get_profile(current_user: dict = Depends(get_current_user)):
             "filename": resume.get("original_filename", ""),
             "uploaded_at": resume.get("uploaded_at", ""),
             "parsed_text": resume.get("parsed_text", ""),
+            "parsed_data": resume.get("parsed_data", {}),
         }
     else:
         profile["resume"] = None
@@ -39,6 +41,7 @@ async def get_profile(current_user: dict = Depends(get_current_user)):
     # Get skills
     skills_doc = await db[SKILLS].find_one({"user_id": current_user["user_id"]})
     profile["skills"] = skills_doc.get("skills", []) if skills_doc else []
+    profile["clustered_skills"] = cluster_skills(profile["skills"])
 
     return profile
 
@@ -49,7 +52,7 @@ async def update_user_skills(
 ):
     """Update the current user's extracted skills."""
     db = get_db()
-    skills = request_data.get("skills", [])
+    skills = normalize_skill_list(request_data.get("skills", []))
     
     # Upsert the skills document for this user
     await db[SKILLS].update_one(
@@ -59,3 +62,24 @@ async def update_user_skills(
     )
     
     return {"message": "Skills updated successfully", "skills": skills}
+
+
+@router.put("/resume-data")
+async def update_resume_data(
+    request_data: dict, 
+    current_user: dict = Depends(get_current_user)
+):
+    """Update the detailed parsed data of the user's resume."""
+    db = get_db()
+    parsed_data = request_data.get("parsed_data", {})
+    
+    # Update only the parsed_data property inside the RESUMES collection
+    result = await db[RESUMES].update_one(
+        {"user_id": current_user["user_id"]},
+        {"$set": {"parsed_data": parsed_data}}
+    )
+
+    if result.matched_count == 0:
+        raise HTTPException(status_code=404, detail="Resume not found. Upload a resume first.")
+
+    return {"message": "Resume details updated successfully", "parsed_data": parsed_data}

backend/schemas/admin.py

@@ -23,7 +23,9 @@ class JobRoleResponse(BaseModel):
 
 
 class QuestionCreate(BaseModel):
-    role_id: str
+    role_id: Optional[str] = None
+    topic_id: Optional[str] = None
+    interview_type: str = "resume"
     question: str
     difficulty: str = "medium"
     category: Optional[str] = None
@@ -39,13 +41,38 @@ class QuestionUpdate(BaseModel):
 
 class QuestionResponse(BaseModel):
     id: str
-    role_id: str
+    role_id: Optional[str] = None
+    topic_id: Optional[str] = None
+    interview_type: str = "resume"
     question: str
     difficulty: str
     category: Optional[str] = None
     created_at: str
 
 
+class TopicCreate(BaseModel):
+    name: str
+    description: Optional[str] = None
+
+
+class TopicUpdate(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+
+
+class TopicPublishUpdate(BaseModel):
+    is_published: bool
+    timer_enabled: Optional[bool] = None
+    timer_seconds: Optional[int] = None
+
+
+class TopicResponse(BaseModel):
+    id: str
+    name: str
+    description: Optional[str] = None
+    created_at: str
+
+
 class RoleRequirementCreate(BaseModel):
     role_id: str
     skill: str

backend/schemas/interview.py

@@ -4,6 +4,9 @@ from typing import Optional, List, Dict
 
 class StartInterviewRequest(BaseModel):
     role_id: Optional[str] = None
+    custom_role: Optional[str] = None
+    interview_type: Optional[str] = "resume"
+    topic_id: Optional[str] = None
 
 
 class SubmitAnswerRequest(BaseModel):
@@ -12,6 +15,10 @@ class SubmitAnswerRequest(BaseModel):
     answer: str
 
 
+class QuitInterviewRequest(BaseModel):
+    session_id: str
+
+
 class InterviewQuestion(BaseModel):
     question_id: str
     question: str
@@ -33,6 +40,12 @@ class AnswerResponse(BaseModel):
     message: str = ""
 
 
+class QuitInterviewResponse(BaseModel):
+    session_id: str
+    report_generated: bool = False
+    message: str = ""
+
+
 class QuestionScore(BaseModel):
     question: str
     answer: str

backend/schemas/resume.py

@@ -12,3 +12,17 @@ class ResumeResponse(BaseModel):
 
 class UpdateSkillsRequest(BaseModel):
     skills: List[str]
+
+class ParsedDataPayload(BaseModel):
+    name: Optional[str] = None
+    email: Optional[str] = None
+    phone: Optional[str] = None
+    location: Optional[str] = None
+    recommended_roles: Optional[List[str]] = []
+    experience_summary: Optional[str] = None
+    experience: Optional[List[dict]] = []
+    education: Optional[List[dict]] = []
+    projects: Optional[List[dict]] = []
+
+class UpdateResumeDataRequest(BaseModel):
+    parsed_data: ParsedDataPayload

backend/services/admin_service.py

@@ -1,7 +1,12 @@
 from bson import ObjectId
+import json
+import re
+from datetime import datetime
 from database import get_db
-from models.collections import JOB_ROLES, ROLE_REQUIREMENTS, QUESTIONS
+from models.collections import JOB_ROLES, ROLE_REQUIREMENTS, QUESTIONS, TOPICS, TOPIC_QUESTIONS, SESSIONS, USERS, RESULTS, RESUMES, SKILLS, ANSWERS
 from utils.helpers import utc_now, str_objectid, str_objectids
+from utils.gemini import call_gemini
+from utils.resume_text import extract_resume_text
 
 
 # ─── Job Roles ───
@@ -47,18 +52,37 @@ async def list_roles() -> list:
 
 # ─── Questions ───
 
-async def create_question(role_id: str, question: str, difficulty: str = "medium",
-                          category: str = None, expected_answer: str = None) -> dict:
+async def create_question(
+    role_id: str = None,
+    topic_id: str = None,
+    interview_type: str = "resume",
+    question: str = "",
+    difficulty: str = "medium",
+    category: str = None,
+    expected_answer: str = None,
+) -> dict:
     db = get_db()
+    interview_type = (interview_type or "resume").strip().lower()
+    if interview_type not in {"resume", "topic"}:
+        raise ValueError("interview_type must be either 'resume' or 'topic'")
+
+    if interview_type == "resume" and not role_id:
+        raise ValueError("role_id is required for resume interview questions")
+    if interview_type == "topic" and not topic_id:
+        raise ValueError("topic_id is required for topic interview questions")
+
+    collection = QUESTIONS if interview_type == "resume" else TOPIC_QUESTIONS
     doc = {
         "role_id": role_id,
+        "topic_id": topic_id,
+        "interview_type": interview_type,
         "question": question,
         "difficulty": difficulty,
         "category": category,
         "expected_answer": expected_answer,
         "created_at": utc_now(),
     }
-    result = await db[QUESTIONS].insert_one(doc)
+    result = await db[collection].insert_one(doc)
     doc["_id"] = result.inserted_id
     return str_objectid(doc)
 
@@ -69,8 +93,14 @@ async def update_question(question_id: str, data: dict) -> dict:
     if not update_data:
         raise ValueError("No fields to update")
     update_data["updated_at"] = utc_now()
-    await db[QUESTIONS].update_one({"_id": ObjectId(question_id)}, {"$set": update_data})
+    # Try resume question collection first, then topic question collection.
+    result = await db[QUESTIONS].update_one({"_id": ObjectId(question_id)}, {"$set": update_data})
+    if result.matched_count == 0:
+        await db[TOPIC_QUESTIONS].update_one({"_id": ObjectId(question_id)}, {"$set": update_data})
+
     doc = await db[QUESTIONS].find_one({"_id": ObjectId(question_id)})
+    if not doc:
+        doc = await db[TOPIC_QUESTIONS].find_one({"_id": ObjectId(question_id)})
     if not doc:
         raise ValueError("Question not found")
     return str_objectid(doc)
@@ -79,17 +109,326 @@ async def update_question(question_id: str, data: dict) -> dict:
 async def delete_question(question_id: str) -> bool:
     db = get_db()
     result = await db[QUESTIONS].delete_one({"_id": ObjectId(question_id)})
+    if result.deleted_count > 0:
+        return True
+    result = await db[TOPIC_QUESTIONS].delete_one({"_id": ObjectId(question_id)})
     return result.deleted_count > 0
 
 
-async def list_questions(role_id: str = None) -> list:
+async def list_questions(
+    role_id: str = None,
+    topic_id: str = None,
+    interview_type: str = None,
+    difficulty: str = None,
+) -> list:
     db = get_db()
-    query = {"role_id": role_id} if role_id else {}
-    cursor = db[QUESTIONS].find(query).sort("created_at", -1)
+    interview_type = (interview_type or "").strip().lower()
+    difficulty = (difficulty or "").strip().lower()
+
+    docs = []
+    if interview_type in {"", "resume"}:
+        query = {"role_id": role_id} if role_id else {}
+        if difficulty:
+            query["difficulty"] = difficulty
+        cursor = db[QUESTIONS].find(query).sort("created_at", -1)
+        resume_docs = await cursor.to_list(length=200)
+        docs.extend(resume_docs)
+
+    if interview_type in {"", "topic"}:
+        query = {"topic_id": topic_id} if topic_id else {}
+        if difficulty:
+            query["difficulty"] = difficulty
+        cursor = db[TOPIC_QUESTIONS].find(query).sort("created_at", -1)
+        topic_docs = await cursor.to_list(length=200)
+        docs.extend(topic_docs)
+
+    docs.sort(key=lambda d: d.get("created_at", ""), reverse=True)
+    return str_objectids(docs)
+
+
+async def get_question_by_id(question_id: str) -> dict:
+    db = get_db()
+    doc = await db[QUESTIONS].find_one({"_id": ObjectId(question_id)})
+    if not doc:
+        doc = await db[TOPIC_QUESTIONS].find_one({"_id": ObjectId(question_id)})
+    if not doc:
+        raise ValueError("Question not found")
+    return str_objectid(doc)
+
+
+# ─── Topics ───
+
+async def create_topic(name: str, description: str = None) -> dict:
+    db = get_db()
+    existing = await db[TOPICS].find_one({"name": {"$regex": f"^{re.escape(name)}$", "$options": "i"}})
+    if existing:
+        raise ValueError("Topic already exists")
+
+    doc = {
+        "name": name,
+        "description": description,
+        "is_published": False,
+        "timer_enabled": False,
+        "timer_seconds": None,
+        "created_at": utc_now(),
+    }
+    result = await db[TOPICS].insert_one(doc)
+    doc["_id"] = result.inserted_id
+    return str_objectid(doc)
+
+
+async def list_topics(only_published: bool = False) -> list:
+    db = get_db()
+    query = {"is_published": True} if only_published else {}
+    cursor = db[TOPICS].find(query).sort("created_at", -1)
     docs = await cursor.to_list(length=200)
     return str_objectids(docs)
 
 
+async def update_topic(topic_id: str, data: dict) -> dict:
+    db = get_db()
+    update_data = {k: v for k, v in data.items() if v is not None}
+    if not update_data:
+        raise ValueError("No fields to update")
+    update_data["updated_at"] = utc_now()
+    await db[TOPICS].update_one({"_id": ObjectId(topic_id)}, {"$set": update_data})
+    doc = await db[TOPICS].find_one({"_id": ObjectId(topic_id)})
+    if not doc:
+        raise ValueError("Topic not found")
+    return str_objectid(doc)
+
+
+async def delete_topic(topic_id: str) -> bool:
+    db = get_db()
+    await db[TOPIC_QUESTIONS].delete_many({"topic_id": topic_id})
+    result = await db[TOPICS].delete_one({"_id": ObjectId(topic_id)})
+    return result.deleted_count > 0
+
+
+async def set_topic_publish_status(
+    topic_id: str,
+    is_published: bool,
+    timer_enabled: bool | None = None,
+    timer_seconds: int | None = None,
+) -> dict:
+    db = get_db()
+
+    update_data = {
+        "is_published": is_published,
+        "updated_at": utc_now(),
+    }
+
+    if timer_enabled is not None:
+        update_data["timer_enabled"] = bool(timer_enabled)
+
+    if timer_seconds is not None:
+        if timer_seconds <= 0:
+            raise ValueError("timer_seconds must be greater than 0")
+        update_data["timer_seconds"] = int(timer_seconds)
+
+    if timer_enabled is False:
+        update_data["timer_seconds"] = None
+
+    await db[TOPICS].update_one(
+        {"_id": ObjectId(topic_id)},
+        {"$set": update_data},
+    )
+    doc = await db[TOPICS].find_one({"_id": ObjectId(topic_id)})
+    if not doc:
+        raise ValueError("Topic not found")
+    return str_objectid(doc)
+
+
+def _extract_json_object(text: str) -> str:
+    value = (text or "").strip()
+    if value.startswith("```"):
+        value = value.split("\n", 1)[1]
+    if value.endswith("```"):
+        value = value.rsplit("```", 1)[0]
+    value = value.strip()
+
+    if value.startswith("{") and value.endswith("}"):
+        return value
+
+    start = value.find("{")
+    end = value.rfind("}")
+    if start != -1 and end != -1 and end > start:
+        return value[start:end + 1]
+
+    return value
+
+
+def _normalize_subject(subject: str, allowed_subjects: list[str]) -> str:
+    raw = (subject or "").strip().lower()
+    if not raw:
+        return ""
+
+    for allowed in allowed_subjects:
+        if raw == allowed.lower():
+            return allowed
+
+    for allowed in allowed_subjects:
+        a = allowed.lower()
+        if raw in a or a in raw:
+            return allowed
+
+    return ""
+
+
+async def import_questions_from_pdf(
+    role_id: str | None,
+    topic_id: str | None,
+    interview_type: str,
+    subjects: list[str] | None,
+    filename: str,
+    file_content: bytes,
+) -> dict:
+    db = get_db()
+
+    interview_type = (interview_type or "resume").strip().lower()
+    if interview_type not in {"resume", "topic"}:
+        raise ValueError("interview_type must be either 'resume' or 'topic'")
+
+    clean_subjects = []
+    for item in (subjects or []):
+        value = (item or "").strip()
+        if value and value.lower() not in [s.lower() for s in clean_subjects]:
+            clean_subjects.append(value)
+
+    if interview_type == "resume" and not role_id:
+        raise ValueError("role_id is required for resume question import")
+
+    if interview_type == "topic" and not topic_id:
+        raise ValueError("topic_id is required for topic question import")
+
+    if interview_type == "resume" and not clean_subjects:
+        raise ValueError("At least one subject is required")
+
+    text = extract_resume_text(filename, file_content)
+    if not text or len(text) < 20:
+        raise ValueError("Could not extract readable text from PDF")
+
+    topic_name = ""
+    if interview_type == "topic" and topic_id:
+        topic_doc = await db[TOPICS].find_one({"_id": ObjectId(topic_id)})
+        if not topic_doc:
+            raise ValueError("Topic not found")
+        topic_name = (topic_doc.get("name") or "").strip()
+
+    if interview_type == "topic":
+        prompt = f"""You are extracting topic-specific interview questions from a document.
+
+Target topic: {topic_name or "General"}
+
+Rules:
+1. Extract only actual interview questions relevant to the target topic.
+2. Ignore headings, instructions, answers, explanations, and duplicates.
+3. Keep each question concise and interview-ready.
+4. Assign a difficulty: easy, medium, or hard.
+
+Return ONLY valid JSON in this format:
+{{
+  "questions": [
+    {{"question": "...", "difficulty": "medium"}}
+  ]
+}}
+
+Document text:
+---
+{text}
+---"""
+    else:
+        prompt = f"""You are extracting interview questions from a document.
+
+Allowed subjects (must choose one of these for each question): {', '.join(clean_subjects)}
+
+Rules:
+1. Extract only actual interview questions from the document.
+2. Ignore headings, instructions, answers, explanations, and duplicates.
+3. Assign each extracted question to ONE allowed subject from the list above.
+4. Assign a difficulty: easy, medium, or hard.
+5. Keep question text clean and concise.
+
+Return ONLY valid JSON in this format:
+{{
+  "questions": [
+    {{"question": "...", "subject": "...", "difficulty": "medium"}}
+  ]
+}}
+
+Document text:
+---
+{text}
+---"""
+
+    raw = await call_gemini(prompt)
+    parsed_text = _extract_json_object(raw)
+    try:
+        parsed = json.loads(parsed_text)
+    except json.JSONDecodeError as exc:
+        raise ValueError("Failed to parse extracted questions from AI response") from exc
+
+    items = parsed.get("questions", []) if isinstance(parsed, dict) else []
+    if not isinstance(items, list) or not items:
+        raise ValueError("No questions were extracted from this PDF")
+
+    allowed_difficulties = {"easy", "medium", "hard"}
+    docs = []
+    seen = set()
+
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+
+        q_text = (item.get("question") or "").strip()
+        if len(q_text) < 8:
+            continue
+        q_text = re.sub(r"\s+", " ", q_text)
+
+        if interview_type == "topic":
+            subject = topic_name or "Topic"
+        else:
+            subject = _normalize_subject(item.get("subject", ""), clean_subjects)
+            if not subject:
+                continue
+
+        difficulty = (item.get("difficulty") or "medium").strip().lower()
+        if difficulty not in allowed_difficulties:
+            difficulty = "medium"
+
+        key = q_text.lower()
+        if key in seen:
+            continue
+        seen.add(key)
+
+        docs.append(
+            {
+                "role_id": role_id,
+                "topic_id": topic_id,
+                "interview_type": interview_type,
+                "question": q_text,
+                "difficulty": difficulty,
+                "category": subject,
+                "source": "pdf_upload",
+                "created_at": utc_now(),
+            }
+        )
+
+    if not docs:
+        if interview_type == "topic":
+            raise ValueError("No valid topic questions found in this PDF")
+        raise ValueError("No valid questions found after subject filtering")
+
+    collection = QUESTIONS if interview_type == "resume" else TOPIC_QUESTIONS
+    result = await db[collection].insert_many(docs)
+    return {
+        "inserted_count": len(result.inserted_ids),
+        "subjects": clean_subjects,
+        "interview_type": interview_type,
+        "topic_id": topic_id,
+    }
+
+
 # ─── Role Requirements ───
 
 async def create_requirement(role_id: str, skill: str, level: str = "intermediate") -> dict:
@@ -116,3 +455,174 @@ async def delete_requirement(req_id: str) -> bool:
     db = get_db()
     result = await db[ROLE_REQUIREMENTS].delete_one({"_id": ObjectId(req_id)})
     return result.deleted_count > 0
+
+
+async def list_quit_interviews(limit: int = 100) -> list:
+    """List interviews quit by users with full admin-facing details."""
+    db = get_db()
+
+    cursor = db[SESSIONS].find(
+        {"status": {"$in": ["quit", "quit_with_report"]}}
+    ).sort("quit_at", -1).limit(limit)
+    sessions = await cursor.to_list(length=limit)
+
+    output = []
+    for session in sessions:
+        user_id = session.get("user_id")
+        user_doc = None
+        if user_id:
+            try:
+                user_doc = await db[USERS].find_one({"_id": ObjectId(user_id)})
+            except Exception:
+                user_doc = await db[USERS].find_one({"id": user_id})
+
+        result_doc = await db[RESULTS].find_one({"session_id": session.get("session_id")})
+
+        quit_at = session.get("quit_at")
+        quit_dt = None
+        if isinstance(quit_at, str):
+            try:
+                quit_dt = datetime.fromisoformat(quit_at.replace("Z", "+00:00"))
+            except Exception:
+                quit_dt = None
+
+        output.append(
+            {
+                "session_id": session.get("session_id"),
+                "user_id": user_id,
+                "user_name": (user_doc or {}).get("name", "Unknown"),
+                "user_email": (user_doc or {}).get("email", "Unknown"),
+                "role_title": session.get("role_title", "Unknown"),
+                "status": session.get("status"),
+                "quit_reason": session.get("quit_reason", "user_requested"),
+                "answered_count": session.get("answered_count", 0),
+                "max_questions": session.get("max_questions", 0),
+                "quit_at": quit_at,
+                "quit_day": quit_dt.strftime("%A") if quit_dt else None,
+                "quit_date": quit_dt.strftime("%Y-%m-%d") if quit_dt else None,
+                "quit_time": quit_dt.strftime("%H:%M:%S %Z") if quit_dt else None,
+                "report_generated": bool(result_doc),
+                "overall_score": (result_doc or {}).get("overall_score"),
+                "total_questions_evaluated": (result_doc or {}).get("total_questions", 0),
+                "strengths": (result_doc or {}).get("strengths", []),
+                "weaknesses": (result_doc or {}).get("weaknesses", []),
+                "recommendations": (result_doc or {}).get("recommendations", []),
+            }
+        )
+
+    return output
+
+
+async def list_admin_reports(limit: int = 100) -> list:
+    """List all interview results for admin overview."""
+    db = get_db()
+    cursor = db[RESULTS].find().sort("completed_at", -1).limit(limit)
+    reports = await cursor.to_list(length=limit)
+
+    output = []
+    for report in reports:
+        user_id = report.get("user_id")
+        user_doc = None
+        if user_id:
+            try:
+                user_doc = await db[USERS].find_one({"_id": ObjectId(user_id)})
+            except Exception:
+                user_doc = await db[USERS].find_one({"id": user_id})
+
+        output.append(
+            {
+                "session_id": report.get("session_id"),
+                "user_id": user_id,
+                "user_name": (user_doc or {}).get("name", "Unknown"),
+                "user_email": (user_doc or {}).get("email", "Unknown"),
+                "role_title": report.get("role_title", "Unknown"),
+                "overall_score": report.get("overall_score", 0),
+                "total_questions": report.get("total_questions", 0),
+                "completed_at": report.get("completed_at", ""),
+                "session_status": report.get("session_status", "completed"),
+                "is_quit": bool(report.get("is_quit", False)),
+            }
+        )
+
+    return output
+
+
+async def get_admin_report_detail(session_id: str) -> dict:
+    """Get full interview result detail for admin view."""
+    db = get_db()
+    report = await db[RESULTS].find_one({"session_id": session_id})
+    if not report:
+        raise ValueError("Report not found")
+
+    user_id = report.get("user_id")
+    user_doc = None
+    if user_id:
+        try:
+            user_doc = await db[USERS].find_one({"_id": ObjectId(user_id)})
+        except Exception:
+            user_doc = await db[USERS].find_one({"id": user_id})
+
+    payload = str_objectid(report)
+    payload["user_name"] = (user_doc or {}).get("name", "Unknown")
+    payload["user_email"] = (user_doc or {}).get("email", "Unknown")
+    return payload
+
+
+async def list_admin_users(limit: int = 500) -> list:
+    """List users for admin management with lightweight activity stats."""
+    db = get_db()
+
+    user_cursor = db[USERS].find({"role": "student"}, {"password": 0}).sort("created_at", -1).limit(limit)
+    users = await user_cursor.to_list(length=limit)
+
+    interview_counts = await db[SESSIONS].aggregate([
+        {"$group": {"_id": "$user_id", "count": {"$sum": 1}}},
+    ]).to_list(length=2000)
+    report_counts = await db[RESULTS].aggregate([
+        {"$group": {"_id": "$user_id", "count": {"$sum": 1}}},
+    ]).to_list(length=2000)
+
+    interview_map = {str(item.get("_id")): item.get("count", 0) for item in interview_counts}
+    report_map = {str(item.get("_id")): item.get("count", 0) for item in report_counts}
+
+    output = []
+    for user in users:
+        normalized = str_objectid(user)
+        user_id = normalized.get("id", "")
+        output.append(
+            {
+                "id": user_id,
+                "name": normalized.get("name", ""),
+                "email": normalized.get("email", ""),
+                "role": normalized.get("role", "student"),
+                "created_at": normalized.get("created_at", ""),
+                "interview_count": interview_map.get(user_id, 0),
+                "report_count": report_map.get(user_id, 0),
+            }
+        )
+
+    return output
+
+
+async def delete_admin_user(target_user_id: str, current_admin_user_id: str) -> bool:
+    """Delete a user and associated data. Admin users cannot be deleted from this endpoint."""
+    db = get_db()
+
+    if target_user_id == current_admin_user_id:
+        raise ValueError("You cannot delete your own account")
+
+    user_doc = await db[USERS].find_one({"_id": ObjectId(target_user_id)})
+    if not user_doc:
+        raise ValueError("User not found")
+
+    if user_doc.get("role") == "admin":
+        raise ValueError("Admin users cannot be deleted from this page")
+
+    await db[RESUMES].delete_many({"user_id": target_user_id})
+    await db[SKILLS].delete_many({"user_id": target_user_id})
+    await db[SESSIONS].delete_many({"user_id": target_user_id})
+    await db[ANSWERS].delete_many({"user_id": target_user_id})
+    await db[RESULTS].delete_many({"user_id": target_user_id})
+
+    result = await db[USERS].delete_one({"_id": ObjectId(target_user_id)})
+    return result.deleted_count > 0

backend/services/analytics_service.py

@@ -1,3 +1,5 @@
+from datetime import datetime, timezone
+
 from database import get_db
 from models.collections import RESULTS, SESSIONS, USERS
 from utils.helpers import str_objectid, str_objectids
@@ -27,6 +29,14 @@ async def get_admin_analytics() -> dict:
     # Total students
     total_students = await db[USERS].count_documents({"role": "student"})
 
+    # Users with in-progress interview sessions.
+    active_user_ids = await db[SESSIONS].distinct("user_id", {"status": "in_progress"})
+    live_users = len([uid for uid in active_user_ids if uid])
+
+    # New students created since start of current UTC day.
+    day_start = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0).isoformat()
+    new_users_today = await db[USERS].count_documents({"role": "student", "created_at": {"$gte": day_start}})
+
     # Total interviews
     total_interviews = await db[RESULTS].count_documents({})
 
@@ -74,6 +84,8 @@ async def get_admin_analytics() -> dict:
 
     return {
         "total_students": total_students,
+        "live_users": live_users,
+        "new_users_today": new_users_today,
         "total_interviews": total_interviews,
         "average_score": avg_score,
         "top_performers": top_performers,

backend/services/evaluation_service.py

@@ -1,8 +1,19 @@
 from database import get_db, get_redis
+from bson import ObjectId
 from models.collections import RESULTS, ANSWERS, SESSIONS
 from utils.helpers import utc_now
 from utils.gemini import evaluate_interview
-from services.interview_service import get_session_qa
+from services.interview_service import get_session_qa, cleanup_interview_local_state
+
+
+def _json_safe(value):
+    if isinstance(value, ObjectId):
+        return str(value)
+    if isinstance(value, dict):
+        return {k: _json_safe(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_json_safe(item) for item in value]
+    return value
 
 
 async def generate_report(session_id: str, user_id: str) -> dict:
@@ -15,7 +26,7 @@ async def generate_report(session_id: str, user_id: str) -> dict:
     if existing:
         existing["id"] = str(existing["_id"])
         del existing["_id"]
-        return existing
+        return _json_safe(existing)
 
     # Get session info
     session = await db[SESSIONS].find_one({"session_id": session_id})
@@ -26,6 +37,8 @@ async def generate_report(session_id: str, user_id: str) -> dict:
         raise ValueError("Unauthorized access to session")
 
     role_title = session.get("role_title", "Software Developer")
+    session_status = session.get("status", "completed")
+    quit_at = session.get("quit_at")
 
     # Get all Q&A from Redis
     qa_pairs = await get_session_qa(session_id)
@@ -40,6 +53,9 @@ async def generate_report(session_id: str, user_id: str) -> dict:
         "session_id": session_id,
         "user_id": user_id,
         "role_title": role_title,
+        "session_status": session_status,
+        "is_quit": session_status in {"quit", "quit_with_report"},
+        "quit_at": quit_at,
         "overall_score": evaluation.get("overall_score", 0),
         "total_questions": len(qa_pairs),
         "detailed_scores": evaluation.get("detailed_scores", []),
@@ -48,7 +64,7 @@ async def generate_report(session_id: str, user_id: str) -> dict:
         "recommendations": evaluation.get("recommendations", []),
         "completed_at": utc_now(),
     }
-    await db[RESULTS].insert_one(result_doc)
+    inserted = await db[RESULTS].insert_one(result_doc)
 
     # Store final answers in MongoDB
     for qa in qa_pairs:
@@ -69,6 +85,7 @@ async def generate_report(session_id: str, user_id: str) -> dict:
     keys_to_delete = [
         f"session:{session_id}",
         f"session:{session_id}:questions",
+        f"session:{session_id}:pending_questions",
         f"session:{session_id}:answers",
     ]
     for qid in question_ids:
@@ -78,6 +95,23 @@ async def generate_report(session_id: str, user_id: str) -> dict:
     if keys_to_delete:
         await redis.delete(*keys_to_delete)
 
-    result_doc["id"] = str(result_doc["_id"])
-    del result_doc["_id"]
-    return result_doc
+    if session_status in {"quit", "quit_with_report"}:
+        await db[SESSIONS].update_one(
+            {"session_id": session_id},
+            {
+                "$set": {
+                    "status": "quit_with_report",
+                    "report_generated_at": utc_now(),
+                }
+            },
+        )
+    elif session_status == "completed":
+        await db[SESSIONS].update_one(
+            {"session_id": session_id},
+            {"$set": {"status": "completed_with_report", "report_generated_at": utc_now()}},
+        )
+
+    cleanup_interview_local_state(session_id)
+
+    result_doc["id"] = str(inserted.inserted_id)
+    return _json_safe(result_doc)

backend/services/interview_graph.py

@@ -0,0 +1,105 @@
+from typing import Any, Dict, List, Optional, TypedDict
+
+from langgraph.graph import END, StateGraph
+
+from utils.gemini import generate_interview_question
+
+
+class InterviewGraphState(TypedDict, total=False):
+    role_title: str
+    skills: List[str]
+    previous_questions: List[str]
+    previous_answer: Optional[str]
+    question_count: int
+    max_questions: int
+    current_difficulty: str
+    next_difficulty: str
+    question_stage: str
+    is_complete: bool
+    question_data: Dict[str, Any]
+
+
+FOUNDATION_QUESTION_LIMIT = 3
+
+
+def _difficulty_for_question_number(question_number: int, foundation_limit: int = FOUNDATION_QUESTION_LIMIT) -> str:
+    if question_number <= foundation_limit:
+        return "easy"
+    if question_number <= foundation_limit + 3:
+        return "medium"
+    return "hard"
+
+
+async def _check_completion(state: InterviewGraphState) -> InterviewGraphState:
+    question_count = int(state.get("question_count", 0))
+    max_questions = int(state.get("max_questions", 10))
+    return {"is_complete": question_count >= max_questions}
+
+
+def _route_after_completion(state: InterviewGraphState) -> str:
+    return "end" if state.get("is_complete") else "difficulty"
+
+
+async def _set_next_difficulty(state: InterviewGraphState) -> InterviewGraphState:
+    question_count = int(state.get("question_count", 0))
+    # We are generating the next question, so use question_count + 1.
+    next_question_number = question_count + 1
+    stage = "foundation" if next_question_number <= FOUNDATION_QUESTION_LIMIT else "deep"
+    return {
+        "next_difficulty": _difficulty_for_question_number(next_question_number),
+        "question_stage": stage,
+    }
+
+
+async def _generate_question(state: InterviewGraphState) -> InterviewGraphState:
+    role_title = state.get("role_title", "Software Developer")
+    skills = state.get("skills", ["general"])
+    previous_questions = state.get("previous_questions", [])
+    previous_answer = state.get("previous_answer")
+    difficulty = state.get("next_difficulty", state.get("current_difficulty", "medium"))
+    question_stage = state.get("question_stage", "deep")
+
+    question_data = await generate_interview_question(
+        skills=skills,
+        role_title=role_title,
+        previous_questions=previous_questions,
+        previous_answer=previous_answer,
+        difficulty=difficulty,
+        question_stage=question_stage,
+        foundation_limit=FOUNDATION_QUESTION_LIMIT,
+    )
+
+    return {
+        "question_data": question_data,
+        "current_difficulty": question_data.get("difficulty", difficulty),
+    }
+
+
+def _build_graph():
+    graph = StateGraph(InterviewGraphState)
+
+    graph.add_node("check", _check_completion)
+    graph.add_node("difficulty", _set_next_difficulty)
+    graph.add_node("generate", _generate_question)
+
+    graph.set_entry_point("check")
+    graph.add_conditional_edges(
+        "check",
+        _route_after_completion,
+        {
+            "end": END,
+            "difficulty": "difficulty",
+        },
+    )
+    graph.add_edge("difficulty", "generate")
+    graph.add_edge("generate", END)
+
+    return graph.compile()
+
+
+_INTERVIEW_GRAPH = _build_graph()
+
+
+async def run_interview_graph(state: InterviewGraphState) -> InterviewGraphState:
+    result = await _INTERVIEW_GRAPH.ainvoke(state)
+    return result

backend/services/interview_service.py

@@ -1,46 +1,369 @@
 import json
+import asyncio
 from database import get_db, get_redis
-from models.collections import SESSIONS, JOB_ROLES, SKILLS, QUESTIONS
+from models.collections import SESSIONS, JOB_ROLES, SKILLS, QUESTIONS, TOPICS, TOPIC_QUESTIONS, ROLE_REQUIREMENTS
 from utils.helpers import generate_id, utc_now, str_objectid
-from utils.gemini import generate_interview_question
+from utils.skills import normalize_skill_list, find_matching_skills, find_missing_skills, build_interview_focus_skills
+from services.interview_graph import run_interview_graph
 
 MAX_QUESTIONS = 10
 SESSION_TTL = 7200  # 2 hours
+BATCH_SIZE = 5
+PREGEN_MIN_PENDING = 2
 
+# Local process memory summary requested in workflow.
+_LOCAL_SUMMARIES: dict[str, str] = {}
+_PREGEN_IN_FLIGHT: set[str] = set()
 
-async def start_interview(user_id: str, role_id: str = None) -> dict:
+
+def _safe_json_list(value: str) -> list:
+    try:
+        data = json.loads(value or "[]")
+        return data if isinstance(data, list) else []
+    except Exception:
+        return []
+
+
+def _update_local_summary(session_id: str, question: str, answer: str) -> None:
+    existing = _LOCAL_SUMMARIES.get(session_id, "")
+    combined = f"{existing}\nQ: {question}\nA: {answer}".strip()
+    # Keep summary bounded in memory.
+    _LOCAL_SUMMARIES[session_id] = combined[-1500:]
+
+
+async def _get_generated_question_texts(redis, session_id: str) -> list[str]:
+    qids = await redis.lrange(f"session:{session_id}:questions", 0, -1)
+    questions = []
+    for qid in qids:
+        q = await redis.hgetall(f"session:{session_id}:q:{qid}")
+        if q and q.get("question"):
+            questions.append(q["question"])
+    return questions
+
+
+async def _generate_question_batch(
+    role_title: str,
+    skills: list[str],
+    previous_questions: list[str],
+    generated_count: int,
+    max_questions: int,
+    current_difficulty: str,
+    local_summary: str | None,
+    batch_size: int,
+) -> tuple[list[dict], str]:
+    remaining = max(0, max_questions - generated_count)
+    target = min(batch_size, remaining)
+    if target <= 0:
+        return [], current_difficulty
+
+    generated: list[dict] = []
+    rolling_questions = list(previous_questions)
+    rolling_difficulty = current_difficulty
+    rolling_count = generated_count
+
+    for i in range(target):
+        state = {
+            "role_title": role_title,
+            "skills": skills,
+            "previous_questions": rolling_questions,
+            # Feed the local summary once per batch as extra context.
+            "previous_answer": local_summary if i == 0 else None,
+            "question_count": rolling_count,
+            "max_questions": max_questions,
+            "current_difficulty": rolling_difficulty,
+        }
+        graph_result = await run_interview_graph(state)
+        q_data = graph_result.get("question_data", {})
+        difficulty = q_data.get("difficulty", graph_result.get("current_difficulty", "medium"))
+        generated.append(
+            {
+                "question": q_data.get("question", "Can you explain your approach?"),
+                "difficulty": difficulty,
+                "category": q_data.get("category", "general"),
+            }
+        )
+        rolling_questions.append(generated[-1]["question"])
+        rolling_count += 1
+        rolling_difficulty = difficulty
+
+    return generated, rolling_difficulty
+
+
+async def _append_batch_to_redis(redis, session_id: str, batch: list[dict]) -> list[str]:
+    created_ids: list[str] = []
+    for item in batch:
+        qid = generate_id()
+        created_ids.append(qid)
+        await redis.hset(
+            f"session:{session_id}:q:{qid}",
+            mapping={
+                "question_id": qid,
+                "question": item.get("question", "Can you explain your approach?"),
+                "difficulty": item.get("difficulty", "medium"),
+                "category": item.get("category", "general"),
+            },
+        )
+        await redis.rpush(f"session:{session_id}:questions", qid)
+        await redis.expire(f"session:{session_id}:q:{qid}", SESSION_TTL)
+    if created_ids:
+        await redis.expire(f"session:{session_id}:questions", SESSION_TTL)
+    return created_ids
+
+
+async def _start_topic_interview(user_id: str, topic_id: str) -> dict:
+    """Start a topic-wise interview with admin-created questions."""
+    db = get_db()
+    redis = get_redis()
+
+    topic = await db[TOPICS].find_one({"_id": __import__("bson").ObjectId(topic_id)})
+    if not topic:
+        raise ValueError("Topic not found")
+    if not topic.get("is_published", False):
+        raise ValueError("This topic interview is not published yet")
+
+    topic_questions = await db[TOPIC_QUESTIONS].find({"topic_id": topic_id}).sort("created_at", -1).to_list(length=200)
+    if not topic_questions:
+        raise ValueError("No questions found for selected topic")
+
+    timer_enabled = bool(topic.get("timer_enabled", False))
+    timer_seconds = topic.get("timer_seconds") if timer_enabled else None
+
+    total_questions = min(MAX_QUESTIONS, len(topic_questions))
+    selected = topic_questions[:total_questions]
+
+    session_id = generate_id()
+    _LOCAL_SUMMARIES[session_id] = ""
+
+    session_doc = {
+        "session_id": session_id,
+        "user_id": user_id,
+        "role_id": None,
+        "role_title": topic.get("name", "Topic Interview"),
+        "topic_id": topic_id,
+        "interview_type": "topic",
+        "status": "in_progress",
+        "question_count": 1,
+        "max_questions": total_questions,
+        "current_difficulty": selected[0].get("difficulty", "medium"),
+        "timer_enabled": timer_enabled,
+        "timer_seconds": timer_seconds,
+        "started_at": utc_now(),
+    }
+    await db[SESSIONS].insert_one(session_doc)
+
+    session_state = {
+        "user_id": user_id,
+        "role_title": topic.get("name", "Topic Interview"),
+        "topic_id": topic_id,
+        "interview_type": "topic",
+        "skills": json.dumps([topic.get("name", "general")]),
+        "user_skills": json.dumps([]),
+        "required_skills": json.dumps([]),
+        "matched_skills": json.dumps([]),
+        "missing_skills": json.dumps([]),
+        "question_count": 1,
+        "answered_count": 0,
+        "served_count": 1,
+        "generated_count": total_questions,
+        "max_questions": total_questions,
+        "current_difficulty": selected[0].get("difficulty", "medium"),
+        "timer_enabled": str(timer_enabled),
+        "timer_seconds": str(timer_seconds or ""),
+        "status": "in_progress",
+    }
+    await redis.hset(f"session:{session_id}", mapping=session_state)
+    await redis.expire(f"session:{session_id}", SESSION_TTL)
+
+    created_ids = []
+    for q in selected:
+        qid = generate_id()
+        created_ids.append(qid)
+        await redis.hset(
+            f"session:{session_id}:q:{qid}",
+            mapping={
+                "question_id": qid,
+                "question": q.get("question", "Can you explain this concept?"),
+                "difficulty": q.get("difficulty", "medium"),
+                "category": q.get("category", topic.get("name", "topic")),
+            },
+        )
+        await redis.rpush(f"session:{session_id}:questions", qid)
+        await redis.expire(f"session:{session_id}:q:{qid}", SESSION_TTL)
+    await redis.expire(f"session:{session_id}:questions", SESSION_TTL)
+
+    first_id = created_ids[0]
+    pending_ids = created_ids[1:]
+    if pending_ids:
+        await redis.rpush(f"session:{session_id}:pending_questions", *pending_ids)
+        await redis.expire(f"session:{session_id}:pending_questions", SESSION_TTL)
+
+    first_q_data = await redis.hgetall(f"session:{session_id}:q:{first_id}")
+    return {
+        "session_id": session_id,
+        "interview_type": "topic",
+        "topic": {
+            "topic_id": topic_id,
+            "name": topic.get("name", "Topic Interview"),
+            "description": topic.get("description", ""),
+        },
+        "skill_alignment": {
+            "user_skills": [],
+            "required_skills": [topic.get("name", "")],
+            "matched_skills": [],
+            "missing_skills": [],
+            "interview_focus": [topic.get("name", "")],
+        },
+        "question": {
+            "question_id": first_id,
+            "question": first_q_data.get("question", "Can you explain this concept?"),
+            "difficulty": first_q_data.get("difficulty", "medium"),
+            "question_number": 1,
+            "total_questions": total_questions,
+        },
+        "timer": {
+            "enabled": timer_enabled,
+            "seconds": timer_seconds,
+        },
+        "message": "Topic interview started. Good luck!",
+    }
+
+
+async def _async_pregenerate_next_batch(session_id: str) -> None:
+    redis = get_redis()
+    try:
+        session = await redis.hgetall(f"session:{session_id}")
+        if not session or session.get("status") != "in_progress":
+            return
+
+        pending_len = await redis.llen(f"session:{session_id}:pending_questions")
+        generated_count = int(session.get("generated_count", 0))
+        max_questions = int(session.get("max_questions", MAX_QUESTIONS))
+
+        if pending_len >= PREGEN_MIN_PENDING or generated_count >= max_questions:
+            return
+
+        previous_questions = await _get_generated_question_texts(redis, session_id)
+        skills = _safe_json_list(session.get("skills", "[]"))
+        role_title = session.get("role_title", "Software Developer")
+        current_difficulty = session.get("current_difficulty", "medium")
+        local_summary = _LOCAL_SUMMARIES.get(session_id)
+
+        batch, last_difficulty = await _generate_question_batch(
+            role_title=role_title,
+            skills=skills,
+            previous_questions=previous_questions,
+            generated_count=generated_count,
+            max_questions=max_questions,
+            current_difficulty=current_difficulty,
+            local_summary=local_summary,
+            batch_size=BATCH_SIZE,
+        )
+        if not batch:
+            return
+
+        new_ids = await _append_batch_to_redis(redis, session_id, batch)
+        if new_ids:
+            await redis.rpush(f"session:{session_id}:pending_questions", *new_ids)
+            await redis.expire(f"session:{session_id}:pending_questions", SESSION_TTL)
+            await redis.hset(
+                f"session:{session_id}",
+                mapping={
+                    "generated_count": str(generated_count + len(new_ids)),
+                    "current_difficulty": last_difficulty,
+                },
+            )
+    finally:
+        _PREGEN_IN_FLIGHT.discard(session_id)
+
+
+def _schedule_pregen(session_id: str, answered_count: int) -> None:
+    # Start pre-generation after user answers Q1 and Q2, then keep it topped up.
+    if answered_count < 2:
+        return
+    if session_id in _PREGEN_IN_FLIGHT:
+        return
+    _PREGEN_IN_FLIGHT.add(session_id)
+    asyncio.create_task(_async_pregenerate_next_batch(session_id))
+
+
+async def start_interview(
+    user_id: str,
+    role_id: str = None,
+    custom_role: str = None,
+    interview_type: str = "resume",
+    topic_id: str = None,
+) -> dict:
     """Start a new interview session."""
+    interview_type = (interview_type or "resume").strip().lower()
+    if interview_type == "topic":
+        if not topic_id:
+            raise ValueError("topic_id is required for topic interviews")
+        return await _start_topic_interview(user_id=user_id, topic_id=topic_id)
+
     db = get_db()
     redis = get_redis()
 
     # Get user skills
     skills_doc = await db[SKILLS].find_one({"user_id": user_id})
-    skills = skills_doc.get("skills", ["general"]) if skills_doc else ["general"]
+    user_skills = skills_doc.get("skills", ["general"]) if skills_doc else ["general"]
+    user_skills = normalize_skill_list(user_skills)
 
     # Get role
     role_title = "Software Developer"
-    if role_id:
+    if custom_role:
+        role_title = custom_role
+    elif role_id:
         from bson import ObjectId
-        role = await db[JOB_ROLES].find_one({"_id": ObjectId(role_id)})
-        if role:
-            role_title = role["title"]
+        try:
+            role = await db[JOB_ROLES].find_one({"_id": ObjectId(role_id)})
+            if role:
+                role_title = role["title"]
+        except Exception:
+            # If it's not a valid ObjectId, assume it's a raw generic title passed from frontend
+            role_title = role_id
+
+    # Compare role requirements with user skills when admin role requirements exist.
+    required_skills = []
+    if role_id and not custom_role:
+        req_cursor = db[ROLE_REQUIREMENTS].find({"role_id": role_id})
+        req_docs = await req_cursor.to_list(length=100)
+        required_skills = [d.get("skill", "") for d in req_docs if d.get("skill")]
+
+    matched_role_skills = find_matching_skills(user_skills, required_skills)
+    missing_role_skills = find_missing_skills(user_skills, required_skills)
+
+    # Prioritize matched required skills and compress them into cluster-aware focus areas.
+    base_skills_for_interview = matched_role_skills if matched_role_skills else user_skills
+    skills_for_interview = build_interview_focus_skills(base_skills_for_interview)
+    if not skills_for_interview:
+        skills_for_interview = ["general"]
 
     # Check for existing questions in question bank
     bank_questions = []
-    if role_id:
-        cursor = db[QUESTIONS].find({"role_id": role_id}).limit(5)
-        async for q in cursor:
-            bank_questions.append(q["question"])
-
-    # Generate first question
-    question_data = await generate_interview_question(
-        skills=skills,
+    if role_id and not custom_role:
+        try:
+            cursor = db[QUESTIONS].find({"role_id": role_id}).limit(5)
+            async for q in cursor:
+                bank_questions.append(q["question"])
+        except Exception:
+            pass
+
+    # Workflow: generate first batch upfront, store in Redis, serve Q1.
+    initial_batch, last_difficulty = await _generate_question_batch(
         role_title=role_title,
-        difficulty="medium",
+        skills=skills_for_interview,
+        previous_questions=[],
+        generated_count=0,
+        max_questions=MAX_QUESTIONS,
+        current_difficulty="medium",
+        local_summary=None,
+        batch_size=BATCH_SIZE,
     )
+    if not initial_batch:
+        raise ValueError("Failed to generate initial interview questions")
 
     session_id = generate_id()
-    question_id = generate_id()
+    _LOCAL_SUMMARIES[session_id] = ""
 
     # Create session in MongoDB
     session_doc = {
@@ -49,9 +372,10 @@ async def start_interview(user_id: str, role_id: str = None) -> dict:
         "role_id": role_id,
         "role_title": role_title,
         "status": "in_progress",
+        "interview_type": "resume",
         "question_count": 1,
         "max_questions": MAX_QUESTIONS,
-        "current_difficulty": "medium",
+        "current_difficulty": initial_batch[0].get("difficulty", "medium"),
         "started_at": utc_now(),
     }
     await db[SESSIONS].insert_one(session_doc)
@@ -60,36 +384,53 @@ async def start_interview(user_id: str, role_id: str = None) -> dict:
     session_state = {
         "user_id": user_id,
         "role_title": role_title,
-        "skills": json.dumps(skills),
+        "skills": json.dumps(skills_for_interview),
+        "user_skills": json.dumps(user_skills),
+        "required_skills": json.dumps(normalize_skill_list(required_skills)),
+        "matched_skills": json.dumps(matched_role_skills),
+        "missing_skills": json.dumps(missing_role_skills),
         "question_count": 1,
+        "answered_count": 0,
+        "served_count": 1,
+        "generated_count": len(initial_batch),
         "max_questions": MAX_QUESTIONS,
-        "current_difficulty": "medium",
+        "current_difficulty": last_difficulty,
+        "interview_type": "resume",
         "status": "in_progress",
     }
     await redis.hset(f"session:{session_id}", mapping=session_state)
     await redis.expire(f"session:{session_id}", SESSION_TTL)
 
-    # Store question in Redis
-    q_data = {
-        "question_id": question_id,
-        "question": question_data.get("question", "Tell me about yourself."),
-        "difficulty": question_data.get("difficulty", "medium"),
-        "category": question_data.get("category", "general"),
-    }
-    await redis.hset(f"session:{session_id}:q:{question_id}", mapping=q_data)
-    await redis.rpush(f"session:{session_id}:questions", question_id)
-    await redis.expire(f"session:{session_id}:q:{question_id}", SESSION_TTL)
-    await redis.expire(f"session:{session_id}:questions", SESSION_TTL)
+    # Store batch in Redis and queue remaining for later serving.
+    new_ids = await _append_batch_to_redis(redis, session_id, initial_batch)
+    first_id = new_ids[0]
+    pending_ids = new_ids[1:]
+    if pending_ids:
+        await redis.rpush(f"session:{session_id}:pending_questions", *pending_ids)
+        await redis.expire(f"session:{session_id}:pending_questions", SESSION_TTL)
+
+    first_q_data = await redis.hgetall(f"session:{session_id}:q:{first_id}")
 
     return {
         "session_id": session_id,
+        "skill_alignment": {
+            "user_skills": user_skills,
+            "required_skills": normalize_skill_list(required_skills),
+            "matched_skills": matched_role_skills,
+            "missing_skills": missing_role_skills,
+            "interview_focus": skills_for_interview,
+        },
         "question": {
-            "question_id": question_id,
-            "question": q_data["question"],
-            "difficulty": q_data["difficulty"],
+            "question_id": first_id,
+            "question": first_q_data.get("question", "Tell me about yourself."),
+            "difficulty": first_q_data.get("difficulty", "medium"),
             "question_number": 1,
             "total_questions": MAX_QUESTIONS,
         },
+        "timer": {
+            "enabled": False,
+            "seconds": None,
+        },
         "message": "Interview started. Good luck!",
     }
 
@@ -118,12 +459,23 @@ async def submit_answer(session_id: str, question_id: str, answer: str) -> dict:
     await redis.expire(f"session:{session_id}:answers", SESSION_TTL)
 
     question_count = int(session.get("question_count", 1))
+    answered_count = int(session.get("answered_count", 0)) + 1
+    served_count = int(session.get("served_count", 1))
+    generated_count = int(session.get("generated_count", 0))
     max_questions = int(session.get("max_questions", MAX_QUESTIONS))
+    interview_type = session.get("interview_type", "resume")
+
+    # Update local summary in-memory (requested local summary step).
+    current_q = await redis.hgetall(f"session:{session_id}:q:{question_id}")
+    _update_local_summary(session_id, current_q.get("question", ""), answer)
 
     # Check if interview is complete
-    if question_count >= max_questions:
+    if answered_count >= max_questions:
         # Mark session as completed
-        await redis.hset(f"session:{session_id}", "status", "completed")
+        await redis.hset(
+            f"session:{session_id}",
+            mapping={"status": "completed", "answered_count": str(answered_count)},
+        )
         await db[SESSIONS].update_one(
             {"session_id": session_id},
             {"$set": {"status": "completed", "completed_at": utc_now()}},
@@ -135,74 +487,142 @@ async def submit_answer(session_id: str, question_id: str, answer: str) -> dict:
             "message": "Interview complete! Generating your report...",
         }
 
-    # Adjust difficulty based on question count
-    difficulty = _adjust_difficulty(question_count, session.get("current_difficulty", "medium"))
-
-    # Get previous questions from Redis
-    question_ids = await redis.lrange(f"session:{session_id}:questions", 0, -1)
-    previous_questions = []
-    for qid in question_ids:
-        q = await redis.hgetall(f"session:{session_id}:q:{qid}")
-        if q:
-            previous_questions.append(q.get("question", ""))
-
-    # Get the current question text for context
-    current_q = await redis.hgetall(f"session:{session_id}:q:{question_id}")
-
-    skills = json.loads(session.get("skills", "[]"))
-    role_title = session.get("role_title", "Software Developer")
-
-    # Generate next question
-    next_question_data = await generate_interview_question(
-        skills=skills,
-        role_title=role_title,
-        previous_questions=previous_questions,
-        previous_answer=answer,
-        difficulty=difficulty,
-    )
-
-    new_question_id = generate_id()
+    # Serve from pending queue first.
+    next_question_id = await redis.lpop(f"session:{session_id}:pending_questions")
+
+    # If queue is empty, generate only for resume interviews.
+    if not next_question_id:
+        if interview_type == "topic":
+            await redis.hset(
+                f"session:{session_id}",
+                mapping={"status": "completed", "answered_count": str(answered_count)},
+            )
+            await db[SESSIONS].update_one(
+                {"session_id": session_id},
+                {"$set": {"status": "completed", "completed_at": utc_now()}},
+            )
+            return {
+                "session_id": session_id,
+                "next_question": None,
+                "is_complete": True,
+                "message": "Interview complete! Generating your report...",
+            }
+
+        previous_questions = await _get_generated_question_texts(redis, session_id)
+        skills = _safe_json_list(session.get("skills", "[]"))
+        role_title = session.get("role_title", "Software Developer")
+
+        sync_batch, last_difficulty = await _generate_question_batch(
+            role_title=role_title,
+            skills=skills,
+            previous_questions=previous_questions,
+            generated_count=generated_count,
+            max_questions=max_questions,
+            current_difficulty=session.get("current_difficulty", "medium"),
+            local_summary=_LOCAL_SUMMARIES.get(session_id),
+            batch_size=BATCH_SIZE,
+        )
+        new_ids = await _append_batch_to_redis(redis, session_id, sync_batch)
+        generated_count += len(new_ids)
+        if new_ids:
+            next_question_id = new_ids[0]
+            if len(new_ids) > 1:
+                await redis.rpush(f"session:{session_id}:pending_questions", *new_ids[1:])
+                await redis.expire(f"session:{session_id}:pending_questions", SESSION_TTL)
+            await redis.hset(
+                f"session:{session_id}",
+                mapping={
+                    "generated_count": str(generated_count),
+                    "current_difficulty": last_difficulty,
+                },
+            )
+
+    if not next_question_id:
+        raise ValueError("Unable to fetch or generate next question")
+
+    q_data = await redis.hgetall(f"session:{session_id}:q:{next_question_id}")
+    next_difficulty = q_data.get("difficulty", session.get("current_difficulty", "medium"))
     new_count = question_count + 1
-
-    # Store new question in Redis
-    q_data = {
-        "question_id": new_question_id,
-        "question": next_question_data.get("question", "Can you elaborate further?"),
-        "difficulty": next_question_data.get("difficulty", difficulty),
-        "category": next_question_data.get("category", "general"),
-    }
-    await redis.hset(f"session:{session_id}:q:{new_question_id}", mapping=q_data)
-    await redis.rpush(f"session:{session_id}:questions", new_question_id)
-    await redis.expire(f"session:{session_id}:q:{new_question_id}", SESSION_TTL)
+    new_served_count = served_count + 1
 
     # Update session state
     await redis.hset(f"session:{session_id}", mapping={
         "question_count": str(new_count),
-        "current_difficulty": difficulty,
+        "answered_count": str(answered_count),
+        "served_count": str(new_served_count),
+        "current_difficulty": next_difficulty,
     })
 
+    if interview_type == "resume":
+        _schedule_pregen(session_id, answered_count)
+
     return {
         "session_id": session_id,
         "next_question": {
-            "question_id": new_question_id,
-            "question": q_data["question"],
-            "difficulty": q_data["difficulty"],
-            "question_number": new_count,
+            "question_id": next_question_id,
+            "question": q_data.get("question", "Can you elaborate further?"),
+            "difficulty": q_data.get("difficulty", "medium"),
+            "question_number": new_served_count,
             "total_questions": max_questions,
         },
         "is_complete": False,
-        "message": f"Question {new_count} of {max_questions}",
+        "message": f"Question {new_served_count} of {max_questions}",
     }
 
 
-def _adjust_difficulty(question_number: int, current: str) -> str:
-    """Dynamically adjust difficulty based on progress."""
-    if question_number <= 3:
-        return "easy"
-    elif question_number <= 6:
-        return "medium"
-    else:
-        return "hard"
+async def quit_interview(session_id: str, user_id: str) -> dict:
+    """Mark an interview as quit and indicate whether a partial report can be generated."""
+    db = get_db()
+    redis = get_redis()
+
+    session = await db[SESSIONS].find_one({"session_id": session_id})
+    if not session:
+        raise ValueError("Session not found")
+    if session.get("user_id") != user_id:
+        raise ValueError("Unauthorized access to session")
+
+    if session.get("status") in {"completed", "quit", "quit_with_report"}:
+        return {
+            "session_id": session_id,
+            "report_generated": session.get("status") == "quit_with_report",
+            "message": "Interview already finalized",
+        }
+
+    quit_at = utc_now()
+
+    # Update Redis state if still present.
+    redis_session_key = f"session:{session_id}"
+    redis_session = await redis.hgetall(redis_session_key)
+    answered_count = int(redis_session.get("answered_count", 0)) if redis_session else 0
+    if redis_session:
+        await redis.hset(
+            redis_session_key,
+            mapping={
+                "status": "quit",
+                "quit_at": quit_at,
+            },
+        )
+        await redis.expire(redis_session_key, SESSION_TTL)
+
+    # Persist quit metadata for admin visibility.
+    await db[SESSIONS].update_one(
+        {"session_id": session_id},
+        {
+            "$set": {
+                "status": "quit",
+                "quit_at": quit_at,
+                "quit_reason": "user_requested",
+                "answered_count": answered_count,
+            }
+        },
+    )
+
+    has_answers = answered_count > 0
+    return {
+        "session_id": session_id,
+        "report_generated": has_answers,
+        "message": "Interview quit successfully" if has_answers else "Interview quit. No answers to evaluate yet.",
+    }
 
 
 async def get_session_qa(session_id: str) -> list:
@@ -225,3 +645,9 @@ async def get_session_qa(session_id: str) -> list:
             })
 
     return qa_pairs
+
+
+def cleanup_interview_local_state(session_id: str) -> None:
+    """Cleanup process-local state for a completed session."""
+    _LOCAL_SUMMARIES.pop(session_id, None)
+    _PREGEN_IN_FLIGHT.discard(session_id)

backend/services/resume_service.py

@@ -4,6 +4,8 @@ from database import get_db
 from models.collections import RESUMES, SKILLS
 from utils.helpers import utc_now, str_objectid
 from utils.gemini import parse_resume_with_gemini
+from utils.resume_text import extract_resume_text
+from utils.skills import normalize_skill_list
 from config import get_settings
 
 settings = get_settings()
@@ -20,12 +22,14 @@ async def upload_and_parse_resume(user_id: str, filename: str, file_content: byt
     async with aiofiles.open(file_path, "wb") as f:
         await f.write(file_content)
 
-    # Read file text (for parsing)
-    resume_text = file_content.decode("utf-8", errors="ignore")
+    # Extract readable text by file type before sending to Gemini.
+    resume_text = extract_resume_text(filename, file_content)
 
     # Parse with Gemini
     parsed_data = await parse_resume_with_gemini(resume_text)
-    skills = parsed_data.get("skills", [])
+    raw_skills = parsed_data.get("skills", [])
+    skills = normalize_skill_list(raw_skills)
+    parsed_data["skills"] = skills
 
     # Upsert resume document
     resume_doc = {
@@ -50,6 +54,7 @@ async def upload_and_parse_resume(user_id: str, filename: str, file_content: byt
         {"$set": {
             "user_id": user_id,
             "skills": skills,
+            "raw_skills": raw_skills,
             "updated_at": utc_now(),
         }},
         upsert=True,

backend/utils/gemini.py

@@ -1,5 +1,9 @@
 from google import genai
 from config import get_settings
+from utils.skills import normalize_skill_list
+import json
+import re
+from langchain_core.prompts import PromptTemplate
 
 settings = get_settings()
 
@@ -11,6 +15,7 @@ async def call_gemini(prompt: str, system_instruction: str = None) -> str:
     config = {}
     if system_instruction:
         config["system_instruction"] = system_instruction
+    config["response_mime_type"] = "application/json"
 
     response = client.models.generate_content(
         model=settings.GEMINI_MODEL,
@@ -20,14 +25,62 @@ async def call_gemini(prompt: str, system_instruction: str = None) -> str:
     return response.text
 
 
+def _extract_json_object(text: str) -> str:
+    value = (text or "").strip()
+    if value.startswith("```"):
+        value = value.split("\n", 1)[1]
+    if value.endswith("```"):
+        value = value.rsplit("```", 1)[0]
+    value = value.strip()
+
+    if value.startswith("{") and value.endswith("}"):
+        return value
+
+    # Fallback when model wraps JSON with extra text.
+    start = value.find("{")
+    end = value.rfind("}")
+    if start != -1 and end != -1 and end > start:
+        return value[start:end + 1]
+
+    return value
+
+
+def _fallback_skill_scan(resume_text: str) -> list:
+    common = [
+        "python", "java", "javascript", "typescript", "react", "next.js", "node.js",
+        "fastapi", "django", "flask", "spring", "mongodb", "postgresql", "mysql",
+        "redis", "docker", "kubernetes", "aws", "gcp", "azure", "git", "linux",
+        "rest api", "graphql", "machine learning", "data analysis", "sql",
+    ]
+    text = (resume_text or "").lower()
+    found = []
+    for skill in common:
+        pattern = r"\b" + re.escape(skill.lower()) + r"\b"
+        if re.search(pattern, text):
+            found.append(skill)
+    return normalize_skill_list(found)
+
+
 async def parse_resume_with_gemini(resume_text: str) -> dict:
     """Parse resume text and extract structured data using Gemini."""
     prompt = f"""Analyze the following resume and extract structured information.
-Return a JSON object with these fields:
-- "skills": list of technical and soft skills
-- "experience_summary": brief summary of work experience
-- "education": list of educational qualifications
-- "projects": list of notable projects
+CRITICAL INSTRUCTION FOR SKILLS:
+1) Extract concrete tools/technologies/frameworks/languages from the resume text.
+2) Exclude vague traits such as "hardworking", "leadership", "problem solving", "communication".
+3) If a line contains multiple skills (comma-separated), split them into separate list items.
+4) Do NOT add skills that are not present in the resume.
+
+Return a JSON object with these exact fields:
+- "name": full name of the candidate (string or null)
+- "email": candidate's email address (string or null)
+- "phone": candidate's phone number (string or null)
+- "location": candidate's location/address (string or null)
+- "skills": list of technical and soft skills verbatim from the text (array of strings)
+- "recommended_roles": list of 3-5 recommended job role titles the user is qualified for based on these skills (array of strings)
+- "experience_summary": brief summary of work experience (string)
+- "experience": list of dictionaries, each with "company", "role", "duration", and "description"
+- "education": list of dictionaries, each with "institution", "degree", "graduation_year"
+- "projects": list of dictionaries, each with "name" and "description"
 
 Resume text:
 ---
@@ -37,19 +90,37 @@ Resume text:
 Return ONLY valid JSON, no markdown formatting."""
 
     result = await call_gemini(prompt)
-    # Clean up markdown code blocks if present
-    result = result.strip()
-    if result.startswith("```"):
-        result = result.split("\n", 1)[1]
-    if result.endswith("```"):
-        result = result.rsplit("```", 1)[0]
-    result = result.strip()
-
-    import json
+    result = _extract_json_object(result)
+
     try:
-        return json.loads(result)
+        parsed = json.loads(result)
+        parsed.setdefault("name", None)
+        parsed.setdefault("email", None)
+        parsed.setdefault("phone", None)
+        parsed.setdefault("location", None)
+        parsed.setdefault("recommended_roles", [])
+        parsed.setdefault("experience_summary", "")
+        parsed.setdefault("experience", [])
+        parsed.setdefault("education", [])
+        parsed.setdefault("projects", [])
+
+        parsed["skills"] = normalize_skill_list(parsed.get("skills", []))
+        if not parsed["skills"]:
+            parsed["skills"] = _fallback_skill_scan(resume_text)
+        return parsed
     except json.JSONDecodeError:
-        return {"skills": [], "experience_summary": result, "education": [], "projects": []}
+        return {
+            "name": None,
+            "email": None,
+            "phone": None,
+            "location": None,
+            "skills": _fallback_skill_scan(resume_text),
+            "recommended_roles": [],
+            "experience_summary": result, 
+            "experience": [],
+            "education": [], 
+            "projects": []
+        }
 
 
 async def generate_interview_question(
@@ -58,9 +129,13 @@ async def generate_interview_question(
     previous_questions: list = None,
     previous_answer: str = None,
     difficulty: str = "medium",
+    question_stage: str = "deep",
+    foundation_limit: int = 3,
 ) -> dict:
     """Generate an interview question using Gemini."""
-    context = f"Role: {role_title}\nCandidate Skills: {', '.join(skills)}\nDifficulty: {difficulty}"
+    context = f"Role: {role_title}\nCandidate Skill Focus Areas: {', '.join(skills)}\nDifficulty: {difficulty}"
+    context += f"\nCurrent Stage: {question_stage}"
+    context += f"\nFoundation Question Limit: {foundation_limit}"
 
     if previous_questions:
         context += f"\n\nPrevious questions asked (do NOT repeat these):\n"
@@ -71,13 +146,19 @@ async def generate_interview_question(
         context += f"\nCandidate's last answer: {previous_answer}"
         context += "\nGenerate a follow-up question based on this answer to probe deeper."
 
-    prompt = f"""{context}
+    prompt_template = PromptTemplate.from_template(
+        """{context}
 
 Generate ONE interview question for this candidate. The question should:
 1. Be relevant to the role and candidate's skills
 2. Match the {difficulty} difficulty level
 3. Be clear and specific
 4. Test practical knowledge
+5. If a skill is a cluster label like "Deep Learning (CNN, LSTM)", pick one member skill from that cluster and ask a concrete question on it
+6. Rotate topics to avoid repeatedly asking from the same cluster
+7. If Current Stage is "foundation": ask only core/fundamental basics
+8. If Current Stage is "deep": DO NOT ask basic definition/foundation questions; ask applied, scenario-based, debugging, optimization, or trade-off questions only
+9. Treat Foundation Question Limit as a strict cap: once foundation stage is done, never return to foundation-style prompts
 
 Return ONLY a JSON object with:
 - "question": the interview question text
@@ -85,16 +166,10 @@ Return ONLY a JSON object with:
 - "category": the skill category this tests
 
 Return ONLY valid JSON, no markdown formatting."""
+    )
+    prompt = prompt_template.format(context=context, difficulty=difficulty)
 
-    result = await call_gemini(prompt)
-    result = result.strip()
-    if result.startswith("```"):
-        result = result.split("\n", 1)[1]
-    if result.endswith("```"):
-        result = result.rsplit("```", 1)[0]
-    result = result.strip()
-
-    import json
+    result = _extract_json_object(await call_gemini(prompt))
     try:
         return json.loads(result)
     except json.JSONDecodeError:
@@ -111,7 +186,8 @@ async def evaluate_interview(questions_and_answers: list, role_title: str) -> di
     for i, qa in enumerate(questions_and_answers, 1):
         qa_text += f"\nQ{i}: {qa['question']}\nA{i}: {qa['answer']}\n"
 
-    prompt = f"""You are an expert technical interviewer evaluating a candidate for the role: {role_title}
+    prompt_template = PromptTemplate.from_template(
+        """You are an expert technical interviewer evaluating a candidate for the role: {role_title}
 
 Here are the interview questions and the candidate's answers:
 {qa_text}
@@ -128,16 +204,10 @@ Evaluate the candidate and return a JSON object with:
 - "recommendations": list of 3-5 actionable recommendations
 
 Be fair but thorough. Return ONLY valid JSON, no markdown formatting."""
+    )
+    prompt = prompt_template.format(role_title=role_title, qa_text=qa_text)
 
-    result = await call_gemini(prompt)
-    result = result.strip()
-    if result.startswith("```"):
-        result = result.split("\n", 1)[1]
-    if result.endswith("```"):
-        result = result.rsplit("```", 1)[0]
-    result = result.strip()
-
-    import json
+    result = _extract_json_object(await call_gemini(prompt))
     try:
         return json.loads(result)
     except json.JSONDecodeError:

backend/utils/resume_text.py

@@ -0,0 +1,36 @@
+import io
+
+
+def _extract_pdf_text(file_content: bytes) -> str:
+    from pypdf import PdfReader
+
+    reader = PdfReader(io.BytesIO(file_content))
+    pages = []
+    for page in reader.pages:
+        pages.append(page.extract_text() or "")
+    return "\n".join(pages)
+
+
+def _extract_docx_text(file_content: bytes) -> str:
+    from docx import Document
+
+    doc = Document(io.BytesIO(file_content))
+    paragraphs = [p.text for p in doc.paragraphs if p.text and p.text.strip()]
+    return "\n".join(paragraphs)
+
+
+def extract_resume_text(filename: str, file_content: bytes) -> str:
+    ext = (filename or "").lower().rsplit(".", 1)
+    ext = f".{ext[-1]}" if len(ext) > 1 else ""
+
+    if ext == ".pdf":
+        text = _extract_pdf_text(file_content)
+    elif ext == ".docx":
+        text = _extract_docx_text(file_content)
+    else:
+        # Fallback path for txt/doc and unknown formats.
+        text = file_content.decode("utf-8", errors="ignore")
+
+    cleaned = text.replace("\x00", " ")
+    cleaned = "\n".join(line.strip() for line in cleaned.splitlines() if line.strip())
+    return cleaned

backend/utils/skills.py

@@ -0,0 +1,288 @@
+import re
+from typing import Iterable, List, Set
+
+
+_SKILL_ALIASES = {
+    "node": "Node.js",
+    "nodejs": "Node.js",
+    "node.js": "Node.js",
+    "express": "Express.js",
+    "expressjs": "Express.js",
+    "express.js": "Express.js",
+    "react": "React",
+    "reactjs": "React",
+    "react.js": "React",
+    "next": "Next.js",
+    "nextjs": "Next.js",
+    "next.js": "Next.js",
+    "js": "JavaScript",
+    "javascript": "JavaScript",
+    "ts": "TypeScript",
+    "typescript": "TypeScript",
+    "py": "Python",
+    "mongo": "MongoDB",
+    "mongodb": "MongoDB",
+    "postgres": "PostgreSQL",
+    "postgresql": "PostgreSQL",
+    "mysql": "MySQL",
+    "aws": "AWS",
+    "gcp": "GCP",
+    "azure": "Azure",
+    "ci/cd": "CI/CD",
+    "ci cd": "CI/CD",
+    "rest": "REST API",
+    "rest api": "REST API",
+    "fastapi": "FastAPI",
+    "langchain": "LangChain",
+    "langgraph": "LangGraph",
+    "langsmith": "LangSmith",
+    "rag": "RAG",
+    "rag pipeline": "RAG Pipelines",
+    "rag pipelines": "RAG Pipelines",
+    "chromadb": "ChromaDB",
+    "scikit learn": "Scikit-learn",
+    "scikit-learn": "Scikit-learn",
+    "pytorch": "PyTorch",
+    "llama": "Llama",
+    "llama 4": "Llama 4",
+    "gemini api": "Gemini API",
+    "sentence transformers": "Sentence Transformers",
+    "e5 multilingual embeddings": "E5 Multilingual Embeddings",
+    "cnn": "CNN",
+    "cnns": "CNN",
+    "rnn": "RNN",
+    "rnns": "RNN",
+    "gan": "GAN",
+    "gans": "GAN",
+    "bert": "BERT",
+    "bert fine tuning": "BERT Fine-tuning",
+    "ocr": "OCR",
+    "ocr based extraction": "OCR Based Extraction",
+    "k means": "K-Means",
+    "cross validation": "Cross-validation",
+    "oop": "OOP",
+    "ml": "Machine Learning",
+}
+
+
+_SKILL_CLUSTER_RULES = [
+    (
+        "Deep Learning",
+        ["cnn", "rnn", "lstm", "gru", "gan", "transformers", "bert", "pytorch", "tensorflow", "encoder decoder"],
+    ),
+    (
+        "Machine Learning",
+        [
+            "machine learning",
+            "random forest",
+            "svm",
+            "logistic regression",
+            "linear regression",
+            "k means",
+            "model evaluation",
+            "cross validation",
+            "scikit learn",
+        ],
+    ),
+    (
+        "LLM and GenAI",
+        [
+            "langchain",
+            "langgraph",
+            "langsmith",
+            "prompt engineering",
+            "rag",
+            "rag pipeline",
+            "rag pipelines",
+            "semantic search",
+            "gemini api",
+            "llama",
+            "embedding models",
+            "e5 multilingual embeddings",
+            "sentence transformers",
+        ],
+    ),
+    (
+        "Data and Databases",
+        ["sql", "mysql", "postgresql", "mongodb", "pinecone", "chromadb", "vector similarity search"],
+    ),
+    (
+        "Backend and APIs",
+        ["python", "java", "javascript", "typescript", "fastapi", "django", "flask", "node", "express", "rest api"],
+    ),
+    (
+        "Cloud and DevOps",
+        ["docker", "kubernetes", "aws", "gcp", "azure", "git", "github", "ci cd"],
+    ),
+    (
+        "Document AI and OCR",
+        ["ocr", "ocr based extraction", "document extraction"],
+    ),
+]
+
+
+def _normalize_key(value: str) -> str:
+    value = value.strip().lower()
+    value = re.sub(r"[\u2010-\u2015]", "-", value)
+    value = value.replace("&", " and ")
+    value = re.sub(r"[^a-z0-9+#.\-/ ]+", " ", value)
+    value = value.replace("/", " ")
+    value = value.replace("-", " ")
+    value = re.sub(r"\s+", " ", value).strip()
+    return value
+
+
+def canonicalize_skill(skill: str) -> str:
+    if not isinstance(skill, str):
+        return ""
+
+    cleaned = skill.strip()
+    if not cleaned:
+        return ""
+
+    normalized = _normalize_key(cleaned)
+    if normalized in _SKILL_ALIASES:
+        return _SKILL_ALIASES[normalized]
+
+    # Keep all-caps acronyms readable (e.g., SQL, API, OOP).
+    if cleaned.isupper() and len(cleaned) <= 6:
+        return cleaned
+
+    return " ".join(part.capitalize() for part in normalized.split(" "))
+
+
+def _split_skill_chunks(skill: str) -> List[str]:
+    if not isinstance(skill, str):
+        return []
+
+    parts = re.split(r",|\||;", skill)
+    chunks = []
+    for part in parts:
+        candidate = part.strip()
+        if not candidate:
+            continue
+        chunks.append(candidate)
+    return chunks
+
+
+def normalize_skill_list(skills: Iterable[str], limit: int = 80) -> List[str]:
+    unique: List[str] = []
+    seen: Set[str] = set()
+
+    for raw in skills or []:
+        for token in _split_skill_chunks(raw):
+            canon = canonicalize_skill(token)
+            if not canon:
+                continue
+            key = _normalize_key(canon)
+            if key in seen:
+                continue
+            seen.add(key)
+            unique.append(canon)
+            if len(unique) >= limit:
+                return unique
+
+    return unique
+
+
+def _classify_cluster(skill: str) -> str | None:
+    key = _normalize_key(skill)
+    if not key:
+        return None
+
+    for cluster_name, rules in _SKILL_CLUSTER_RULES:
+        for rule in rules:
+            if rule in key or key in rule:
+                return cluster_name
+    return None
+
+
+def cluster_skills(skills: Iterable[str], max_members_per_cluster: int = 4) -> List[dict]:
+    """Return grouped skills with compact labels for UI and prompting."""
+    normalized = normalize_skill_list(skills)
+    grouped: dict[str, list[str]] = {}
+
+    for skill in normalized:
+        cluster_name = _classify_cluster(skill)
+        if not cluster_name:
+            continue
+        grouped.setdefault(cluster_name, [])
+        if skill not in grouped[cluster_name]:
+            grouped[cluster_name].append(skill)
+
+    # Prefer denser clusters first for cleaner UX.
+    ordered = sorted(grouped.items(), key=lambda item: len(item[1]), reverse=True)
+
+    result = []
+    for cluster_name, members in ordered:
+        sampled = members[:max_members_per_cluster]
+        label = f"{cluster_name} ({', '.join(sampled)})"
+        result.append(
+            {
+                "cluster": cluster_name,
+                "members": members,
+                "label": label,
+                "count": len(members),
+            }
+        )
+
+    return result
+
+
+def build_interview_focus_skills(skills: Iterable[str], max_clusters: int = 6, max_extras: int = 2) -> List[str]:
+    """Build a compact, cluster-aware skill list for interview question generation."""
+    normalized = normalize_skill_list(skills)
+    grouped = cluster_skills(normalized)
+
+    focus = [g["label"] for g in grouped[:max_clusters]]
+
+    # Add a couple of non-clustered items so niche tools are not ignored.
+    extras = []
+    clustered_members = {m for g in grouped for m in g["members"]}
+    for skill in normalized:
+        if skill in clustered_members:
+            continue
+        extras.append(skill)
+        if len(extras) >= max_extras:
+            break
+
+    combined = focus + extras
+    return combined if combined else normalized[: max_clusters + max_extras]
+
+
+def skill_match(candidate_skill: str, required_skill: str) -> bool:
+    c_key = _normalize_key(canonicalize_skill(candidate_skill))
+    r_key = _normalize_key(canonicalize_skill(required_skill))
+    if not c_key or not r_key:
+        return False
+    if c_key == r_key:
+        return True
+
+    # Soft phrase matching for related forms like "rest api" vs "restful api".
+    if c_key in r_key or r_key in c_key:
+        return True
+
+    return False
+
+
+def find_matching_skills(candidate_skills: Iterable[str], required_skills: Iterable[str]) -> List[str]:
+    matched: List[str] = []
+    for req in required_skills or []:
+        for cand in candidate_skills or []:
+            if skill_match(cand, req):
+                matched.append(canonicalize_skill(req))
+                break
+    return normalize_skill_list(matched)
+
+
+def find_missing_skills(candidate_skills: Iterable[str], required_skills: Iterable[str]) -> List[str]:
+    missing: List[str] = []
+    for req in required_skills or []:
+        has_match = False
+        for cand in candidate_skills or []:
+            if skill_match(cand, req):
+                has_match = True
+                break
+        if not has_match:
+            missing.append(canonicalize_skill(req))
+    return normalize_skill_list(missing)