intent_engine.py

"""
RUBRA — intent_engine.py
Adaptive Intent Recognition System

- Predicts what user ACTUALLY wants (not just literal words)
- Tracks intent history per session → learns patterns
- Self-updates routing decisions within chat
- Handles ambiguous requests gracefully
"""

import re
import json
import logging
from typing import Optional, List, Dict, Tuple
from dataclasses import dataclass, field

log = logging.getLogger("rubra.intent")

# ═══════════════════════════════════════════════════════
#  INTENT CATEGORIES
# ═══════════════════════════════════════════════════════

INTENTS = {
    # Conversational
    "greeting":     {"agents": ["fast"],    "priority": 10, "patterns": [r'\b(hi|hello|hey|salam|kemon|how are you|what\'s up)\b']},
    "farewell":     {"agents": ["fast"],    "priority": 10, "patterns": [r'\b(bye|goodbye|thanks|thank you|see you)\b']},
    "casual":       {"agents": ["fast"],    "priority": 8,  "patterns": [r'\b(how|what do you think|opinion|feel|like)\b']},

    # Technical
    "coding":       {"agents": ["coding"],  "priority": 9,  "patterns": [
        r'\b(write|build|create|make|code|program|function|class|api|app|website|script)\b',
        r'\b(debug|fix|error|bug|issue|broken|not working)\b',
        r'\b(python|javascript|react|html|css|sql|bash|typescript)\b',
    ]},
    "code_explain": {"agents": ["coding"],  "priority": 8,  "patterns": [
        r'\b(explain|understand|what does|how does|review|analyze)\b.*\b(code|function|class|script)\b',
    ]},
    "run_code":     {"agents": ["coding"],  "priority": 9,  "patterns": [
        r'\b(run|execute|test|output|result)\b.{0,20}\b(this|it|code|script)\b',
        r'\b(run it|run this|execute this|test this)\b',
    ]},

    # Knowledge
    "search":       {"agents": ["search"],  "priority": 8,  "patterns": [
        r'\b(weather|temperature|forecast)\b',
        r'\b(price|bitcoin|crypto|stock|exchange rate)\b',
        r'\b(news|latest|current|today|recent|trending)\b',
    ]},
    "research":     {"agents": ["browse"],  "priority": 7,  "patterns": [
        r'\b(research|find|look up|search for|browse|visit)\b',
        r'https?://',
        r'\b(who is|what is|tell me about)\b.{5,50}',
    ]},
    "education":    {"agents": ["tutor"],   "priority": 7,  "patterns": [
        r'\b(explain|teach|how does|what is|learn|study|understand)\b',
        r'\b(math|physics|chemistry|biology|history|geography)\b',
        r'\b(class|exam|homework|assignment|ssc|hsc|nctb)\b',
    ]},

    # Creative
    "creative":     {"agents": ["general"], "priority": 6,  "patterns": [
        r'\b(write|create|generate|compose)\b.{0,20}\b(story|poem|essay|article|email|letter)\b',
    ]},

    # File operations
    "file_analyze": {"agents": ["vision"],  "priority": 9,  "patterns": [
        r'\b(analyze|read|extract|summarize|what is in)\b.{0,20}\b(file|image|pdf|document|photo)\b',
    ]},

    # HyperFrames — HTML/CSS to MP4 (feature-flagged, see video_tool.py)
    "video":        {"agents": ["coding"],  "priority": 8,  "patterns": [
        r'\b(video|render|animate|mp4|motion graphic)\b',
    ]},

    # Project generation
    "project_gen":  {"agents": ["coding"],  "priority": 9,  "patterns": [
        r'\b(generate|create|build|make)\b.{0,30}\b(project|app|website|system|platform|store)\b',
        r'\b(full.?stack|nextjs|react app|fastapi|landing page|e.?commerce|online store)\b',
        r'\b(separate.{0,15}files|multiple files|run the files)\b',
    ]},

    # General
    "general":      {"agents": ["general"], "priority": 1,  "patterns": []},
}

# Compile all patterns
_COMPILED: Dict[str, List] = {
    intent: [re.compile(p, re.IGNORECASE) for p in data["patterns"]]
    for intent, data in INTENTS.items()
}

# ═══════════════════════════════════════════════════════
#  CONTEXT SIGNALS — detect implicit intent from history
# ═══════════════════════════════════════════════════════

def extract_context_signals(hist: List[Dict]) -> Dict:
    """Extract signals from conversation history."""
    signals = {
        "last_intent":      None,
        "has_code":         False,
        "has_file":         False,
        "topic":            None,
        "user_expertise":   "intermediate",
        "preferred_lang":   "en",
        "consecutive_coding": 0,
    }
    if not hist:
        return signals

    coding_count = 0
    for h in hist[-10:]:
        content = h.get("content", "").lower()
        role    = h.get("role", "")

        if role == "assistant":
            if re.search(r'```(python|javascript|jsx|html|css|sql)', content):
                signals["has_code"] = True
                coding_count += 1

        if role == "user":
            if re.search(r'[\u0980-\u09FF]', content):
                signals["preferred_lang"] = "bn"
            if re.search(r'\b(expert|senior|advanced|professional)\b', content):
                signals["user_expertise"] = "expert"
            elif re.search(r'\b(beginner|new|learning|student)\b', content):
                signals["user_expertise"] = "beginner"

    signals["consecutive_coding"] = coding_count
    return signals


# ═══════════════════════════════════════════════════════
#  INTENT CLASSIFIER
# ═══════════════════════════════════════════════════════

