Spaces:
Running
Running
| """In-memory chat session store with sliding idle timeout (default 10 minutes). | |
| Uses last_active timestamps for forget-behavior so Gradio history is still | |
| trusted on the first message of a session (TTLCache key absence alone must | |
| NOT wipe UI history). TTLCache max age is a longer GC safety net. | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from cachetools import TTLCache | |
| import config | |
| _TIMEOUT = max(60, int(config.SESSION_TIMEOUT_SECONDS)) | |
| # Keep entries around longer than the idle timeout so we can detect expiry | |
| # via last_active (instead of treating "missing key" as brand-new). | |
| _sessions: TTLCache = TTLCache( | |
| maxsize=1000, | |
| ttl=max(3600, _TIMEOUT * 12), | |
| ) | |
| def is_alive(session_id: str) -> bool: | |
| return bool(session_id) and session_id in _sessions | |
| def get(session_id: str) -> dict: | |
| if not session_id: | |
| return {} | |
| data = _sessions.get(session_id) | |
| return dict(data) if data else {} | |
| def idle_expired(session_id: str) -> bool: | |
| """True only when we had a session and it sat idle past the timeout.""" | |
| data = get(session_id) | |
| if not data: | |
| return False | |
| last = data.get("last_active") | |
| if last is None: | |
| return False | |
| return (time.time() - float(last)) > _TIMEOUT | |
| def touch( | |
| session_id: str, | |
| *, | |
| product_focus: dict | None = None, | |
| customer_profile: dict | None = None, | |
| history: list | None = None, | |
| ) -> dict: | |
| """Save session data and reset the idle timer.""" | |
| if not session_id: | |
| return {} | |
| data = get(session_id) | |
| if product_focus is not None: | |
| # Store a JSON-friendly copy (category terms as list) | |
| focus = dict(product_focus) | |
| cat = focus.get("category") | |
| if isinstance(cat, tuple): | |
| focus["category"] = [cat[0], cat[1], list(cat[2])] | |
| data["product_focus"] = focus | |
| if customer_profile is not None: | |
| profile = dict(customer_profile) | |
| cat = profile.get("category") | |
| if isinstance(cat, tuple): | |
| profile["category"] = [cat[0], cat[1], list(cat[2])] | |
| data["customer_profile"] = profile | |
| if history is not None: | |
| trimmed = history[-config.MAX_HISTORY_TURNS :] if history else [] | |
| data["history"] = trimmed | |
| data["last_active"] = time.time() | |
| _sessions[session_id] = data | |
| return data | |
| def clear(session_id: str) -> None: | |
| if session_id and session_id in _sessions: | |
| del _sessions[session_id] | |