"""Unified HTML admin dashboard.
Renders a single self-contained page that aggregates the read-only views
over everything the admin endpoints expose:
* cache stats (v1.31)
* rate-limiter stats (v1.31)
* observability: counters, latency percentiles, recent request log (v1.33)
* audit trail (v1.33)
* API-key inventory (hash prefixes only, no plaintext) (v1.32)
* document lifecycle — total count + first N IDs (v1.38)
* readiness (v1.34)
The endpoint is admin-scoped (inherits from the `/v1/admin/*` middleware
prefix). Auto-refresh interval configurable via the ?refresh=N query param.
No writes — a read-only cockpit. Any state-changing action is left to the
dedicated POST/DELETE endpoints.
"""
from __future__ import annotations
import html as _h
import json
import time
from typing import Any, Dict, List
def _esc(s: object) -> str:
return _h.escape(str(s), quote=True)
def _kv_table(rows: List[tuple]) -> str:
cells = "".join(
f'| {_esc(k)} | '
f'{_esc(v)} |
'
for k, v in rows
)
return (f'')
def _card(title: str, body_html: str) -> str:
return (
f''
f'
{_esc(title)}
{body_html}'
)
def render_admin_ui(
*,
cache_stats: Dict[str, Any],
limiter_stats: Dict[str, Any],
obs_stats: Dict[str, Any],
recent_requests: List[Dict[str, Any]],
recent_audits: List[Dict[str, Any]],
keys: List[Dict[str, Any]],
documents: List[Dict[str, Any]],
readiness: Dict[str, Any],
refresh_sec: int = 0,
) -> str:
"""Assemble the dashboard HTML from pre-computed stat dicts."""
ts_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
# ---------- summary cards
summary_cards = []
# Cache
ch = cache_stats.get("hits", 0)
cm = cache_stats.get("misses", 0)
summary_cards.append(_card("query cache", _kv_table([
("size", cache_stats.get("size", 0)),
("hits", ch),
("misses", cm),
("hit_rate", f'{cache_stats.get("hit_rate", 0.0):.3f}'),
("ttl_sec", cache_stats.get("ttl_sec", "—")),
])))
# Rate limiter
summary_cards.append(_card("rate limiter", _kv_table([
("rate/sec", limiter_stats.get("rate_per_sec", "—")),
("capacity", limiter_stats.get("capacity", "—")),
("buckets", limiter_stats.get("buckets", 0)),
("allowed", limiter_stats.get("allowed", 0)),
("denied", limiter_stats.get("denied", 0)),
("pass_rate", f'{limiter_stats.get("pass_rate", 0.0):.3f}'),
])))
# Observability
summary_cards.append(_card("observability", _kv_table([
("total", obs_stats.get("total", 0)),
("by_status", obs_stats.get("by_status", {})),
("avg_ms", f'{obs_stats.get("avg_latency_ms", 0.0):.2f}'),
("p50_ms", f'{obs_stats.get("p50_ms", 0.0):.2f}'),
("p95_ms", f'{obs_stats.get("p95_ms", 0.0):.2f}'),
("p99_ms", f'{obs_stats.get("p99_ms", 0.0):.2f}'),
])))
# Readiness
summary_cards.append(_card("readiness", _kv_table([
(k, v) for k, v in (readiness or {}).items()
])))
# Auth keys
if keys:
key_rows = "".join(
f'| {_esc(k.get("key_hash_prefix","?"))} | '
f'{_esc(k.get("label",""))} | '
f'{_esc(", ".join(k.get("scopes") or []))} | '
f''
f'{"revoked" if k.get("revoked") else "active"} |
'
for k in keys
)
keys_body = (
''
'| hash | '
'label | '
'scopes | '
'state |
|---|
'
+ key_rows + '
'
)
else:
keys_body = 'auth disabled — no keys configured
'
summary_cards.append(_card(f"API keys ({len(keys)})", keys_body))
# Documents
doc_rows = "".join(
f'| {_esc(d.get("id",""))} | '
f'{d.get("text_len",0)} chars | '
f'{_esc(json.dumps(d.get("metadata") or {}, ensure_ascii=False))} | '
f'
' for d in documents[:50]
) or '| no documents indexed |
'
docs_body = (
''
'| id | '
'length | '
'metadata |
|---|
'
+ doc_rows + '
'
)
summary_cards.append(_card(f"documents ({len(documents)})", docs_body))
# ---------- log rows
def _log_rows(rows: List[Dict[str, Any]]) -> str:
if not rows:
return 'no entries
'
out = ['']
out.append(
''
'| ts | '
'method | '
'path | '
'status | '
'ms | '
'request_id |
'
)
for r in rows[-20:][::-1]:
status = r.get("status") or 0
color = ("#16a34a" if 200 <= int(status or 0) < 300
else "#d97706" if 300 <= int(status or 0) < 500
else "#dc2626" if int(status or 0) >= 500 else "#6b7280")
ts = time.strftime("%H:%M:%S",
time.localtime(r.get("ts") or time.time()))
out.append(
f'| {ts} | '
f'{_esc(r.get("method",""))} | '
f'{_esc(r.get("path",""))} | '
f'{_esc(status)} | '
f'{r.get("latency_ms",0):.1f} | '
f'{_esc(r.get("request_id",""))} |
'
)
out.append('
')
return "".join(out)
meta_refresh = (f''
if refresh_sec and refresh_sec > 0 else '')
body = f"""
{meta_refresh}
TAU-RAG · Admin UI
TAU-RAG · Admin UI
rendered {ts_str}
{"· auto-refresh " + str(refresh_sec) + "s" if refresh_sec else ""}
/health · /livez · /readyz · /metrics · /docs
System
{"".join(summary_cards)}
Recent requests
{_log_rows(recent_requests)}
Recent audit events
{_log_rows(recent_audits)}
"""
return body
__all__ = ["render_admin_ui"]