@dataclass
class IntentResult:
    intent:     str
    confidence: float        # 0.0 - 1.0
    agent:      str
    reasoning:  str = ""
    sub_intent: Optional[str] = None

def classify_intent(
    message:  str,
    hist:     List[Dict] = None,
    signals:  Dict = None,
) -> IntentResult:
    """
    Classify user intent using:
    1. Pattern matching (fast)
    2. Context signals from history
    3. Implicit intent detection
    """
    if hist is None: hist = []
    if signals is None: signals = extract_context_signals(hist)

    lower   = message.lower().strip()
    scores: Dict[str, float] = {}

    # Score each intent by pattern matches
    for intent, patterns in _COMPILED.items():
        if not patterns:
            scores[intent] = 0.1
            continue
        match_count = sum(1 for p in patterns if p.search(lower))
        if match_count > 0:
            base_score  = INTENTS[intent]["priority"] * match_count
            scores[intent] = base_score

    # Context boosters
    # "project_gen" vs "coding" frequently TIE on requests like "make a website
    # for me" — both patterns match equally. Whoever happens to come first in
    # the INTENTS dict used to win by accident, and that was "coding", which
    # routes to a free-chat agent that asks clarifying questions instead of
    # the templates pipeline that just builds with sensible defaults. Project
    # generation should always win ties — it's the more specific, more useful
    # outcome and never stalls waiting for answers.
    if scores.get("project_gen", -1) >= scores.get("coding", -1) and "project_gen" in scores:
        scores["project_gen"] += 0.01

    if signals.get("has_code") and re.search(r'\b(run|test|execute|it|this)\b', lower):
        scores["run_code"] = scores.get("run_code", 0) + 15

    if signals.get("consecutive_coding", 0) >= 2:
        scores["coding"] = scores.get("coding", 0) + 5

    # Detect "continue" / "more" → same as last intent
    if re.search(r'\b(continue|more|next|keep going|go on|abar|aro)\b', lower):
        last = signals.get("last_intent")
        if last and last in scores:
            scores[last] += 20

    # Pick highest score
    if not scores or max(scores.values()) <= 0:
        return IntentResult(
            intent="general", confidence=0.5,
            agent="general", reasoning="No pattern match — defaulting to general"
        )

    best_intent  = max(scores, key=scores.get)
    best_score   = scores[best_intent]
    total        = sum(scores.values()) or 1
    confidence   = min(best_score / total, 1.0)

    agent = INTENTS[best_intent]["agents"][0]

    return IntentResult(
        intent     = best_intent,
        confidence = confidence,
        agent      = agent,
        reasoning  = f"Score {best_score:.1f} from {len([s for s in scores.values() if s > 0])} matching patterns",
    )


# ═══════════════════════════════════════════════════════
#  ADAPTIVE INTENT TRACKER — learns within session
# ═══════════════════════════════════════════════════════

class AdaptiveIntentTracker:
    """
    Tracks intent patterns per session.
    Adapts routing decisions based on user behavior.
    """

    def __init__(self):
        # session_id → {intents: [], corrections: [], preferences: {}}
        self._sessions: Dict[str, Dict] = {}

    def _get_session(self, sid: str) -> Dict:
        if sid not in self._sessions:
            self._sessions[sid] = {
                "intents":     [],
                "corrections": [],
                "preferences": {},
                "turn_count":  0,
            }
        return self._sessions[sid]

    def track(self, sid: str, message: str, intent: IntentResult):
        """Record intent for this turn."""
        s = self._get_session(sid)
        s["intents"].append({
            "message":    message[:80],
            "intent":     intent.intent,
            "confidence": intent.confidence,
            "agent":      intent.agent,
        })
        s["turn_count"] += 1
        # Keep last 20
        s["intents"] = s["intents"][-20:]

    def get_session_context(self, sid: str) -> Dict:
        """Get session-level context for better predictions."""
        s = self._get_session(sid)
        intents = s["intents"]
        if not intents:
            return {}

        # Most common intent
        from collections import Counter
        intent_counts = Counter(i["intent"] for i in intents)
        dominant      = intent_counts.most_common(1)[0][0] if intent_counts else "general"

        # Last 3 intents
        recent = [i["intent"] for i in intents[-3:]]

        return {
            "dominant_intent": dominant,
            "recent_intents":  recent,
            "turn_count":      s["turn_count"],
            "last_intent":     recent[-1] if recent else None,
        }

    def predict_next_intent(self, sid: str, message: str, hist: List[Dict]) -> IntentResult:
        """
        Predict intent using both pattern matching and session history.
        Self-corrects based on patterns learned in this session.
        """
        session_ctx = self.get_session_context(sid)
        signals     = extract_context_signals(hist)

        # Inject session context into signals
        if session_ctx.get("last_intent"):
            signals["last_intent"] = session_ctx["last_intent"]

        result = classify_intent(message, hist, signals)

        # Session-level correction:
        # If user has been coding for 3+ turns, "run this" should definitely be coding
        if (session_ctx.get("dominant_intent") == "coding"
                and result.intent == "general"
                and re.search(r'\b(run|test|execute|it|this|try)\b', message, re.I)):
            result = IntentResult(
                intent     = "run_code",
                confidence = 0.85,
                agent      = "coding",
                reasoning  = "Session context: user in coding mode",
            )

        self.track(sid, message, result)
        return result


# Global tracker instance
intent_tracker = AdaptiveIntentTracker()