python-services/conversation_manager.py

# conversation_manager.py
# Manages conversation sessions — memory within an active voice call.
#
# v2 — arabic_clarify removed
#   get_arabic_clarification_flag() removed from ConversationSession.
#   All callers (voice_pipeline._build_prompt, llm_server._build_prompt)
#   updated to not pass arabic_clarification to build_system_prompt.

import time
import uuid
import logging
from typing import List, Dict, Any, Optional

from language_config import normalize_language
from utils import needs70B
from language_detector import SmartLanguageDetector
from caller_info_extractor import CallerInfoCollector

logger = logging.getLogger("conversation_manager")

SESSION_TIMEOUT_SECONDS = 30 * 60
MAX_HISTORY_MESSAGES = 20


class ConversationSession:
    """Single conversation session tracking an active phone call instance."""

    def __init__(
        self,
        session_id: str,
        company_name: str,
        agent_name: str,
        agent_gender: str = "male",
        custom_rules: Optional[List[str]] = None,
        document_context: str = "",
        caller_collection_fields: Optional[List[Dict[str, Any]]] = None,
    ):
        self.session_id = session_id
        self.company_name = company_name
        self.agent_name = agent_name
        self.agent_gender = agent_gender
        self.custom_rules = custom_rules or []
        self.document_context = document_context
        self.caller_collection_fields: List[Dict[str, Any]] = caller_collection_fields or []

        self.messages: List[Dict[str, str]] = []
        self.language_detector = SmartLanguageDetector()
        self.caller_info = CallerInfoCollector(fields_spec=self.caller_collection_fields)
        self.created_at = time.time()
        self.last_active = time.time()
        self.turn_count = 0
        self.greeting_streak = 0

        # Database mapping properties
        self.company_id = None
        self.company_name_ur = None
        self.overview = None
        self.overview_ur = None
        self.agent_name_ur = None
        self.agent_role = None
        self.agent_role_ur = None
        self.phone_line_id = None
        self.line_token = None
        self.agent_config_id = None
        self.rag_enabled = False

    @classmethod
    def from_tenant_config(cls, config: Dict[str, Any], session_id: str) -> "ConversationSession":
        """Creates a ConversationSession from a database configuration dictionary."""
        rules_raw = config.get("system_instruction_rules") or ""
        custom_rules = []
        if isinstance(rules_raw, list):
            custom_rules = rules_raw
        elif isinstance(rules_raw, str):
            custom_rules = [r.strip() for r in rules_raw.split(".") if r.strip()]

        session = cls(
            session_id=session_id,
            company_name=config.get("company_name") or "Our Company",
            agent_name=config.get("agent_name") or "Sara",
            agent_gender=config.get("agent_gender") or "male",
            custom_rules=custom_rules,
            document_context="",
            caller_collection_fields=config.get("caller_details_to_collect") or []
        )

        session.company_id = config.get("company_id")
        session.company_name_ur = config.get("company_name_ur")
        session.overview = config.get("overview")
        session.overview_ur = config.get("overview_ur")
        session.agent_name_ur = config.get("agent_name_ur")
        session.agent_role = config.get("agent_role")
        session.agent_role_ur = config.get("agent_role_ur")
        session.phone_line_id = config.get("phone_line_id")
        session.line_token = config.get("line_token")
        session.agent_config_id = config.get("agent_config_id")
        session.rag_enabled = config.get("rag_enabled", False)

        return session

    def add_message(self, role: str, content: str) -> None:
        self.messages.append({"role": role, "content": content})
        self.last_active = time.time()
        if len(self.messages) > MAX_HISTORY_MESSAGES:
            self.messages = self.messages[-MAX_HISTORY_MESSAGES:]

    def process_user_message(self, text: str) -> str:
        """
        Process user message — update language detector and caller info.
        Returns the resolved language code string.
        """
        self.turn_count += 1
        self.last_active = time.time()

        language = normalize_language(self.language_detector.process_message(text))
        self.caller_info.process_message(text, language)
        return language

    def classify_intent(self, text: str, lang_code: str) -> str:
        quick_exact = [
            "yes", "no", "hello", "hi", "hey", "hold on", "wait", "stop", "ok",
            "okay", "thanks", "thank you", "جی", "ہاں", "نا", "شکریہ", "نہیں"
        ]
        normalized = (text or "").strip().lower()
        effective_lang = normalize_language(lang_code)

        if normalized in quick_exact or len(normalized) <= 6:
            return "fast"

        return "heavy" if needs70B(text, effective_lang) else "fast"

    def get_current_language(self) -> str:
        return normalize_language(self.language_detector.get_language())

    # REMOVED: get_arabic_clarification_flag()
    # arabic_clarify tracking has been removed entirely (Urdu-only mode).
    # prompt_builder.py no longer accepts arabic_clarification parameter.
    # Re-enable alongside multi-language mode if needed.

    def is_expired(self) -> bool:
        return (time.time() - self.last_active) > SESSION_TIMEOUT_SECONDS

    def get_history_for_api(self) -> List[Dict[str, str]]:
        return self.messages.copy()

    def to_summary(self) -> Dict[str, Any]:
        return {
            "session_id": self.session_id,
            "company_id": self.company_id,
            "company": self.company_name,
            "agent": self.agent_name,
            "agent_gender": self.agent_gender,
            "language": self.get_current_language(),
            "turns": self.turn_count,
            "caller_info": self.caller_info.to_dict(),
            "duration_seconds": int(time.time() - self.created_at),
        }


class ConversationManager:
    """Manages active voice-channel conversation tracks."""

    def __init__(self):
        self.sessions: Dict[str, ConversationSession] = {}
        logger.info("ConversationManager tracking pool initialized successfully.")

    def create_session(
        self,
        company_name: str,
        agent_name: str,
        agent_gender: str = "male",
        custom_rules: List[str] = None,
        document_context: str = "",
        session_id: Optional[str] = None,
        caller_collection_fields: Optional[List[Dict[str, Any]]] = None,
    ) -> ConversationSession:
        sid = session_id or str(uuid.uuid4())

        session = ConversationSession(
            session_id=sid,
            company_name=company_name,
            agent_name=agent_name,
            agent_gender=agent_gender,
            custom_rules=custom_rules,
            document_context=document_context,
            caller_collection_fields=caller_collection_fields,
        )

        self.sessions[sid] = session
        logger.info(f"Session established: {sid} [{company_name} / Agent: {agent_name}]")
        return session

    def get_session(self, session_id: str) -> Optional[ConversationSession]:
        session = self.sessions.get(session_id)
        if session:
            if session.is_expired():
                logger.info(f"Evicting expired communication stack: {session_id}")
                del self.sessions[session_id]
                return None
            return session
        return None

    def end_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        session = self.sessions.pop(session_id, None)
        if session:
            summary = session.to_summary()
            logger.info(f"Session terminated via clean closure: {session_id} — Summary metrics: {summary}")
            return summary
        return None

    def cleanup_expired(self) -> None:
        expired = [
            sid for sid, session in self.sessions.items()
            if session.is_expired()
        ]
        for sid in expired:
            logger.info(f"Garbage collection cycle purged expired session: {sid}")
            del self.sessions[sid]
        if expired:
            logger.info(f"Garbage collection cycle purged {len(expired)} expired session allocations.")

    def active_count(self) -> int:
        return len(self.sessions)