""" Nancy HF Space — Admin Dashboard. Exposes a beautiful, premium dark-mode, glassmorphic monitoring dashboard at /admin for visualizing system health, queues, circuit breakers, and active sessions. """ from __future__ import annotations import time import uuid import hashlib from fastapi import APIRouter, Request, Body from fastapi.responses import HTMLResponse from core.queue import task_queue from core.router import provider_router from core.sessions import session_store from core.redis_client import redis_client router = APIRouter(prefix="/admin", tags=["Nancy Administration"]) # HTML template string containing premium CSS styled with glassmorphic cards and gradients DASHBOARD_HTML = """ Nancy v2 — Control Center

NANCY CONTROL

Advanced Free Chatbot Orchestrator & API Router

SYS STATE: OPERATIONAL
Task Queue Depth SSE Queue
{queue_depth}
Active tasks waiting for pickup
Active Sessions Redis
{session_count}
Tracked multi-conversation sessions
Extension Hub Relay Status
{ext_status}
Clients connected in real-time
Upstash Persistence Cache
{redis_status}
State persistence engine status
Provider Status & Failover Configuration
{provider_rows}
Active Sessions {session_count} Total
{session_rows}
🔑 Dynamic API Key Handoff Vault Redis Hashed Credentials

How to connect Ultron Swarm (or external agents)

Generate a secure API key below. Paste it along with your Nancy server URL (e.g. https://ghostdriveg1-free-llm-router.hf.space) into the Ultron Swarm Control Dashboard. Nancy hashes and saves all keys securely using SHA-256 in Upstash Redis.

Client Description Key Hash (SHA-256) Created At Last Active Action
Loading API keys vault...
""" @router.get("/", response_class=HTMLResponse) async def admin_dashboard(request: Request): """Renders the control center admin status page.""" # 1. Fetch current queue size try: queue_depth = task_queue.pending_count except Exception: queue_depth = 0 # 2. Check Extension SSE connections status ext_connected = task_queue.is_extension_active() ext_status = "Connected" if ext_connected else "Offline" # 3. Check Redis Connection Status redis_active = redis_client.is_enabled redis_status = "ONLINE" if redis_active else "FALLBACK (IN-MEMORY)" redis_color = "var(--success)" if redis_active else "var(--warning)" # 4. Fetch Sessions list try: sessions = await session_store.list_sessions() session_count = len(sessions) except Exception: sessions = [] session_count = 0 # 5. Build dynamic session items session_rows = "" if not sessions: session_rows = '
No active tracked sessions found.
' else: for sess in sessions[:8]: # Show top 8 active session_rows += f"""

{sess.title}

{sess.conversation_url or 'Fresh Chat (No URL yet)'}

{sess.provider}
""" # 6. Fetch Providers circuit breaker states # Default list of providers all_providers = ["chatgpt", "gemini", "deepseek", "kimi", "claude", "nim", "zai"] provider_rows = "" for provider in all_providers: # Determine status is_healthy = provider_router.is_provider_healthy(provider) badge_class = "status-healthy" if is_healthy else "status-broken" status_text = "HEALTHY" if is_healthy else "DEGRADED / DRAINED" # Determine adapter fallback priority indicator fallback_pos = "Primary" if provider in provider_router.fallback_chain else "Bypass / API" if provider in provider_router.fallback_chain: idx = provider_router.fallback_chain.index(provider) + 1 fallback_pos = f"Fallback Chain #{idx}" provider_rows += f"""
{provider[:2]}

{provider}

Priority: {fallback_pos}
{status_text}
""" # Render dashboard safely using simple replacement to avoid CSS/JS brace clashes rendered = ( DASHBOARD_HTML .replace("{queue_depth}", str(queue_depth)) .replace("{session_count}", str(session_count)) .replace("{ext_status}", str(ext_status)) .replace("{redis_status}", str(redis_status)) .replace("{redis_color}", str(redis_color)) .replace("{session_rows}", str(session_rows)) .replace("{provider_rows}", str(provider_rows)) ) return HTMLResponse(content=rendered) # ── Dynamic API Key AJAX Endpoints ─────────────────────────────────────────── @router.post("/keys/create") async def create_api_key(payload: dict = Body(default={})): """Generates a secure UUID API key prefixed with ny_, hashes it, and caches in Redis.""" description = payload.get("description", "Swarm Integration Client") plaintext_uuid = uuid.uuid4().hex plaintext_key = f"ny_{plaintext_uuid}" hashed = hashlib.sha256(plaintext_key.encode("utf-8")).hexdigest() key_id = str(uuid.uuid4()) metadata = { "id": key_id, "description": description, "hash": hashed, "created_at": int(time.time()), "last_used": None, "request_count": 0 } # Save key metadata and register in active hashes set await redis_client.set_json(f"nancy:api_keys:{hashed}", metadata) await redis_client._execute("SADD", "nancy:active_key_hashes", hashed) return { "key_id": key_id, "plaintext_key": plaintext_key, "description": description, "created_at": metadata["created_at"] } @router.get("/keys/list") async def list_api_keys(): """Lists metadata for all active dynamic API keys.""" try: hashes = await redis_client._execute("SMEMBERS", "nancy:active_key_hashes") or [] keys_list = [] for h in hashes: meta = await redis_client.get_json(f"nancy:api_keys:{h}") if meta: keys_list.append(meta) return sorted(keys_list, key=lambda x: x.get("created_at", 0), reverse=True) except Exception: return [] @router.delete("/keys/revoke/{hashed_key}") async def revoke_api_key(hashed_key: str): """Revokes and deletes an API key using its SHA-256 hash.""" await redis_client.delete(f"nancy:api_keys:{hashed_key}") await redis_client._execute("SREM", "nancy:active_key_hashes", hashed_key) return {"success": True, "message": "API key revoked successfully."}