core/session.py

"""Session management for FixMyNeighborhood.

Provides:
- Session-scoped orchestrator management
- Multi-user isolation via gr.State
- Session ID generation
"""

import hashlib
import time
from typing import Optional

from config import ANTHROPIC_API_KEY
from agents.controller import AutonomousController


def generate_session_id(orchestrator: Optional[AutonomousController] = None) -> str:
    """
    Generate a unique session ID for rate limiting and audit.

    Args:
        orchestrator: Optional orchestrator instance for stable ID

    Returns:
        16-character hex session ID
    """
    if orchestrator:
        seed = str(id(orchestrator))
    else:
        seed = str(time.time())
    return hashlib.sha256(seed.encode()).hexdigest()[:16]


def get_or_create_orchestrator(
    session_orchestrator: Optional[AutonomousController]
) -> Optional[AutonomousController]:
    """
    Get existing session orchestrator or create new one.

    Uses gr.State to maintain orchestrator per browser session:
    - Same user (same tab): Reuses orchestrator for multi-turn context
    - Different users (different tabs/browsers): Separate orchestrators

    Args:
        session_orchestrator: Existing orchestrator from gr.State (None if first request)

    Returns:
        Orchestrator instance for this session, or None if API key not configured
    """
    if session_orchestrator is not None:
        return session_orchestrator

    if not ANTHROPIC_API_KEY:
        return None

    try:
        return AutonomousController()
    except Exception as e:
        print(f"Orchestrator creation error: {e}")
        return None


class SessionManager:
    """
    Manages session state for multi-user isolation.

    In Gradio, each browser tab gets its own gr.State values.
    This class provides additional session management utilities.
    """

    def __init__(self):
        self._active_sessions: dict = {}

    def create_session(self) -> str:
        """Create a new session and return its ID."""
        session_id = generate_session_id()
        self._active_sessions[session_id] = {
            "created_at": time.time(),
            "last_activity": time.time(),
        }
        return session_id

    def update_activity(self, session_id: str) -> None:
        """Update last activity time for a session."""
        if session_id in self._active_sessions:
            self._active_sessions[session_id]["last_activity"] = time.time()

    def cleanup_stale_sessions(self, max_age_seconds: float = 3600) -> int:
        """
        Remove sessions older than max_age.

        Args:
            max_age_seconds: Maximum session age in seconds (default 1 hour)

        Returns:
            Number of sessions cleaned up
        """
        now = time.time()
        stale = [
            sid for sid, data in self._active_sessions.items()
            if now - data["last_activity"] > max_age_seconds
        ]
        for sid in stale:
            del self._active_sessions[sid]
        return len(stale)

    def get_active_count(self) -> int:
        """Get count of active sessions."""
        return len(self._active_sessions)