File size: 1,605 Bytes
fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d 76398fd fbd808d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | """
In-memory session history store with write-time trimming and TTL eviction.
"""
import time
from services.observability import setup_logger
logger = setup_logger(__name__)
MAX_HISTORY_MESSAGES = 10
SESSION_TTL_SECONDS = 3600
_sessions: dict[str, dict] = {}
def get_history(session_id: str) -> list[dict]:
"""Return mutable message list for this session, creating or evicting if needed."""
now = time.time()
entry = _sessions.get(session_id)
if entry is None or (now - entry["ts"]) > SESSION_TTL_SECONDS:
_sessions[session_id] = {"messages": [], "ts": now}
else:
_sessions[session_id]["ts"] = now
return _sessions[session_id]["messages"]
def append_message(session_id: str, message: dict) -> None:
"""Append a message and immediately trim if exceeding the maximum history limit."""
history = get_history(session_id)
history.append(message)
if len(history) > MAX_HISTORY_MESSAGES:
_sessions[session_id]["messages"] = history[-MAX_HISTORY_MESSAGES:]
def get_context_window(session_id: str, limit: int = 6) -> list[dict]:
"""Return the last `limit` messages for injection into an LLM context."""
return list(get_history(session_id)[-limit:])
def clear_session(session_id: str) -> None:
"""Explicitly clear a session (e.g. user logout)."""
_sessions.pop(session_id, None)
def active_session_count() -> int:
"""Returns number of active (non-expired) sessions for monitoring."""
now = time.time()
return sum(
1 for e in _sessions.values()
if (now - e["ts"]) <= SESSION_TTL_SECONDS
)
|