hkuds--deeptutor/deeptutor/services/session/base_session_manager.py

#!/usr/bin/env python
"""
BaseSessionManager - Unified session management base class.

This module provides a base class for managing persistent sessions
across different agent modules (solve, chat, etc.).

Features:
- Consistent JSON storage format
- Session CRUD operations
- Message management within sessions
- Automatic session ordering (newest first)
- Configurable session limits
"""

from abc import ABC, abstractmethod
import json
import time
from typing import Any
import uuid


class BaseSessionManager(ABC):
    """
    Abstract base class for session management.

    Provides common functionality for storing and retrieving sessions,
    with customization points for module-specific behavior.

    Path resolution is deferred to request-time via @property so that
    multi-user isolation works correctly: ``get_path_service()`` is
    called on every access, respecting the per-request user context.

    Subclasses must implement:
    - _get_session_id_prefix(): Return the session ID prefix (e.g., "solve_", "chat_")
    - _get_default_title(): Return the default title for new sessions
    - _create_session_data(): Create module-specific session data structure
    - _get_session_summary(): Create module-specific session summary for listing
    """

    MAX_SESSIONS = 100

    def __init__(self, module_name: str):
        self.module_name = module_name

    @property
    def path_service(self):
        from deeptutor.services.path_service import get_path_service

        return get_path_service()

    @property
    def sessions_file(self):
        return self.path_service.get_session_file(self.module_name)

    def _ensure_file(self) -> None:
        self.sessions_file.parent.mkdir(parents=True, exist_ok=True)
        if not self.sessions_file.exists():
            initial_data = {
                "version": "1.0",
                "sessions": [],
            }
            self._save_data(initial_data)

    # =========================================================================
    # Abstract Methods - Must be implemented by subclasses
    # =========================================================================

    @abstractmethod
    def _get_session_id_prefix(self) -> str:
        pass

    @abstractmethod
    def _get_default_title(self) -> str:
        pass

    @abstractmethod
    def _create_session_data(self, **kwargs) -> dict[str, Any]:
        pass

    @abstractmethod
    def _get_session_summary(self, session: dict[str, Any]) -> dict[str, Any]:
        pass

    # =========================================================================
    # File Operations
    # =========================================================================

    def _load_data(self) -> dict[str, Any]:
        self._ensure_file()
        try:
            with open(self.sessions_file, encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, FileNotFoundError):
            return {"version": "1.0", "sessions": []}

    def _save_data(self, data: dict[str, Any]) -> None:
        self.sessions_file.parent.mkdir(parents=True, exist_ok=True)
        with open(self.sessions_file, "w", encoding="utf-8") as f:
            json.dump(data, f, indent=2, ensure_ascii=False)

    def _get_sessions(self) -> list[dict[str, Any]]:
        data = self._load_data()
        return data.get("sessions", [])

    def _save_sessions(self, sessions: list[dict[str, Any]]) -> None:
        data = self._load_data()
        data["sessions"] = sessions
        self._save_data(data)

    # =========================================================================
    # Session CRUD Operations
    # =========================================================================

    def create_session(
        self,
        title: str | None = None,
        **kwargs,
    ) -> dict[str, Any]:
        prefix = self._get_session_id_prefix()
        session_id = f"{prefix}{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        now = time.time()

        if title is None:
            title = self._get_default_title()

        session = {
            "session_id": session_id,
            "title": title[:100] if title else self._get_default_title(),
            "messages": [],
            "created_at": now,
            "updated_at": now,
        }

        module_data = self._create_session_data(**kwargs)
        session.update(module_data)

        sessions = self._get_sessions()
        sessions.insert(0, session)

        if len(sessions) > self.MAX_SESSIONS:
            sessions = sessions[: self.MAX_SESSIONS]

        self._save_sessions(sessions)

        return session

    def get_session(self, session_id: str) -> dict[str, Any] | None:
        sessions = self._get_sessions()
        for session in sessions:
            if session.get("session_id") == session_id:
                return session
        return None

    def update_session(
        self,
        session_id: str,
        messages: list[dict[str, Any]] | None = None,
        title: str | None = None,
        **kwargs,
    ) -> dict[str, Any] | None:
        sessions = self._get_sessions()

        for i, session in enumerate(sessions):
            if session.get("session_id") == session_id:
                if messages is not None:
                    session["messages"] = messages
                if title is not None:
                    session["title"] = title[:100]

                for key, value in kwargs.items():
                    if value is not None:
                        session[key] = value

                session["updated_at"] = time.time()

                sessions.pop(i)
                sessions.insert(0, session)

                self._save_sessions(sessions)
                return session

        return None

    def add_message(
        self,
        session_id: str,
        role: str,
        content: str,
        **metadata,
    ) -> dict[str, Any] | None:
        session = self.get_session(session_id)
        if not session:
            return None

        message = {
            "role": role,
            "content": content,
            "timestamp": time.time(),
        }

        for key, value in metadata.items():
            if value is not None:
                message[key] = value

        messages = session.get("messages", [])
        messages.append(message)

        title = None
        if session.get("title") == self._get_default_title() and role == "user":
            title = content[:50] + ("..." if len(content) > 50 else "")

        return self.update_session(session_id, messages=messages, title=title)

    def list_sessions(
        self,
        limit: int = 20,
        include_messages: bool = False,
    ) -> list[dict[str, Any]]:
        sessions = self._get_sessions()[:limit]

        if not include_messages:
            return [self._get_session_summary(s) for s in sessions]

        return sessions

    def delete_session(self, session_id: str) -> bool:
        sessions = self._get_sessions()
        original_count = len(sessions)

        sessions = [s for s in sessions if s.get("session_id") != session_id]

        if len(sessions) < original_count:
            self._save_sessions(sessions)
            return True

        return False

    def clear_all_sessions(self) -> int:
        sessions = self._get_sessions()
        count = len(sessions)
        self._save_sessions([])
        return count

    # =========================================================================
    # Utility Methods
    # =========================================================================

    def get_session_count(self) -> int:
        return len(self._get_sessions())

    def session_exists(self, session_id: str) -> bool:
        return self.get_session(session_id) is not None


__all__ = ["BaseSessionManager"]