""" app/services/escalation_engine.py ───────────────────────────────── Claim-based session locking with round-robin escalation and admin-to-admin transfer. Design: • When a session enters QUEUED, assign to the first eligible admin. • If they don't take over within the timeout (default 20s), rotate to next. • After 2 full cycles through all eligible admins, exit queue gracefully. • Admins handling sessions can transfer them to another specific admin. • All WhatsApp notifications use the existing ProviderFactory (Meta + Baileys). """ import asyncio import json import logging from datetime import datetime, timezone from pathlib import Path from typing import List, Optional, Dict from app.core.config import get_settings logger = logging.getLogger(__name__) settings = get_settings() # ── In-memory task registry ─────────────────────────────────────────────────── _escalation_tasks: Dict[str, asyncio.Task] = {} # ── Default timeout ─────────────────────────────────────────────────────────── _DEFAULT_TIMEOUT = 20 # seconds _MIN_TIMEOUT = 5 _MAX_TIMEOUT = 120 # ── App Settings (escalation timeout) ───────────────────────────────────────── def _get_app_settings_path() -> Path: """Returns path to app_settings.json (respects HF Spaces /data volume).""" base = Path("/data") if Path("/data").exists() else Path("data") base.mkdir(parents=True, exist_ok=True) return base / "app_settings.json" def get_escalation_timeout() -> int: """Read escalation timeout from app_settings.json. Returns default if missing.""" path = _get_app_settings_path() try: if path.exists(): with open(path, "r", encoding="utf-8") as f: data = json.load(f) return int(data.get("escalation_timeout_seconds", _DEFAULT_TIMEOUT)) except Exception as e: logger.warning("escalation_engine: failed to read app_settings: %s", e) return _DEFAULT_TIMEOUT def set_escalation_timeout(seconds: int) -> None: """Write escalation timeout to app_settings.json.""" if seconds < _MIN_TIMEOUT or seconds > _MAX_TIMEOUT: raise ValueError(f"Timeout must be between {_MIN_TIMEOUT} and {_MAX_TIMEOUT} seconds.") path = _get_app_settings_path() try: data = {} if path.exists(): with open(path, "r", encoding="utf-8") as f: data = json.load(f) data["escalation_timeout_seconds"] = seconds tmp = path.with_suffix(".tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) import os os.replace(str(tmp), str(path)) logger.info("escalation_engine: timeout set to %ds", seconds) except Exception as e: logger.error("escalation_engine: failed to save app_settings: %s", e) raise # ── Eligible Admins ─────────────────────────────────────────────────────────── async def get_eligible_admins() -> List[dict]: """ Returns sorted list of admins eligible for escalation rotation. Criteria: active, not suspended, has sessions.takeover permission, on_duty=True. Superadmins are placed last (final fallback). """ from app.services.admin_store import get_all_admins all_admins = await get_all_admins() eligible = [] for admin in all_admins: if not admin.get("is_active", True): continue if not admin.get("on_duty", True): continue # Check permissions perms = admin.get("permissions", []) role = admin.get("role", "sub_admin") if role == "superadmin" or "sessions.takeover" in perms or "manage_sessions" in perms: eligible.append(admin) # Sort: sub_admins first (alphabetically), superadmins last eligible.sort(key=lambda a: (0 if a.get("role") != "superadmin" else 1, a.get("username", ""))) return eligible async def get_escalation_status() -> dict: """Returns live escalation status for the admin settings panel.""" from app.services.session_store import get_all_sessions eligible = await get_eligible_admins() queued_sessions = await get_all_sessions(filter_type="queued", limit=100) active_assignments = [] for sess in queued_sessions: if sess.get("status") == "QUEUED" and sess.get("assigned_to"): active_assignments.append({ "session_id": sess["session_id"], "user_name": sess.get("user_name", "Guest"), "assigned_to": sess.get("assigned_to", ""), "assigned_at": sess.get("assigned_at", ""), }) return { "timeout_seconds": get_escalation_timeout(), "max_cycles": 2, "eligible_admins": [ { "username": a.get("username", ""), "display_name": a.get("display_name", "") or a.get("username", ""), "role": a.get("role", "sub_admin"), "on_duty": a.get("on_duty", True), "has_whatsapp": bool(a.get("whatsapp_number", "")), } for a in eligible ], "active_escalations": len(_escalation_tasks), "queued_sessions": len(queued_sessions), "assignments": active_assignments, } # ── Start Escalation ───────────────────────────────────────────────────────── async def start_escalation(session_id: str, resume: bool = False) -> None: """ Start the escalation rotation for a QUEUED session. Called when session status changes to QUEUED (or on startup rehydration). """ from app.services.session_store import get_session, assign_session # Cancel any existing escalation for this session cancel_escalation(session_id) sess = await get_session(session_id) if not sess or sess.get("status") != "QUEUED": return admins = await get_eligible_admins() if not admins: # No eligible admins — exit queue immediately await _exit_queue_no_agents(session_id) return if resume and sess.get("assigned_to"): # Resuming from a server restart — check if assignment is stale assigned_at = sess.get("assigned_at", "") if assigned_at: try: assigned_time = datetime.fromisoformat(assigned_at.replace("Z", "+00:00")) now = datetime.now(timezone.utc) elapsed = (now - assigned_time).total_seconds() timeout = get_escalation_timeout() if elapsed >= timeout: # Stale assignment — advance to next admin idx = sess.get("escalation_index", 0) cycle = sess.get("escalation_cycle", 0) next_idx = (idx + 1) % len(admins) if next_idx == 0: cycle += 1 if cycle >= 2: await _exit_queue_no_agents(session_id) return next_admin = admins[next_idx] await assign_session(session_id, next_admin["username"], next_idx, cycle) await _notify_admin_whatsapp(next_admin, session_id, "assignment") task = asyncio.create_task(_escalation_tick(session_id)) _escalation_tasks[session_id] = task return except Exception as e: logger.error("escalation_engine: rehydration error for %s: %s", session_id, e) # Assignment is still fresh — just restart the timer for remaining time task = asyncio.create_task(_escalation_tick(session_id)) _escalation_tasks[session_id] = task return # Fresh start — assign to the first eligible admin first_admin = admins[0] await assign_session(session_id, first_admin["username"], 0, 0) await _notify_admin_whatsapp(first_admin, session_id, "assignment") if len(admins) > 1: # Start rotation timer task = asyncio.create_task(_escalation_tick(session_id)) _escalation_tasks[session_id] = task # If only 1 admin, no rotation needed — they stay assigned indefinitely # ── Escalation Tick ─────────────────────────────────────────────────────────── async def _escalation_tick(session_id: str) -> None: """ Background task: sleep for timeout, then check if admin took over. If not, rotate to the next admin. After 2 full cycles, exit queue. """ from app.services.session_store import get_session, assign_session timeout = get_escalation_timeout() await asyncio.sleep(timeout) sess = await get_session(session_id) if not sess or sess.get("status") != "QUEUED": # Already taken over, ended, or session gone _escalation_tasks.pop(session_id, None) return admins = await get_eligible_admins() if not admins: await _exit_queue_no_agents(session_id) _escalation_tasks.pop(session_id, None) return current_index = sess.get("escalation_index", 0) current_cycle = sess.get("escalation_cycle", 0) next_index = (current_index + 1) % len(admins) # Check if we completed a full cycle if next_index == 0: current_cycle += 1 # After 2 full cycles, exit queue if current_cycle >= 2: await _exit_queue_no_agents(session_id) _escalation_tasks.pop(session_id, None) return # Assign to next admin next_admin = admins[next_index] await assign_session(session_id, next_admin["username"], next_index, current_cycle) await _notify_admin_whatsapp(next_admin, session_id, "assignment") logger.info("escalation_engine: rotated session %s to %s (idx=%d, cycle=%d)", session_id, next_admin["username"], next_index, current_cycle) # Schedule next tick task = asyncio.create_task(_escalation_tick(session_id)) _escalation_tasks[session_id] = task # ── Cancel Escalation ───────────────────────────────────────────────────────── def cancel_escalation(session_id: str) -> None: """Cancel any pending escalation timer for a session.""" task = _escalation_tasks.pop(session_id, None) if task and not task.done(): task.cancel() logger.info("escalation_engine: cancelled escalation for session %s", session_id) # ── Exit Queue (no agents available) ───────────────────────────────────────── async def _exit_queue_no_agents(session_id: str) -> None: """All admins tried twice. Return session to AI with a formal message.""" from app.services.session_store import update_session_field, save_message, clear_assignment logger.info("escalation_engine: exiting queue for session %s (no agents responded)", session_id) await update_session_field(session_id, "status", "AI") await clear_assignment(session_id) # Send formal bot message to the user await save_message( session_id, "", "We apologize for the inconvenience. No human agent is available at the moment. " "Please try again later or continue chatting with me — I'll do my best to assist you.", ) # Notify all admins about the unattended request await _notify_all_admins_unattended(session_id) async def _notify_all_admins_unattended(session_id: str) -> None: """Send WhatsApp notification to all eligible admins about an unattended session.""" from app.services.session_store import get_session sess = await get_session(session_id) user_name = sess.get("user_name", "a user") if sess else "a user" admins = await get_eligible_admins() for admin in admins: number = admin.get("whatsapp_number", "") if not number: continue msg = ( f"⚠️ *Unattended Support Request*\n\n" f"A support session from *{user_name}* could not be attended.\n" f"All available agents were notified but none responded within the allowed time.\n\n" f"Session: `{session_id}`" ) await _send_whatsapp_safe(number, msg) # ── Admin-to-Admin Transfer ────────────────────────────────────────────────── async def transfer_session( session_id: str, from_username: str, target_username: str, ) -> dict: """ Transfer a session from the current handler to another admin. Returns {"status": "transferred", "display_name": ..., "whatsapp_notified": bool}. Raises ValueError on rule violations. """ from app.services.session_store import get_session, _get_session_path, _read_json, _write_json, _get_lock from app.services.admin_store import get_admin, get_all_admins, _load_admins, _append_audit, _save_admins, _lock as admin_lock sess = await get_session(session_id) if not sess: raise ValueError("Session not found.") if sess.get("status") == "ENDED": raise ValueError("Session has already ended.") if sess.get("status") == "QUEUED": raise ValueError("Cannot transfer a queued session. Take over first.") if sess.get("status") != "HUMAN_TAKEOVER": raise ValueError("Session must be in human takeover to transfer.") # Verify ownership (assigned_to or takeover_by must match, or sender is superadmin) from_admin = await get_admin(from_username) is_superadmin = from_admin and from_admin.get("role") == "superadmin" current_handler = sess.get("assigned_to") or sess.get("takeover_by", "") if not is_superadmin and from_username.lower() != current_handler.lower(): raise ValueError("You can only transfer sessions you are currently handling.") # Validate target if target_username.lower() == from_username.lower(): raise ValueError("Cannot transfer to yourself.") target_admin = await get_admin(target_username) if not target_admin: raise ValueError(f"Admin '{target_username}' not found.") if not target_admin.get("is_active", True): raise ValueError(f"Admin '{target_username}' is suspended.") # Check target has takeover permission target_perms = target_admin.get("permissions", []) target_role = target_admin.get("role", "sub_admin") if target_role != "superadmin" and "sessions.takeover" not in target_perms and "manage_sessions" not in target_perms: raise ValueError(f"Admin '{target_username}' does not have takeover permission.") # Cancel any existing escalation cancel_escalation(session_id) # Update session target_display = target_admin.get("display_name") or target_username path = _get_session_path(session_id) lock = await _get_lock(session_id) async with lock: try: data = await asyncio.to_thread(_read_json, path) data["assigned_to"] = target_username data["takeover_by"] = target_display data["assigned_at"] = datetime.now(timezone.utc).isoformat() await asyncio.to_thread(_write_json, path, data) except Exception as e: raise ValueError(f"Failed to update session: {e}") # Audit log async with admin_lock: admin_data = await _load_admins() from_display = from_admin.get("display_name", from_username) if from_admin else from_username _append_audit( admin_data, from_username, "transfer_session", session_id, f"{from_display} transferred session to {target_display}" ) await _save_admins(admin_data) # Send WhatsApp notification to target admin user_name = sess.get("user_name", "a user") from_display = from_admin.get("display_name", from_username) if from_admin else from_username wa_notified = False target_number = target_admin.get("whatsapp_number", "") if target_number: msg = ( f"📋 *Chat Transfer Request*\n\n" f"Hello {target_display},\n\n" f"{from_display} has transferred a support session to you.\n\n" f"User: *{user_name}*\n" f"Session: `{session_id}`\n\n" f"Please open the admin panel to continue assisting this user." ) wa_notified = await _send_whatsapp_safe(target_number, msg) logger.info("escalation_engine: %s transferred session %s to %s (wa=%s)", from_username, session_id, target_username, wa_notified) return { "status": "transferred", "transferred_to": target_username, "display_name": target_display, "whatsapp_notified": wa_notified, } # ── WhatsApp Notification Helpers ───────────────────────────────────────────── async def _notify_admin_whatsapp(admin: dict, session_id: str, msg_type: str = "assignment") -> None: """Send a formal WhatsApp notification to an admin about a session assignment.""" number = admin.get("whatsapp_number", "") if not number: return from app.services.session_store import get_session sess = await get_session(session_id) user_name = sess.get("user_name", "a user") if sess else "a user" display_name = admin.get("display_name") or admin.get("username", "Admin") timeout = get_escalation_timeout() if msg_type == "assignment": msg = ( f"📋 *New Support Request*\n\n" f"Hello {display_name},\n\n" f"A support session from *{user_name}* has been assigned to you.\n" f"Please open the admin panel and respond within {timeout} seconds to accept.\n\n" f"Session: `{session_id}`" ) else: msg = ( f"📋 *Support Notification*\n\n" f"Hello {display_name},\n\n" f"You have a pending support session from *{user_name}*.\n\n" f"Session: `{session_id}`" ) await _send_whatsapp_safe(number, msg) async def _send_whatsapp_safe(number: str, message: str) -> bool: """ Send a WhatsApp message using the existing ProviderFactory. Returns True on success, False on failure. Never raises. """ try: from app.admin.router import get_whatsapp_settings ws = get_whatsapp_settings() if not ws.get("phone_id") and not ws.get("baileys_url"): logger.debug("escalation_engine: no WhatsApp provider configured, skipping notification") return False from app.services.whatsapp import ProviderFactory provider = ProviderFactory.get_provider(ws) # Clean number — strip + if stored with it (legacy), ensure digits only clean_num = number.strip().lstrip("+") if not clean_num: logger.warning("escalation_engine: empty WhatsApp number, skipping") return False errors = await asyncio.to_thread(provider.send_message, [clean_num], message) if errors: for err in errors: logger.warning("escalation_engine: WhatsApp send error for %s: %s", number, err) return False logger.info("escalation_engine: WhatsApp sent to %s", number) return True except Exception as e: logger.warning("escalation_engine: WhatsApp notification failed for %s: %s", number, e) return False # ── Startup Rehydration ─────────────────────────────────────────────────────── async def rehydrate_escalations() -> None: """ Resume escalation for any sessions stuck in QUEUED with stale assignments. Called on server startup. """ from app.services.session_store import get_all_sessions try: sessions = await get_all_sessions(filter_type="queued", limit=100) rehydrated = 0 for sess in sessions: if sess.get("status") == "QUEUED": await start_escalation(sess["session_id"], resume=True) rehydrated += 1 if rehydrated: logger.info("escalation_engine: rehydrated %d escalation(s)", rehydrated) except Exception as e: logger.error("escalation_engine: rehydration failed: %s", e)