"""Unified admin HTML dashboard (v2.29).
Consolidates the Many(TM) admin endpoints into one page operators can
bookmark. Pattern mirrors v1.44 admin UI, v1.86 content health, v1.98
query analytics, v2.5 span timeline: inline CSS, zero JS, escape-safe,
meta-refresh supported.
Sections:
- HEADER: version + uptime + current state
- TRAFFIC: request volume / latency percentiles / status
- RATE CONTROL: rate limiter + quota + body limit + compression
- OBSERVABILITY: slow queries + obs log size + span count
- COST: total spend + top spenders
- SECURITY: maintenance state + IP allowlist + PII redactor
- HEALTH: readiness probe breakdown
"""
from __future__ import annotations
import html
from typing import Any, Dict, List, Optional
def _h(s: Any) -> str:
"""HTML-escape any value."""
return html.escape(str(s), quote=True)
def _section(title: str, emoji: str, body_html: str) -> str:
return (
f''
f'{emoji} {_h(title)}
'
f'{body_html}'
f''
)
def _kv_table(rows: List[tuple]) -> str:
trs = []
for k, v in rows:
trs.append(f'
| {_h(k)} | '
f'{_h(v)} |
')
return '"
def _status_pill(ok: bool, text: str) -> str:
cls = "pill good" if ok else "pill bad"
return f'{_h(text)}'
def _build_traffic(obs_stats: Dict, limiter_stats: Dict) -> str:
lat = obs_stats.get("latency_ms") or {}
by_status = obs_stats.get("by_status") or {}
rows = [
("total requests (recent)", obs_stats.get("n_requests", 0)),
("p50 ms", f'{lat.get("p50", 0):.1f}'),
("p95 ms", f'{lat.get("p95", 0):.1f}'),
("p99 ms", f'{lat.get("p99", 0):.1f}'),
("2xx", by_status.get("2xx", 0)),
("4xx", by_status.get("4xx", 0)),
("5xx", by_status.get("5xx", 0)),
]
return _kv_table(rows)
def _build_rate_control(limiter_stats: Dict, quota_stats: Dict,
body_stats: Dict, compression_stats: Dict) -> str:
rows = [
("rate limit", f'{limiter_stats.get("rate_per_sec", "?")} req/s, '
f'cap={limiter_stats.get("capacity", "?")}'),
("allowed / denied",
f'{limiter_stats.get("allowed", 0)} / {limiter_stats.get("denied", 0)}'),
("daily quota",
f'{quota_stats.get("n_keys_with_quota", 0)} keys configured'),
("body limit", f'{body_stats.get("max_bytes", 0)} bytes'
if body_stats.get("enabled") else "disabled"),
("compression",
f'{compression_stats.get("n_compressed", 0)} / '
f'{compression_stats.get("n_eligible", 0)} '
f'({int(100 * compression_stats.get("compress_rate", 0))}%)'
if compression_stats.get("enabled") else "disabled"),
]
return _kv_table(rows)
def _build_observability(slow_stats: Dict, obs_stats: Dict,
spans_count: int) -> str:
rows = [
("obs log size", obs_stats.get("n_requests", 0)),
("slow queries", f'{slow_stats.get("n_slow", 0)} / '
f'{slow_stats.get("n_total_observed", 0)}'),
("slow threshold", f'{slow_stats.get("threshold_ms", 0)} ms'
if slow_stats.get("enabled") else "disabled"),
("spans in ring buffer", spans_count),
]
return _kv_table(rows)
def _build_cost(cost_stats: Dict, top_spenders: List[Dict]) -> str:
rows = [
("tracked keys", cost_stats.get("n_keys", 0)),
("total calls", cost_stats.get("total_calls", 0)),
("total tokens", f'{cost_stats.get("total_tokens", 0):,}'),
("total cost (USD)",
f'${float(cost_stats.get("total_cost_usd", 0)):.4f}'),
]
html_out = _kv_table(rows)
if top_spenders:
trs = ['| prefix | calls | '
'tokens | cost |
']
for e in top_spenders[:5]:
tokens_str = f'{e["n_tokens"]:,}'
trs.append(
f'{_h(e["key_prefix"])} | '
f'{_h(e["n_calls"])} | '
f'{_h(tokens_str)} | '
f'${e["cost_usd"]:.4f} |
')
html_out += ('top spenders
")
return html_out
def _build_security(maint_stats: Dict, ip_allow_stats: Dict,
pii_stats: Dict) -> str:
maint_pill = _status_pill(
not maint_stats.get("enabled"),
f'maintenance: {"ON โ " + maint_stats.get("reason", "") if maint_stats.get("enabled") else "off"}',
)
rows = [
("drain state", maint_pill),
("IP allowlists configured",
ip_allow_stats.get("n_keys_with_allowlist", 0)),
("IP blocked / allowed",
f'{ip_allow_stats.get("n_blocked", 0)} / '
f'{ip_allow_stats.get("n_allowed", 0)}'),
("PII redactor",
"ON" if pii_stats.get("enabled") else "off"),
("PII redactions total",
pii_stats.get("total_redacted", 0)),
]
# Use non-escaped render for the pill (it's trusted HTML)
trs = []
for k, v in rows:
trs.append(f'| {_h(k)} | '
f'{v if " |
')
return '"
def _build_readiness(readiness: Dict) -> str:
ready = readiness.get("ready", False)
head = _status_pill(ready, "READY" if ready else "NOT READY")
out = [f'{head}
']
if readiness.get("checks"):
trs = [' | name | '
'critical | reason |
']
for chk in readiness["checks"]:
mark = "โ" if chk["ok"] else "โ"
crit = "critical" if chk.get("critical") else "non-critical"
reason = chk.get("reason") or ""
trs.append(
f'| {mark} | '
f'{_h(chk["name"])} | '
f'{_h(crit)} | '
f'{_h(reason)} |
')
out.append('")
return "".join(out)
def render_dashboard(*,
version: str,
obs_stats: Dict,
limiter_stats: Dict,
quota_stats: Dict,
body_stats: Dict,
compression_stats: Dict,
slow_stats: Dict,
spans_count: int,
cost_stats: Dict,
top_spenders: List[Dict],
maint_stats: Dict,
ip_allow_stats: Dict,
pii_stats: Dict,
readiness: Dict,
refresh_sec: int = 0) -> str:
"""Render the full ops dashboard as a self-contained HTML string."""
meta_refresh = (f''
if refresh_sec and refresh_sec > 0 else "")
sections = [
_section("Traffic", "๐",
_build_traffic(obs_stats, limiter_stats)),
_section("Rate control", "๐ฆ",
_build_rate_control(limiter_stats, quota_stats,
body_stats, compression_stats)),
_section("Observability", "๐ญ",
_build_observability(slow_stats, obs_stats, spans_count)),
_section("Cost", "๐ฐ",
_build_cost(cost_stats, top_spenders)),
_section("Security", "๐",
_build_security(maint_stats, ip_allow_stats, pii_stats)),
_section("Health", "๐ฉบ",
_build_readiness(readiness)),
]
return f"""
{meta_refresh}
tau-rag ยท ops dashboard
tau-rag ยท ops dashboard
version {_h(version)} ยท live snapshot
{"ยท refresh " + str(int(refresh_sec)) + "s" if refresh_sec else ""}
{"".join(sections)}
"""