vent_agent_3/service/session_manager.py

from dataclasses import dataclass, field
import threading
from typing import Dict, Any, Optional, List


@dataclass
class VentMemoryState:
    """In-memory state for a single ventilation-agent conversation session."""

    messages: List[Dict[str, Any]] = field(default_factory=list)
    recent_queries: List[str] = field(default_factory=list)
    last_summary: str = ""


class SessionManager:
    """Thread-safe registry of per-session VentMemoryState instances."""

    def __init__(self) -> None:
        self._sessions: Dict[str, VentMemoryState] = {}
        self._lock = threading.Lock()

    def get_or_create(self, session_id: str) -> VentMemoryState:
        """Return existing state for *session_id*, creating a fresh one if missing."""
        with self._lock:
            if session_id not in self._sessions:
                self._sessions[session_id] = VentMemoryState()
            return self._sessions[session_id]

    def update(self, session_id: str, state: VentMemoryState) -> None:
        """Overwrite the state for a given session."""
        with self._lock:
            self._sessions[session_id] = state

    def remove(self, session_id: str) -> None:
        """Drop a session from the registry."""
        with self._lock:
            self._sessions.pop(session_id, None)


session_manager = SessionManager()