"""HTML views for the Model Usage & Performance Analytics console (#651)."""
from __future__ import annotations
import html
from webui.analytics_loader import AnalyticsSnapshot, GroupMetrics, UsageEvent
from webui.layout import render_page
def _escape(text: object) -> str:
"""HTML-escape dynamic analytics fields (mirrors audit_views / project_views)."""
return html.escape(str(text), quote=True)
def _render_badge(text: str, badge_type: str = "muted") -> str:
return f'{_escape(text)}'
def _render_group_table(title: str, groups: dict[str, GroupMetrics], key_header: str = "Group") -> str:
if not groups:
return (
f"
{_escape(title)}
"
'No telemetry events recorded for this dimension.
'
)
rows = []
for key, g in sorted(groups.items(), key=lambda x: x[1].total_events, reverse=True):
cost_cell = (
f'{_escape(g.display_cost)}'
if g.events_with_cost > 0
else _render_badge("Unknown")
)
tokens_cell = (
_escape(g.display_tokens)
if g.events_with_tokens > 0
else _render_badge("Unknown")
)
lat_p50 = (
_escape(g.display_latency_p50)
if g.events_with_latency > 0
else _render_badge("Unknown")
)
lat_p90 = (
_escape(g.display_latency_p90)
if g.events_with_latency > 0
else _render_badge("Unknown")
)
dur_avg = (
_escape(g.display_duration_avg)
if g.events_with_duration > 0
else _render_badge("Unknown")
)
rows.append(
""
f"| {_escape(key)} | "
f"{g.total_events} | "
f"{tokens_cell} | "
f"{cost_cell} | "
f"{lat_p50} | "
f"{lat_p90} | "
f"{dur_avg} | "
"
"
)
rows_html = "".join(rows)
return f"""
{_escape(title)}
| {_escape(key_header)} |
Events |
Total Tokens |
Est. Cost |
Latency (p50) |
Latency (p90) |
Avg Stage Duration |
{rows_html}
"""
def _render_events_table(events: tuple[UsageEvent, ...]) -> str:
if not events:
return (
"Recent Usage & Instrumentation Events
"
'No individual telemetry events recorded yet. Opt-in instrumentation via session logging or authorized POST /api/v1/analytics/usage.
'
)
rows = []
for e in list(events)[-50:]: # Display latest 50
if e.issue_number is not None:
work_item = f"issue #{e.issue_number}"
elif e.pr_number is not None:
work_item = f"pr #{e.pr_number}"
else:
work_item = "unlinked"
tokens = (
_escape(f"{e.total_tokens:,}")
if e.total_tokens is not None
else _render_badge("Unknown")
)
cost = (
_escape(f"${e.estimated_cost_usd:.4f}")
if e.estimated_cost_usd is not None
else _render_badge("Unknown")
)
latency = (
_escape(f"{e.latency_ms} ms")
if e.latency_ms is not None
else _render_badge("Unknown")
)
duration = (
_escape(f"{e.duration_ms} ms")
if e.duration_ms is not None
else _render_badge("Unknown")
)
status_badge = _render_badge(
e.status, "success" if e.status == "success" else "danger"
)
rows.append(
""
f"| #{e.usage_id} | "
f"{_escape(e.created_at)} | "
f"{_escape(e.role)} | "
f"{_escape(e.model)} | "
f"{_escape(e.stage)} | "
f"{_escape(work_item)} | "
f"{tokens} | "
f"{cost} | "
f"{latency} | "
f"{duration} | "
f"{status_badge} | "
"
"
)
rows_html = "".join(rows)
return f"""
Recent Telemetry Events
| ID |
Timestamp |
Role |
Model |
Stage |
Work Item |
Tokens |
Cost |
Latency |
Duration |
Status |
{rows_html}
"""
def render_analytics_page(snapshot: AnalyticsSnapshot) -> str:
"""Render the main Model Usage & Performance Analytics console page."""
summary = snapshot.overall_summary
kpi_tokens = (
_escape(summary.display_tokens)
if summary.events_with_tokens > 0
else _render_badge("Unknown")
)
kpi_cost = (
_escape(summary.display_cost)
if summary.events_with_cost > 0
else _render_badge("Unknown")
)
kpi_lat_p50 = (
_escape(summary.display_latency_p50)
if summary.events_with_latency > 0
else _render_badge("Unknown")
)
kpi_dur_avg = (
_escape(summary.display_duration_avg)
if summary.events_with_duration > 0
else _render_badge("Unknown")
)
status_notice = ""
if not snapshot.ok:
status_notice = (
f'Degraded Data Source: '
f'{_escape(snapshot.reason)}
'
)
body_html = f"""
Model Usage & Performance Analytics (Phase 4)
Durable console analytics for model usage, token cost, latency percentiles, and workflow-stage performance correlated to issues, PRs, and worker roles.
{status_notice}
Note on telemetry fidelity: Missing data or untracked metrics are explicitly labeled as Unknown. No token costs or latency metrics are zero-fabricated.
Total Events
{summary.total_events}
Total Tokens
{kpi_tokens}
Est. Token Cost
{kpi_cost}
Latency (p50)
{kpi_lat_p50}
Avg Stage Duration
{kpi_dur_avg}
{_render_group_table("Usage & Cost by Model", snapshot.by_model, "Model")}
{_render_group_table("Performance by Workflow Stage", snapshot.by_stage, "Stage")}
{_render_group_table("Usage & Cost by Role", snapshot.by_role, "Role")}
{_render_group_table("Work Item Analytics", snapshot.by_work_item, "Work Item")}
{_render_events_table(snapshot.events)}
"""
return render_page(title="Model Usage & Performance Analytics", body_html=body_html)