marfrit/bullpen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bullseye slice 2: doctor status panel renderer (spec + impl)

Browse Source 
render_status(doctor_json) -> self-contained HTML fragment: host vitals
table (None-safe load/temp/mem, hot marker >=80C, unreachable marker),
agents, grinds, color-coded verdict. 14-test spec by @testdesigner,
impl by @py, reviewed APPLY-as-is by @reviewer (room msgs 216/220/224).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Markus Fritsche
2026-07-21 13:01:18 +02:00
parent 42a89e6088
commit c3cccf9671
2 changed files with 458 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bullseye/status.py
+91
View File
@@ -0,0 +1,91 @@
"""Render bullpen-doctor --json output as a self-contained HTML fragment."""
from html import escape
def render_status(doctor_json):
"""Return a single HTML fragment string describing the doctor's report."""
ts = doctor_json["ts"]
hosts = doctor_json["hosts"]
agents = doctor_json["agents"]
grinds = doctor_json["grinds"]
verdict = doctor_json["verdict"]
flags = doctor_json["flags"]
parts = []
# Timestamp
parts.append(f'<div style="font-family:monospace;font-size:12px;color:#666;">{escape(str(ts))}</div>')
# Hosts table
if hosts:
parts.append('<table>')
for hostname in sorted(hosts):
info = hosts[hostname]
load = info["load"]
temp_c = info["temp_c"]
mem = info["mem"]
reachable = info["reachable"]
# None-safe rendering
load_str = escape(str(load)) if load is not None else "-"
temp_str = escape(str(temp_c)) if temp_c is not None else "-"
mem_str = escape(str(mem)) if mem is not None else "-"
# Hot styling
hot_style = ' style="background-color:#ffe0e0;font-weight:bold;"' if temp_c is not None and temp_c >= 80 else ''
unreachable_text = ' <span style="color:red;">unreachable</span>' if not reachable else ''
parts.append(
f'<tr data-host="{escape(hostname, quote=True)}"{hot_style}>'
f'<td>{escape(hostname)}</td>'
f'<td>{load_str}</td>'
f'<td>{temp_str}</td>'
f'<td>{mem_str}</td>'
f'<td>{unreachable_text}</td>'
f'</tr>'
)
parts.append('</table>')
else:
parts.append('<div>No hosts</div>')
# Agents
if agents:
parts.append('<div style="margin-top:8px;"><strong>Agents:</strong></div>')
for agent in agents:
name = agent["agent"]
model = agent["model"]
age = agent["age"]
parts.append(
f'<div data-agent="{escape(name, quote=True)}">'
f'{escape(name)}: {escape(model)} ({escape(age)})'
f'</div>'
)
else:
parts.append('<div style="margin-top:8px;">No agents</div>')
# Grinds
parts.append(f'<div data-grinds="{len(grinds)}" style="margin-top:8px;">')
if grinds:
parts.append('<strong>Grinds:</strong>')
for g in grinds:
parts.append(f'<div>{escape(g["age"])}</div>')
else:
parts.append('No grinds')
parts.append('</div>')
# Verdict
verdict_styles = {
"busy-healthy": ' style="color:green;font-weight:bold;"',
"idle": ' style="color:grey;font-weight:bold;"',
"needs-a-look": ' style="color:orange;font-weight:bold;"',
}
vs = verdict_styles.get(verdict, '')
parts.append(
f'<div data-verdict="{escape(verdict, quote=True)}"{vs}>'
f'{escape(verdict)}'
f'</div>'
)
return ''.join(parts)
tests/test_bullseye_status.py
+367
View File
@@ -0,0 +1,367 @@
# Executable spec for bullseye/status.py :: render_status(doctor_json) -> str
#
# Input: the dict emitted by `bullpen-doctor --json`:
# ts (epoch int); hosts (hostname -> {load: str|None, temp_c: int|None,
# mem: str|None, reachable: bool} — load/mem/temp_c are ALL null when the
# host is fully unreachable); agents (list of {agent, model, pid, age});
# grinds (list
# of {pid, age}); verdict ("busy-healthy"|"idle"|"needs-a-look");
# flags ({hot: list, long_jobs: int}).
#
# Contract (machine-checkable hooks the implementation MUST provide):
# * Output is a single self-contained HTML fragment string. Inline style="..."
# only — no <link>, <script>, @import, url(), src=, or http(s) URLs.
# * Each host row carries data-host="<hostname>" exactly once (value
# html.escape'd, quote=True). The chunk from that marker to the next
# data-* marker shows the host's load, temp_c digits and mem string.
# * Host rows are <tr>/<td> cells wrapped in a <table>...</table> element:
# "<table" opens before the first "<tr" and "</table>" closes after the
# last "</tr>" — a bare <tr> fragment is silently dropped by innerHTML
# parsing in body context.
# * temp_c is None-safe: no crash, and the literal "None" never rendered.
# * Fully-unreachable host ({"load": null, "temp_c": null, "mem": null,
# "reachable": false} straight from doctor): no crash; the null load and
# null mem cells render as a dash "-" (same None-safe rule as temp_c:
# the literal "None" is never rendered); the row still carries its
# data-host marker and the "unreachable" word.
# * Hot temp marker: a host with temp_c >= 80 carries inline styling that
# cool hosts (temp_c < 80) do not; cool hosts share identical styling
# regardless of their temp value (marker is discrete at the 80 boundary,
# not a per-value gradient).
# * A host with reachable=False shows the word "unreachable" in its chunk;
# reachable hosts never do.
# * Each agent carries data-agent="<agent name>" exactly once; its chunk
# shows the agent's model and age strings.
# * Grinds: one element with data-grinds="<count>" (count = len(grinds),
# including 0); each grind's age string appears in the output.
# * Verdict: one element with data-verdict="<verdict>"; the verdict text is
# visible outside the attribute; its inline styling is pairwise distinct
# across the three verdicts and carries the agreed color signal:
# busy-healthy=green, idle=grey, needs-a-look=amber/orange.
# * All string values HTML-escaped; hostnames/agent names are untrusted.
# * Pure function: identical output on repeated calls, input not mutated.
#
# Fixture-only: no live doctor call, no network.
import copy
import re
from html import escape
from bullseye.status import render_status
# ---------------------------------------------------------------- fixtures
def healthy():
"""Real bullpen-doctor --json sample, all green."""
return {
"ts": 1784624757,
"hosts": {
"boltzmann": {"load": "1.04", "temp_c": 29,
"mem": "15918/31588MB", "reachable": True},
"noether": {"load": "0.56", "temp_c": 48,
"mem": "3445/7820MB", "reachable": True},
},
"agents": [{"agent": "foreman", "model": "fable",
"pid": "345097", "age": "24s"}],
"grinds": [],
"verdict": "busy-healthy",
"flags": {"hot": [], "long_jobs": 0},
}
def degraded():
"""Hot host + unreachable host with null temp + running grinds."""
return {
"ts": 1784624999,
"hosts": {
"boltzmann": {"load": "7.91", "temp_c": 85,
"mem": "30001/31588MB", "reachable": True},
"noether": {"load": "0.00", "temp_c": None,
"mem": "0/7820MB", "reachable": False},
},
"agents": [
{"agent": "foreman", "model": "fable", "pid": "345097", "age": "24s"},
{"agent": "py", "model": "qwen", "pid": "400123", "age": "3h12m"},
],
"grinds": [{"pid": "41234", "age": "3m"}, {"pid": "41777", "age": "47m"}],
"verdict": "needs-a-look",
"flags": {"hot": ["boltzmann"], "long_jobs": 1},
}
def dark_host():
"""One healthy host plus a fully-unreachable one, exactly as
bullpen-doctor --json emits it: load/temp_c/mem all null."""
return {
"ts": 1784625100,
"hosts": {
"boltzmann": {"load": "1.04", "temp_c": 29,
"mem": "15918/31588MB", "reachable": True},
"pve3": {"load": None, "temp_c": None,
"mem": None, "reachable": False},
},
"agents": [],
"grinds": [],
"verdict": "needs-a-look",
"flags": {"hot": [], "long_jobs": 0},
}
def idle_empty():
return {
"ts": 1784600000,
"hosts": {},
"agents": [],
"grinds": [],
"verdict": "idle",
"flags": {"hot": [], "long_jobs": 0},
}
def hostile():
return {
"ts": 1784624757,
"hosts": {
"<script>alert(1)</script>": {"load": "0.10", "temp_c": 40,
"mem": "1/2MB &<x>", "reachable": True},
},
"agents": [{"agent": "<script>alert(2)</script>",
"model": "<b>evil&co</b>", "pid": "1", "age": "9s"}],
"grinds": [],
"verdict": "idle",
"flags": {"hot": [], "long_jobs": 0},
}
def one_host(temp_c):
"""Single host whose only variable is temp_c (for the 80-degree boundary).
flags.hot mirrors doctor behaviour."""
hot = ["vulcan"] if (temp_c is not None and temp_c >= 80) else []
return {
"ts": 1784624757,
"hosts": {"vulcan": {"load": "0.50", "temp_c": temp_c,
"mem": "1000/2000MB", "reachable": True}},
"agents": [],
"grinds": [],
"verdict": "idle",
"flags": {"hot": hot, "long_jobs": 0},
}
# ---------------------------------------------------------------- helpers
MARKER = re.compile(r'data-(host|agent|verdict|grinds)\s*=\s*"([^"]*)"')
def marked_chunks(html):
"""Map (kind, value) -> chunk from that data-* marker to the next one."""
ms = list(MARKER.finditer(html))
out = {}
for i, m in enumerate(ms):
key = (m.group(1), m.group(2))
assert key not in out, "duplicate marker %s=%r" % key
end = ms[i + 1].start() if i + 1 < len(ms) else len(html)
out[key] = html[m.start():end]
return out
def host_chunk(html, hostname):
by_key = marked_chunks(html)
key = ("host", escape(hostname, quote=True))
assert key in by_key, "no data-host marker for %r; markers: %s" \
% (hostname, sorted(by_key))
return by_key[key]
def styles(chunk):
"""Set of inline style attribute values in a chunk."""
found = re.findall(r"style\s*=\s*(?:\"([^\"]*)\"|'([^']*)')", chunk)
return frozenset((a or b).strip() for a, b in found)
# --- crude color classifier: green / grey / amber / other -----------------
NAMED_COLORS = {
"green": (0, 128, 0), "darkgreen": (0, 100, 0),
"forestgreen": (34, 139, 34), "seagreen": (46, 139, 87),
"limegreen": (50, 205, 50), "mediumseagreen": (60, 179, 113),
"grey": (128, 128, 128), "gray": (128, 128, 128),
"dimgrey": (105, 105, 105), "dimgray": (105, 105, 105),
"darkgrey": (169, 169, 169), "darkgray": (169, 169, 169),
"lightgrey": (211, 211, 211), "lightgray": (211, 211, 211),
"silver": (192, 192, 192), "slategray": (112, 128, 144),
"slategrey": (112, 128, 144),
"orange": (255, 165, 0), "darkorange": (255, 140, 0),
"goldenrod": (218, 165, 32), "gold": (255, 215, 0),
}
def classify(r, g, b):
if max(r, g, b) - min(r, g, b) <= 32:
return "grey"
if g > r + 16 and g > b + 16:
return "green"
if r >= 140 and r >= g and g >= b + 40 and g >= 70:
return "amber"
return "other"
def hues_in(chunk):
"""Classes of every color literal found inside style attributes."""
hues = set()
for style in styles(chunk):
s = style.lower()
for h in re.findall(r"#([0-9a-f]{6})\b", s):
hues.add(classify(*(int(h[i:i + 2], 16) for i in (0, 2, 4))))
for h in re.findall(r"#([0-9a-f]{3})\b", s):
hues.add(classify(*(int(c * 2, 16) for c in h)))
for r_, g_, b_ in re.findall(
r"rgba?\(\s*(\d+)\s*,\s*(\d+)\s*,\s*(\d+)", s):
hues.add(classify(int(r_), int(g_), int(b_)))
for name, rgb in NAMED_COLORS.items():
if re.search(r"\b%s\b" % name, s):
hues.add(classify(*rgb))
return hues
# ---------------------------------------------------------------- tests
def test_returns_html_fragment_for_all_fixtures():
for fx in (healthy(), degraded(), idle_empty(), hostile()):
out = render_status(fx)
assert isinstance(out, str) and out.strip()
assert "<" in out and ">" in out
def test_host_row_shows_load_temp_mem():
fx = healthy()
out = render_status(fx)
for name, vitals in fx["hosts"].items():
chunk = host_chunk(out, name)
assert vitals["load"] in chunk, "%s: load missing" % name
assert str(vitals["temp_c"]) in chunk, "%s: temp missing" % name
assert vitals["mem"] in chunk, "%s: mem missing" % name
def test_null_temp_renders_without_crash_or_none_leak():
out = render_status(degraded()) # noether has temp_c=None
chunk = host_chunk(out, "noether")
assert "None" not in chunk, "str(None) leaked into host row"
assert "0.00" in chunk and "0/7820MB" in chunk, \
"load/mem must still render when temp_c is null"
def test_hot_temp_marker_switches_exactly_at_80():
cool_lo = styles(host_chunk(render_status(one_host(45)), "vulcan"))
cool_hi = styles(host_chunk(render_status(one_host(79)), "vulcan"))
hot_edge = styles(host_chunk(render_status(one_host(80)), "vulcan"))
hot = styles(host_chunk(render_status(one_host(85)), "vulcan"))
assert hot_edge, "hot host row has no inline style at all"
assert cool_lo == cool_hi, \
"styling must not vary between cool temps 45 and 79: %s vs %s" \
% (sorted(cool_lo), sorted(cool_hi))
assert hot_edge != cool_hi, "temp_c=80 must be styled as hot (>= boundary)"
assert hot != cool_hi, "temp_c=85 must be styled as hot"
def test_unreachable_host_carries_marker():
out = render_status(degraded())
assert "unreachable" in host_chunk(out, "noether").lower()
assert "unreachable" not in host_chunk(out, "boltzmann").lower(), \
"reachable host must not be flagged unreachable"
def test_agents_show_name_model_and_age():
for fx in (healthy(), degraded()):
out = render_status(fx)
by_key = marked_chunks(out)
for a in fx["agents"]:
key = ("agent", escape(a["agent"], quote=True))
assert key in by_key, "no data-agent marker for %r" % a["agent"]
chunk = by_key[key]
assert a["model"] in chunk, "%s: model missing" % a["agent"]
assert a["age"] in chunk, "%s: age missing" % a["agent"]
def test_grinds_count_and_ages_shown():
fx = degraded()
out = render_status(fx)
assert out.count('data-grinds="2"') == 1, \
"expected exactly one data-grinds=\"2\" marker"
for g in fx["grinds"]:
assert g["age"] in out, "grind age %r missing" % g["age"]
def test_idle_empty_input_renders_empty_sections():
out = render_status(idle_empty())
by_key = marked_chunks(out)
kinds = [k for k, _ in by_key]
assert "host" not in kinds and "agent" not in kinds, \
"empty input must not invent host/agent rows"
assert ("grinds", "0") in by_key, "data-grinds=\"0\" marker missing"
assert ("verdict", "idle") in by_key, "data-verdict marker missing"
def test_verdict_distinct_color_signal_per_verdict():
expected_hue = {"busy-healthy": "green", "idle": "grey",
"needs-a-look": "amber"}
signatures = {}
for verdict, hue in expected_hue.items():
fx = healthy()
fx["verdict"] = verdict
out = render_status(fx)
by_key = marked_chunks(out)
key = ("verdict", verdict)
assert key in by_key, "no data-verdict marker for %r" % verdict
chunk = by_key[key]
sans_marker = chunk.replace('data-verdict="%s"' % verdict, "", 1)
assert verdict in sans_marker, \
"verdict %r not visible as text" % verdict
sig = styles(chunk)
assert sig, "verdict %r element has no inline style" % verdict
assert hue in hues_in(chunk), \
"verdict %r must signal %s; styles: %s" % (verdict, hue, sorted(sig))
signatures[verdict] = sig
assert len(set(signatures.values())) == 3, \
"verdict styles must be pairwise distinct, got: %s" % signatures
def test_html_escaping_no_raw_passthrough():
out = render_status(hostile())
assert "<script" not in out.lower(), "raw <script> passed through"
assert escape("<script>alert(1)</script>") in out, \
"escaped hostname payload missing"
assert escape("<script>alert(2)</script>") in out, \
"escaped agent-name payload missing"
assert "<b>evil" not in out and escape("<b>evil&co</b>") in out, \
"model string not escaped"
assert "&<x>" not in out and "1/2MB &amp;&lt;x&gt;" in out, \
"mem string not escaped"
# markers must use the escaped value too (attribute context)
by_key = marked_chunks(out)
assert ("host", escape("<script>alert(1)</script>", quote=True)) in by_key
assert ("agent", escape("<script>alert(2)</script>", quote=True)) in by_key
def test_no_external_resources_anywhere():
for fx in (healthy(), degraded(), idle_empty(), hostile()):
out = render_status(fx).lower()
for banned in ("<link", "<script", "@import", "url(", "src=",
"http://", "https://"):
assert banned not in out, \
"self-contained fragment must not contain %r" % banned
def test_pure_function_no_mutation_and_deterministic():
for fx in (healthy(), degraded(), idle_empty(), hostile()):
snapshot = copy.deepcopy(fx)
out1 = render_status(fx)
out2 = render_status(fx)
assert out1 == out2, "same input must give byte-identical output"
assert fx == snapshot, "input dict was mutated"
def test_fully_unreachable_host_null_load_mem_render_as_dash():
out = render_status(dark_host()) # must not raise on null load/mem
chunk = host_chunk(out, "pve3") # data-host marker must survive
assert "None" not in chunk, "str(None) leaked into fully-dark host row"
assert "unreachable" in chunk.lower(), \
"fully-dark host must still carry the unreachable marker"
cells = [c.strip() for c in re.findall(r">([^<>]*)<", chunk)]
dashes = [c for c in cells if c in ("-", "–", "—")]
assert len(dashes) >= 2, \
"null load and null mem must each render as a dash cell " \
"(same None-safe approach as temp_c); cell texts: %r" \
% [c for c in cells if c]
assert "1.04" in host_chunk(out, "boltzmann"), \
"reachable sibling host must still render its real load"
def test_host_rows_wrapped_in_table_for_innerhtml():
for fx in (healthy(), degraded(), hostile()):
out = render_status(fx)
first_tr = out.find("<tr")
last_tr_close = out.rfind("</tr>")
assert first_tr != -1 and last_tr_close != -1, \
"host rows must be <tr> elements"
table_open = out.find("<table")
table_close = out.rfind("</table>")
assert table_open != -1 and table_open < first_tr, \
"'<table' must precede the first '<tr' -- a bare <tr> fragment " \
"is dropped by innerHTML parsing in body context"
assert table_close > last_tr_close, \
"'</table>' must follow the last '</tr>'"