"""Per-request span timeline HTML dashboard (v2.5).
Renders the v2.4 span tree as a visual gantt-style timeline:
* Horizontal bars scaled to the root span's duration.
* Indented by parent → child relationship.
* Color-coded by span name family (retrieve = blue, generate =
purple, verify = green, ...).
Same design as v1.86 content-health UI and v1.98 query-analytics UI:
inline CSS, zero JS, escape-safe, ``?refresh=N`` meta-refresh.
"""
from __future__ import annotations
import html as _h
from typing import Any, Dict, List, Optional
def _esc(s: object) -> str:
return _h.escape(str(s), quote=True)
_SPAN_COLORS = {
"pipeline": "#2563eb", # blue-600
"understand": "#0891b2", # cyan-600
"retrieve": "#059669", # emerald-600
"fuse": "#d97706", # amber-600
"rerank": "#dc2626", # red-600
"generate": "#7c3aed", # violet-600
"verify": "#16a34a", # green-600
}
_DEFAULT_COLOR = "#6b7280" # gray-500
def _color_for(name: str) -> str:
"""Pick bar color by the first token of the span name."""
head = name.split(".", 1)[0].lower()
for key, color in _SPAN_COLORS.items():
if head.startswith(key):
return color
return _DEFAULT_COLOR
def _depth_of(span: Dict, by_id: Dict[str, Dict]) -> int:
"""Walk parent_id chain to compute indentation depth."""
depth = 0
cur = span
while cur.get("parent_id"):
parent = by_id.get(cur["parent_id"])
if parent is None:
break
depth += 1
cur = parent
if depth > 20: # safety
break
return depth
def render_span_timeline(
request_id: str,
n_spans: int,
total_ms: float,
spans: List[Dict[str, Any]],
refresh_sec: int = 0,
) -> str:
"""Render the v2.4 spans payload as an HTML timeline."""
# Build parent map
by_id: Dict[str, Dict] = {s["span_id"]: s for s in spans}
# Sort spans by (depth, start-ish order). We don't have start_ts
# here, so we fall back to duration_ms and name — good enough for
# the admin viewer which just needs *a* stable order.
ordered = sorted(spans, key=lambda s: (
_depth_of(s, by_id), -float(s.get("duration_ms", 0.0))))
# Scale — root is longest, so normalize against that.
max_ms = max((float(s.get("duration_ms", 0.0)) for s in spans),
default=0.0) or 1.0
rows_html = ""
for s in ordered:
depth = _depth_of(s, by_id)
dur = float(s.get("duration_ms", 0.0))
pct = max(0.1, min(100.0, (dur / max_ms) * 100.0))
color = _color_for(s["name"])
indent_px = depth * 16
attrs = s.get("attrs") or {}
# Filter noise attrs; only show interesting ones
interesting = {}
for k in ("query_type", "method", "provider", "n_fused",
"n_candidates", "n_cited", "answer_chars",
"retrievers"):
if k in attrs:
interesting[k] = attrs[k]
attr_str = " ".join(f"{_esc(k)}={_esc(v)}"
for k, v in interesting.items())
rows_html += (
f''
f'| '
f''
f''
f'{_esc(s["name"])}'
f' | '
f'{dur:.2f}ms | '
f''
f' | '
f''
f'{_esc(attr_str)} | '
f'
'
)
# Header / table shell
if not spans:
table_html = (
'no spans for this request_id
'
)
else:
table_html = (
''
''
'| span | '
'duration | '
''
'timeline (% of root) | '
'attrs | '
'
'
+ rows_html + '
'
)
banner = (
f''
f'
request timeline
'
f'
'
f'
request_id
'
f'
'
f'{_esc(request_id)}
'
f'
n_spans
'
f'
'
f'{int(n_spans)}
'
f'
total_ms
'
f'
'
f'{float(total_ms):.2f}
'
f'
⏱️ tau-rag · span timeline
v2.4 spans rendered as a gantt-style timeline. Bars
are normalized against the root span; depth indents by parent/
child. {'Auto-refresh every ' + str(int(refresh_sec)) + 's.'
if refresh_sec > 0 else ''}
{banner}
{table_html}
"""
__all__ = ["render_span_timeline"]