""" session_store.py ──────────────── Async, fire-and-forget JSON-based chat session storage. Each browser session gets its own JSON file in data/sessions/. Rules: • NEVER raises exceptions that propagate upward — any failure is silently logged. • ALL disk I/O is done with asyncio to avoid blocking the event loop. • Thread-safe: uses an asyncio.Lock per session_id to avoid race conditions. """ import asyncio import json import logging import os from datetime import datetime, timezone from pathlib import Path from typing import Dict from app.core.config import get_settings logger = logging.getLogger(__name__) settings = get_settings() # ── Directory where session JSON files are stored ─────────────────────────── def _get_sessions_dir() -> Path: """Returns the current sessions directory from settings.""" return settings.sessions_dir # ── Per-session locks to prevent concurrent write corruption ───────────────── _session_locks: Dict[str, asyncio.Lock] = {} _locks_meta_lock = asyncio.Lock() # protects the _session_locks dict itself def _get_session_path(session_id: str) -> Path: """Returns the absolute path for a session's JSON file.""" return _get_sessions_dir() / f"{session_id}.json" def _now_iso() -> str: """Returns the current UTC time as an ISO-8601 string.""" return datetime.now(timezone.utc).isoformat() async def _get_lock(session_id: str) -> asyncio.Lock: """Gets or creates a per-session asyncio.Lock in a thread-safe manner.""" async with _locks_meta_lock: if session_id not in _session_locks: _session_locks[session_id] = asyncio.Lock() return _session_locks[session_id] def _ensure_sessions_dir() -> None: """Creates the sessions directory if it does not exist. Sync, called once.""" try: _get_sessions_dir().mkdir(parents=True, exist_ok=True) except OSError as exc: logger.error("session_store: cannot create sessions dir: %s", exc) # Call once at import time so the directory is always ready. _ensure_sessions_dir() async def ensure_session(session_id: str) -> None: """ Creates the session JSON file if it doesn't exist yet. Safe to call on every page load — does nothing if file exists. """ path = _get_session_path(session_id) if path.exists(): return lock = await _get_lock(session_id) async with lock: # Double-check after acquiring lock if path.exists(): return skeleton = { "session_id": session_id, "user_name": "default name", "created_at": _now_iso(), "last_active": _now_iso(), "message_count": 0, "current_chat_id": 1, "messages": [], "status": "AI", "blocked": False, "rating": None, "chat_ratings": {}, "alert_id": 0, "alert_type": "", "archived": False, "archived_chats": [], "sla_metrics": {}, } try: await asyncio.to_thread(_write_json, path, skeleton) logger.info("session_store: created session %s", session_id) except Exception as exc: logger.error("session_store: failed to create session %s: %s", session_id, exc) async def save_message(session_id: str, question: str, answer: str, chat_id: int = None) -> None: """ Appends a Question+Answer pair with timestamps to the session's JSON file. Fire-and-forget safe — caller should wrap in asyncio.create_task(). Args: chat_id: If provided, message is saved to this specific chat thread. If None, uses current_chat_id from the session file. """ path = _get_session_path(session_id) logger.info("session_store: save_message called for session_id=%s, path=%s", session_id, path) lock = await _get_lock(session_id) async with lock: try: # Load current data if path.exists(): data = await asyncio.to_thread(_read_json, path) else: # Session file missing — recreate it gracefully data = { "session_id": session_id, "user_name": "default name", "created_at": _now_iso(), "last_active": _now_iso(), "message_count": 0, "current_chat_id": 1, "messages": [], "status": "AI", "blocked": False, "rating": None, "chat_ratings": {}, "alert_id": 0, "alert_type": "", "archived": False, "archived_chats": [], "sla_metrics": {}, } timestamp = _now_iso() # Use explicit chat_id if provided, otherwise fall back to current if chat_id is None: chat_id = data.get("current_chat_id", 1) # --- SLA Agent Response Tracking --- if data.get("status") == "HUMAN_TAKEOVER" and not question and answer: if "sla_metrics" not in data: data["sla_metrics"] = {} if "agent_reply_times" not in data["sla_metrics"]: data["sla_metrics"]["agent_reply_times"] = [] # Find the last message from the user in this chat. # Skip if the last message was already an admin reply (no question), # which means the admin is sending a follow-up without a new user # question in between — we shouldn't double-count that. last_user_msg = None for m in reversed(data["messages"]): if m.get("chat_id", 1) != chat_id: continue if m.get("question"): last_user_msg = m break if not m.get("question") and m.get("answer"): # Last message in this chat was an admin reply — skip tracking break if last_user_msg: try: u_time = datetime.fromisoformat(last_user_msg["timestamp"].replace("Z", "+00:00")) a_time = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) delta_sec = (a_time - u_time).total_seconds() if delta_sec > 0: data["sla_metrics"]["agent_reply_times"].append(delta_sec) except Exception as e: logger.error("Error calculating agent response time: %s", e) # ----------------------------------- data["messages"].append({ "id": data["message_count"] + 1, "chat_id": chat_id, "timestamp": timestamp, "question": question, "answer": answer, }) data["message_count"] += 1 data["last_active"] = timestamp await asyncio.to_thread(_write_json, path, data) logger.info("session_store: successfully saved message to %s", path) except Exception as exc: logger.error( "session_store: failed to save message for session %s: %s", session_id, exc, ) async def get_all_sessions(search: str = "", filter_type: str = "", skip: int = 0, limit: int = 50) -> list: """ Returns a list of all session summaries (metadata only, no full messages). Used by the Admin Panel sidebar. Supports pagination, searching, and filtering. """ try: files = await asyncio.to_thread(_list_json_files) sessions = [] now_utc = datetime.now(timezone.utc) today_date = now_utc.date() search_lower = search.lower() if search else "" for fpath in files: try: data = await asyncio.to_thread(_read_json, fpath) s_id = data.get("session_id", fpath.stem) u_name = data.get("user_name", "") if search_lower: if search_lower not in str(s_id).lower() and search_lower not in str(u_name).lower(): continue status = data.get("status", "AI") blocked = data.get("blocked", False) archived = data.get("archived", False) archived_chats = data.get("archived_chats") or [] last_active = data.get("last_active", "") if filter_type == "archived": if not (archived or len(archived_chats) > 0): continue elif filter_type == "blocked": if not blocked: continue elif filter_type == "queued": if status != "QUEUED" or archived or blocked: continue elif filter_type == "daily": if blocked or archived: continue created_at = data.get("created_at", "") if not created_at: continue try: dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) if dt.astimezone(timezone.utc).date() != today_date: continue except Exception: continue elif filter_type == "sessions" or not filter_type: if blocked or archived: continue sessions.append({ "session_id": s_id, "user_name": u_name, "created_at": data.get("created_at", ""), "last_active": last_active, "message_count": data.get("message_count", 0), "current_chat_id": data.get("current_chat_id", 1), "status": status, "blocked": blocked, "rating": data.get("rating", None), "chat_ratings": data.get("chat_ratings", {}), "alert_id": data.get("alert_id", 0), "alert_type": data.get("alert_type", ""), "archived": archived, "archived_chats": archived_chats, }) except Exception as exc: logger.warning("session_store: could not read %s: %s", fpath, exc) # Sort newest first sessions.sort(key=lambda s: s.get("last_active", ""), reverse=True) return sessions[skip:skip+limit] except Exception as exc: logger.error("session_store: get_all_sessions failed: %s", exc) return [] async def get_metrics() -> dict: """Calculates global metrics efficiently for the dashboard.""" try: files = await asyncio.to_thread(_list_json_files) live = 0 queued = 0 daily = 0 blocked_count = 0 now_utc = datetime.now(timezone.utc) today_date = now_utc.date() for fpath in files: try: data = await asyncio.to_thread(_read_json, fpath) blocked = data.get("blocked", False) archived = data.get("archived", False) status = data.get("status", "AI") last_active = data.get("last_active", "") if blocked: blocked_count += 1 if not blocked and not archived: if status == "QUEUED": queued += 1 elif status != "ENDED": live += 1 created_at = data.get("created_at", "") if created_at: try: dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) if dt.astimezone(timezone.utc).date() == today_date: daily += 1 except Exception: pass except Exception: pass return { "live": live, "queued": queued, "daily": daily, "blocked": blocked_count } except Exception as exc: logger.error("session_store: get_metrics failed: %s", exc) return {"live": 0, "queued": 0, "daily": 0, "blocked": 0} async def get_session(session_id: str) -> dict: """ Returns the full session data (including all messages) for the Admin Panel. Returns an empty dict if the session does not exist. """ path = _get_session_path(session_id) try: if not path.exists(): return {} return await asyncio.to_thread(_read_json, path) except Exception as exc: logger.error("session_store: get_session(%s) failed: %s", session_id, exc) return {} async def delete_session(session_id: str) -> bool: """ Deletes a session file. Returns True on success, False on failure. Used by Admin Panel if needed in the future. """ path = _get_session_path(session_id) lock = await _get_lock(session_id) async with lock: try: if path.exists(): await asyncio.to_thread(os.remove, path) return True except Exception as exc: logger.error("session_store: delete_session(%s) failed: %s", session_id, exc) return False async def update_session_name(session_id: str, user_name: str) -> None: """Updates the user_name field in the session JSON file.""" path = _get_session_path(session_id) lock = await _get_lock(session_id) async with lock: try: if path.exists(): data = await asyncio.to_thread(_read_json, path) data["user_name"] = user_name await asyncio.to_thread(_write_json, path, data) logger.info("session_store: updated name for session %s to %s", session_id, user_name) else: # If session doesn't exist yet, we can't update it, but we could create it? # Usually ensure_session is called first. logger.warning("session_store: attempt to update name for non-existent session %s", session_id) except Exception as exc: logger.error("session_store: failed to update name for session %s: %s", session_id, exc) async def rename_session(old_id: str, new_name: str) -> str: """ Renames/merges a temp session into the canonical named session for new_name. Strategy: - If a session file for `new_name` already exists (same user logging in from a different browser or device), we MERGE the temp session's messages into that existing session and discard the temp file. - If no named session exists yet, we simply rename the temp file. - If the session is already correctly named, we just update user_name in-place. Always returns the canonical session_id that the caller should use. """ import re # 1. Sanitize the name for use as a filename clean_name = re.sub(r'[^\w\s-]', '', new_name).strip().replace(' ', '_') if not clean_name: clean_name = "User" new_id = clean_name old_path = _get_session_path(old_id) new_path = _get_session_path(new_id) # ── Case A: session file is already correctly named ────────────────────── if new_id == old_id: update_lock = await _get_lock(old_id) async with update_lock: try: if old_path.exists(): data = await asyncio.to_thread(_read_json, old_path) if data.get("user_name") != new_name: data["user_name"] = new_name await asyncio.to_thread(_write_json, old_path, data) logger.info( "session_store: updated user_name in-place for %s → '%s'", old_id, new_name ) except Exception as exc: logger.error( "session_store: failed to update user_name for %s: %s", old_id, exc ) return old_id # ── Case B: canonical named session already exists — MERGE ─────────────── if new_path.exists(): lock_old = await _get_lock(old_id) lock_new = await _get_lock(new_id) async with lock_old: async with lock_new: try: existing_data = await asyncio.to_thread(_read_json, new_path) # Merge messages from the temp session (if any) into the # existing canonical session so no history is lost. if old_path.exists(): temp_data = await asyncio.to_thread(_read_json, old_path) new_messages = temp_data.get("messages", []) if new_messages: existing_data["messages"].extend(new_messages) existing_data["message_count"] = len(existing_data["messages"]) existing_data["last_active"] = _now_iso() await asyncio.to_thread(_write_json, new_path, existing_data) logger.info( "session_store: merged %d message(s) from temp session '%s' into '%s'", len(new_messages), old_id, new_id ) # Remove the now-redundant temp file await asyncio.to_thread(os.remove, old_path) logger.info( "session_store: removed temp session file '%s' after merge", old_id ) else: logger.info( "session_store: temp session '%s' had no file; reusing '%s'", old_id, new_id ) except Exception as exc: logger.error( "session_store: failed to merge session %s -> %s: %s", old_id, new_id, exc ) return new_id # ── Case C: no existing named session — simple rename ──────────────────── lock_old = await _get_lock(old_id) lock_new = await _get_lock(new_id) async with lock_old: async with lock_new: try: if old_path.exists(): data = await asyncio.to_thread(_read_json, old_path) data["session_id"] = new_id data["user_name"] = new_name await asyncio.to_thread(_write_json, new_path, data) await asyncio.to_thread(os.remove, old_path) logger.info("session_store: renamed session %s -> %s", old_id, new_id) return new_id else: logger.info( "session_store: old session %s did not exist; creating new named session %s", old_id, new_id ) skeleton = { "session_id": new_id, "user_name": new_name, "created_at": _now_iso(), "last_active": _now_iso(), "message_count": 0, "messages": [], "status": "AI", "blocked": False, "rating": None, "chat_ratings": {}, "alert_id": 0, "alert_type": "", "archived": False, "archived_chats": [], "sla_metrics": {}, } await asyncio.to_thread(_write_json, new_path, skeleton) return new_id except Exception as exc: logger.error("session_store: failed to rename session %s: %s", old_id, exc) return old_id # ── Sync helper functions (run via asyncio.to_thread) ──────────────────────── def _write_json(path: Path, data: dict) -> None: """Atomically writes JSON to a file using a temp file + rename pattern.""" tmp_path = path.with_suffix(".tmp") with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp_path, path) # Atomic on POSIX, best-effort on Windows def _read_json(path: Path) -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) def _list_json_files() -> list: return list(_get_sessions_dir().glob("*.json")) async def update_session_field(session_id: str, field: str, value) -> bool: """Generic helper to update a single root field in the session JSON.""" path = _get_session_path(session_id) lock = await _get_lock(session_id) async with lock: try: if not path.exists(): return False data = await asyncio.to_thread(_read_json, path) if field == "status": if "sla_metrics" not in data: data["sla_metrics"] = {} # If changing to QUEUED, record queued_at if value == "QUEUED" and data.get("status") != "QUEUED": data["sla_metrics"]["queued_at"] = _now_iso() # Clear old takeover/ended to prep for new escalation data["sla_metrics"].pop("takeover_at", None) data["sla_metrics"].pop("ended_at", None) # If changing to HUMAN_TAKEOVER, record takeover_at and calculate wait time elif value == "HUMAN_TAKEOVER" and data.get("status") != "HUMAN_TAKEOVER": takeover_time = _now_iso() data["sla_metrics"]["takeover_at"] = takeover_time queued_at = data["sla_metrics"].get("queued_at") if queued_at: try: q_time = datetime.fromisoformat(queued_at.replace("Z", "+00:00")) t_time = datetime.fromisoformat(takeover_time.replace("Z", "+00:00")) wait_sec = (t_time - q_time).total_seconds() if wait_sec >= 0: data["sla_metrics"].setdefault("queue_wait_times", []).append(wait_sec) except Exception: pass # If leaving human-support phase, record ended_at and calculate resolution time elif data.get("status") in ["HUMAN_TAKEOVER", "QUEUED"] and value not in ["HUMAN_TAKEOVER", "QUEUED"]: ended_time = _now_iso() data["sla_metrics"]["ended_at"] = ended_time start_str = data["sla_metrics"].get("takeover_at") or data["sla_metrics"].get("queued_at") if start_str: try: s_time = datetime.fromisoformat(start_str.replace("Z", "+00:00")) e_time = datetime.fromisoformat(ended_time.replace("Z", "+00:00")) delta = (e_time - s_time).total_seconds() if delta >= 0: data["sla_metrics"].setdefault("resolution_times", []).append(delta) except Exception: pass # Catch-all for explicitly ended (if not handled above) elif value == "ENDED" and data.get("status") != "ENDED": data["sla_metrics"]["ended_at"] = _now_iso() data[field] = value await asyncio.to_thread(_write_json, path, data) logger.info("session_store: updated %s for session %s", field, session_id) return True except Exception as exc: logger.error("session_store: failed to update %s for session %s: %s", field, session_id, exc) return False async def archive_chat_thread(session_id: str, chat_id: int, archive: bool) -> bool: """Adds or removes a chat_id from the archived_chats list.""" path = _get_session_path(session_id) lock = await _get_lock(session_id) async with lock: try: if not path.exists(): return False data = await asyncio.to_thread(_read_json, path) archived_chats = data.get("archived_chats", []) if archive: if chat_id not in archived_chats: archived_chats.append(chat_id) else: if chat_id in archived_chats: archived_chats.remove(chat_id) data["archived_chats"] = archived_chats await asyncio.to_thread(_write_json, path, data) logger.info("session_store: updated archived_chats for session %s: %s", session_id, archived_chats) return True except Exception as exc: logger.error("session_store: failed to archive chat thread %s for session %s: %s", chat_id, session_id, exc) return False async def get_sla_metrics() -> dict: """Calculates SLA metrics across all sessions.""" try: files = await asyncio.to_thread(_list_json_files) total_queue_time = 0 queued_count = 0 total_resolution_time = 0 resolved_count = 0 total_agent_reply_time = 0 agent_reply_count = 0 takeover_count = 0 total_ai_sessions = 0 breach_count = 0 SLA_THRESHOLD_SEC = 3 * 60 # 3 minutes for fpath in files: try: data = await asyncio.to_thread(_read_json, fpath) if data.get("message_count", 0) > 0: total_ai_sessions += 1 sla = data.get("sla_metrics", {}) if "takeover_at" in sla or data.get("status") == "HUMAN_TAKEOVER": takeover_count += 1 # Read from arrays first (new logic) for w in sla.get("queue_wait_times", []): total_queue_time += w queued_count += 1 if w > SLA_THRESHOLD_SEC: breach_count += 1 for r in sla.get("resolution_times", []): total_resolution_time += r resolved_count += 1 # Fallback for legacy data (before arrays were added) if not sla.get("queue_wait_times") and not sla.get("resolution_times"): queued_at_str = sla.get("queued_at") takeover_at_str = sla.get("takeover_at") ended_at_str = sla.get("ended_at") if queued_at_str and takeover_at_str: try: q_time = datetime.fromisoformat(queued_at_str.replace("Z", "+00:00")) t_time = datetime.fromisoformat(takeover_at_str.replace("Z", "+00:00")) wait_sec = (t_time - q_time).total_seconds() if wait_sec >= 0: total_queue_time += wait_sec queued_count += 1 if wait_sec > SLA_THRESHOLD_SEC: breach_count += 1 except Exception: pass start_str = takeover_at_str or queued_at_str if ended_at_str and start_str: try: s_time = datetime.fromisoformat(start_str.replace("Z", "+00:00")) e_time = datetime.fromisoformat(ended_at_str.replace("Z", "+00:00")) delta = (e_time - s_time).total_seconds() if delta >= 0: total_resolution_time += delta resolved_count += 1 except Exception: pass reply_times = sla.get("agent_reply_times", []) for rt in reply_times: total_agent_reply_time += rt agent_reply_count += 1 except Exception: pass avg_queue_time = (total_queue_time / queued_count) if queued_count > 0 else 0 avg_resolution_time = (total_resolution_time / resolved_count) if resolved_count > 0 else 0 avg_agent_reply_time = (total_agent_reply_time / agent_reply_count) if agent_reply_count > 0 else 0 takeover_rate = (takeover_count / total_ai_sessions * 100) if total_ai_sessions > 0 else 0 breach_rate = (breach_count / queued_count * 100) if queued_count > 0 else 0 return { "avg_queue_time_sec": avg_queue_time, "avg_resolution_time_sec": avg_resolution_time, "avg_agent_reply_time_sec": avg_agent_reply_time, "takeover_rate_pct": takeover_rate, "breach_count": breach_count, "breach_rate_pct": breach_rate, "total_ai_sessions": total_ai_sessions, "queued_count": queued_count } except Exception as exc: logger.error("session_store: get_sla_metrics failed: %s", exc) return { "avg_queue_time_sec": 0, "avg_resolution_time_sec": 0, "avg_agent_reply_time_sec": 0, "takeover_rate_pct": 0, "breach_count": 0, "breach_rate_pct": 0, "total_ai_sessions": 0, "queued_count": 0 }