"""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 '' + "".join(trs) + "
" 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 = ['prefixcalls' 'tokenscost'] 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

' + "".join(trs) + "
") 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 '' + "".join(trs) + "
" 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' 'criticalreason'] 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('' + "".join(trs) + "
") 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)}
"""