5fd5e6d729
The scripts said "flow chart" while the artifact the audience is following has a name. Telling a room to follow along on the flow chart and then handing them something titled Test Automation Diagram makes them hunt for the match. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
468 lines
20 KiB
Python
Executable File
468 lines
20 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Narrate the triage flow live, showing every command it runs.
|
|
|
|
Run in a second terminal beside the demo. When a stage of
|
|
the Test Automation Diagram (`mermaid/TestAutomation.mmd`) is reached this
|
|
prints, in order:
|
|
|
|
the stage banner, naming the diagram subgraph
|
|
the service UI to look at, if one changes at that stage
|
|
each command, echoed before it runs, then its output
|
|
|
|
Echoing the commands is the point. An audience watching a dashboard has to
|
|
take the result on trust; watching `kubectl` run against the cluster and
|
|
reading the raw answer is the difference between a demonstration and an
|
|
assertion.
|
|
|
|
Two honest limits. This reports at subgraph granularity, not per node: the
|
|
diagram draws the collector, response check, marker check and authorizer
|
|
separately, and a refusal here names the gate rather than walking all twelve.
|
|
And it follows one incident on one job, whereas the diagram describes the whole
|
|
system.
|
|
|
|
Read-only. Every command below is a read; nothing here changes the cluster.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
JOB = os.environ.get("MONITOR_JOB", "hermes-triage-demo")
|
|
NS_DEMO = "hermes-triage-demo"
|
|
NS_ARIADNE = "maintenance"
|
|
JENKINS = os.environ.get("JENKINS_URL", "https://ci.bstein.dev")
|
|
GITEA = os.environ.get("GITEA_URL", "https://scm.bstein.dev")
|
|
GRAFANA = os.environ.get("GRAFANA_URL", "https://metrics.bstein.dev")
|
|
HERMES_UI = os.environ.get("HERMES_URL", "https://agent.bstein.dev")
|
|
POLL_SECONDS = 6
|
|
|
|
# The fixture job repairs a ConfigMap; the code job proposes a patch. Evidence
|
|
# that suits one is false for the other, so the stages branch on it.
|
|
IS_CODE_JOB = JOB == "hermes-code-demo"
|
|
|
|
BOLD, DIM, GREEN, CYAN, YELLOW, RED, RESET = (
|
|
"\033[1m", "\033[2m", "\033[32m", "\033[36m", "\033[33m", "\033[31m", "\033[0m"
|
|
)
|
|
|
|
# stage key -> (diagram subgraph, what it means, diagram path, UI worth showing)
|
|
STAGES: dict[str, tuple[str, str, str, str]] = {
|
|
"detect": (
|
|
"Detect and gather",
|
|
"A failed build becomes one incident",
|
|
"Jenkins detector -> Evidence sources",
|
|
f"{JENKINS}/job/{JOB}/ — the red build",
|
|
),
|
|
"evidence": (
|
|
"Detect and gather",
|
|
"Ariadne assembles the bounded evidence bundle",
|
|
"Console reader -> Failure ranker -> Context filter -> Incident bundle",
|
|
"",
|
|
),
|
|
"hermes": (
|
|
"Hermes analysis",
|
|
"Hermes returns a recommendation it cannot act on",
|
|
"Triage skills -> Structured recommendation",
|
|
f"{HERMES_UI} — the agent run appears here",
|
|
),
|
|
"gates": (
|
|
"Ariadne policy gates",
|
|
"Ariadne decides on its own reading of the evidence",
|
|
"Response check -> Scoped repair guard -> Action authorizer",
|
|
"",
|
|
),
|
|
"route": (
|
|
"Ariadne response",
|
|
"The policy result, and which branch it opens",
|
|
"Policy result -> Action registry",
|
|
"",
|
|
),
|
|
"response": (
|
|
"Ariadne response",
|
|
"Ariadne executes, proposes, or escalates",
|
|
"Action registry -> Scoped ConfigMap repair -> Action result",
|
|
"",
|
|
),
|
|
"verify": (
|
|
"Ariadne response",
|
|
"The branch build checks the proposal" if JOB == "hermes-code-demo"
|
|
else "One rebuild decides whether the repair held",
|
|
"Action result -> validation build",
|
|
f"{JENKINS}/job/{JOB}/ — a new build starts on its own",
|
|
),
|
|
"outputs": (
|
|
"Inspectable outputs",
|
|
"Incident closed; the artifacts remain",
|
|
"Ariadne records every outcome -> Audit events, Triage metrics",
|
|
f"{GRAFANA}/d/atlas-testing — triage panels",
|
|
),
|
|
}
|
|
ORDER = ["detect", "evidence", "hermes", "gates", "route", "response", "verify", "outputs"]
|
|
|
|
|
|
def stamp() -> str:
|
|
return datetime.now(timezone.utc).strftime("%H:%M:%S")
|
|
|
|
|
|
def run(cmd: list[str], *, show: bool = True, limit: int = 12) -> str:
|
|
"""Echo a command, run it, print its output, and return stdout."""
|
|
|
|
if show:
|
|
print(f" {DIM}${RESET} {CYAN}{' '.join(cmd)}{RESET}")
|
|
try:
|
|
# Ariadne emits ~200 lines per tick, so a ten-minute window is a
|
|
# few thousand lines; the fetch needs room to finish.
|
|
done = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
|
out = done.stdout.strip()
|
|
except Exception as exc: # a failed read must never stop the narration
|
|
print(f" {RED}(command failed: {exc}){RESET}")
|
|
return ""
|
|
if show:
|
|
for line in (out.split("\n")[:limit] or ["(no output)"]):
|
|
print(f" {line[:150]}")
|
|
print()
|
|
return out
|
|
|
|
|
|
def quiet(cmd: list[str]) -> str:
|
|
return run(cmd, show=False)
|
|
|
|
|
|
def banner(key: str) -> None:
|
|
subgraph, meaning, path, ui = STAGES[key]
|
|
print(f"\n{GREEN}{'─' * 70}{RESET}")
|
|
print(f"{GREEN}{BOLD} {stamp()} {subgraph}{RESET}{GREEN} — {meaning}{RESET}")
|
|
print(f"{DIM} diagram: {path}{RESET}")
|
|
if ui:
|
|
print(f"{YELLOW} look at: {ui}{RESET}")
|
|
print(f"{GREEN}{'─' * 70}{RESET}")
|
|
|
|
|
|
def ariadne_records(since: str = "10m") -> list[dict]:
|
|
"""Return recent Ariadne log records that parse as JSON.
|
|
|
|
Windowed by time, not line count. Ariadne emits roughly two hundred lines
|
|
per tick once it is polling every job, so a --tail window wide enough to
|
|
be useful is impossible to guess and a narrow one silently drops ticks.
|
|
"""
|
|
|
|
raw = quiet(
|
|
["kubectl", "-n", NS_ARIADNE, "logs", "deploy/ariadne", "-c", "ariadne", f"--since={since}"]
|
|
)
|
|
records = []
|
|
for line in raw.split("\n"):
|
|
line = line.strip()
|
|
if line.startswith("{"):
|
|
try:
|
|
records.append(json.loads(line))
|
|
except ValueError:
|
|
continue
|
|
return records
|
|
|
|
|
|
def job_states(records: list[dict]) -> list[dict]:
|
|
"""Return every tick result for this job in the window, oldest first.
|
|
|
|
The stages are reconstructed from all of them rather than from the newest
|
|
alone. A tick reporting the repair is replaced by the next tick within
|
|
seconds, so reading only the latest state means a poll landing at the
|
|
wrong moment loses that step permanently.
|
|
"""
|
|
|
|
states = []
|
|
for record in records:
|
|
if record.get("event") == "hermes_autotriage" and record.get("jobs"):
|
|
try:
|
|
state = json.loads(record["jobs"]).get(JOB)
|
|
except ValueError:
|
|
continue
|
|
if state:
|
|
states.append(state)
|
|
return states
|
|
|
|
|
|
_DIAG_QUERY = (
|
|
". /vault/secrets/ariadne-env.sh >/dev/null 2>&1; python3 -c \""
|
|
"import json,os,psycopg;"
|
|
"conn=psycopg.connect(os.environ['ARIADNE_DATABASE_URL']);"
|
|
"cur=conn.cursor();"
|
|
"cur.execute(\\\"select detail from ariadne_events where event_type='hermes_autotriage_diagnosis'"
|
|
" order by id desc limit 1\\\");"
|
|
"r=cur.fetchone()[0];"
|
|
"d=r if isinstance(r,dict) else json.loads(r);"
|
|
"print(json.dumps({'incident':d.get('incident_id'),'authorized':d.get('authorized'),"
|
|
"'authorize_reason':d.get('authorize_reason'),'evidence_marker':d.get('evidence_marker'),"
|
|
"'run_id':(d.get('run') or {}).get('run_id'),"
|
|
"'run_seconds':(d.get('run') or {}).get('duration_seconds'),"
|
|
"'outcome':d.get('outcome')},indent=1))\""
|
|
)
|
|
|
|
|
|
_HISTORY_QUERY = (
|
|
". /vault/secrets/ariadne-env.sh >/dev/null 2>&1; python3 -c \""
|
|
"import json,os,psycopg;"
|
|
"conn=psycopg.connect(os.environ['ARIADNE_DATABASE_URL']);"
|
|
"cur=conn.cursor();"
|
|
"cur.execute(\\\"select created_at,detail from ariadne_events where"
|
|
" event_type='hermes_autotriage_incident' order by id desc limit 40\\\");"
|
|
"rows=[(t,d if isinstance(d,dict) else json.loads(d)) for t,d in cur.fetchall()];"
|
|
"rows=[r for r in rows if r[1].get('incident_id')=='__INCIDENT__'];"
|
|
"[print(str(t)[11:19], v.get('status'), json.dumps(v.get('phase') or {})[:80])"
|
|
" for t,v in reversed(rows)]\""
|
|
)
|
|
|
|
|
|
_BUNDLE_QUERY = (
|
|
"python3 -c \""
|
|
"from ariadne.services import hermes_autotriage_evidence as ev;"
|
|
"lb={'number':__BUILD__,'result':'FAILURE','building':False,'timestamp':0,'duration':0,'url':''};"
|
|
"b=ev.collect_evidence('__INCIDENT__','__JOB__',lb);"
|
|
"j=b['jenkins']; regions=j.get('console_failures') or []; tests=j.get('failed_tests') or [];"
|
|
"recs=(b.get('log_evidence') or {}).get('records') or [];"
|
|
"print('jenkins.console_failures : %d region(s), truncated=%s' % (len(regions), j.get('console_truncated')));"
|
|
"[print(' | ' + l[:110]) for l in ((regions[0].get('text') or '').strip().split(chr(10))[-3:] if regions else [])];"
|
|
"print('jenkins.failed_tests : %d' % len(tests));"
|
|
"[print(' | %s :: %s' % (t.get('className'), t.get('name'))) for t in tests[:2]];"
|
|
"print('log_evidence.records : %d from OpenSearch kube-*' % len(recs));"
|
|
"[print(' | [%s] %s' % (r.get('namespace'), (r.get('message') or '')[:90])) for r in recs[:2]]\""
|
|
)
|
|
|
|
|
|
def bundle_sample(incident: str) -> None:
|
|
"""Show a trimmed sample of the bundle that was sent to Hermes.
|
|
|
|
Rebuilt from the same collector Ariadne used. A finished build's console
|
|
does not change, so this is durable rather than a live reading, and it is
|
|
trimmed hard on purpose: the point is to show what kind of evidence each
|
|
source contributes, not to reprint the bundle.
|
|
"""
|
|
|
|
if "/" not in incident:
|
|
return
|
|
job, _, build = incident.rpartition("/")
|
|
query = (
|
|
_BUNDLE_QUERY.replace("__BUILD__", build)
|
|
.replace("__INCIDENT__", incident)
|
|
.replace("__JOB__", job)
|
|
)
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"sh", "-c", ". /vault/secrets/ariadne-env.sh >/dev/null 2>&1; " + query], limit=14)
|
|
|
|
|
|
def incident_history(incident: str) -> None:
|
|
"""Show this incident's recorded state changes, oldest first.
|
|
|
|
Durable evidence on purpose. A stage describes a moment that has passed,
|
|
so reading live cluster state at that point misrepresents it: by the time
|
|
the detection stage is narrated the repair has already run, and the
|
|
fixture would read healthy as though it had never failed.
|
|
"""
|
|
|
|
if not incident:
|
|
return
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"sh", "-c", _HISTORY_QUERY.replace("__INCIDENT__", incident)], limit=10)
|
|
|
|
|
|
def diagnosis_event() -> None:
|
|
"""Show the diagnosis Ariadne stored: what Hermes said, and the verdict."""
|
|
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"sh", "-c", _DIAG_QUERY], limit=26)
|
|
|
|
|
|
def evidence_for(key: str, incident: str = "") -> None:
|
|
"""Run the reads that show this stage actually happened."""
|
|
|
|
if key == "detect":
|
|
print(f" {DIM}the incident's recorded state changes; durable, so it still reads"
|
|
f" true after the repair has run:{RESET}")
|
|
incident_history(incident)
|
|
elif key == "evidence":
|
|
print(f" {DIM}a sample of what each source contributed to the bundle Hermes"
|
|
f" received:{RESET}")
|
|
bundle_sample(incident)
|
|
elif key == "hermes":
|
|
print(f" {DIM}what Hermes actually returned, as Ariadne stored it. The run_id below"
|
|
f" identifies this run in the Hermes dashboard, where the prompt it was given"
|
|
f" and every tool call it made are visible:{RESET}")
|
|
diagnosis_event()
|
|
print(f" {DIM}Hermes holds no Git or Kubernetes write access; this JSON is its"
|
|
f" entire output. When outcome.suggested_remediation is populated, Hermes found"
|
|
f" no action that fits and is proposing one for a maintainer to build; it is"
|
|
f" recorded and printed in the issue, and no gate reads it{RESET}\n")
|
|
elif key == "gates":
|
|
print(f" {DIM}the authorization is a match between three separate things: an action"
|
|
f" Ariadne already has code to perform, an action id Hermes is permitted to"
|
|
f" request, and the action Hermes actually recommended. Ariadne holds the"
|
|
f" registry below; Hermes cannot add to it, and the recommendation only"
|
|
f" proceeds because it names something already in it.{RESET}")
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"printenv", "ARIADNE_HERMES_ALLOWED_ACTIONS"], limit=2)
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"printenv", "ARIADNE_HERMES_MIN_CONFIDENCE"], limit=2)
|
|
print(f" {DIM}a separate setting governs what Hermes may be asked to propose a fix"
|
|
f" for. It is separate because these become pull requests rather than"
|
|
f" changes Ariadne makes itself:{RESET}")
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"printenv", "ARIADNE_HERMES_FIX_CATEGORIES"], limit=2)
|
|
print()
|
|
elif key == "route":
|
|
if IS_CODE_JOB:
|
|
print(f" {DIM}a source fix is not one of the registered actions, so this takes"
|
|
f" the Optional source proposal branch rather than the Action registry.{RESET}\n")
|
|
else:
|
|
print(f" {DIM}the fixture repair is an operational action, so the Optional source"
|
|
f" proposal branch is not taken for this incident{RESET}\n")
|
|
elif key in {"response", "outputs"}:
|
|
if IS_CODE_JOB:
|
|
print(f" {DIM}the pull request Ariadne opened is recorded on the incident above"
|
|
f" (branch, pr_number, url). Hermes produced the patch as data; Ariadne"
|
|
f" validated it and pushed the branch.{RESET}")
|
|
print(f" {YELLOW}pull requests: {GITEA}/bstein/hermes-code-demo/pulls{RESET}\n")
|
|
else:
|
|
run(["kubectl", "-n", NS_DEMO, "get", "cm", "hermes-triage-demo-fixture",
|
|
"-o", "jsonpath={.data.state}"])
|
|
if key == "response":
|
|
print(f" {DIM}that value read 'unhealthy' when the build failed - the seeded"
|
|
f" fault - and reads 'healthy' above because Ariadne has just patched"
|
|
f" it. That single field changing is the repair.{RESET}\n")
|
|
if key == "outputs":
|
|
print(f" {DIM}on the diagram this is the 'records and artifacts' edge out of the"
|
|
f" whole Ariadne response box, not out of one branch. Every path ends"
|
|
f" here: an executed action, an escalation, or a pull request all record"
|
|
f" the same audit events and metrics.{RESET}")
|
|
print(f" {YELLOW}issues filed by triage: {GITEA}/bstein/ariadne/issues{RESET}\n")
|
|
elif key == "verify":
|
|
if IS_CODE_JOB:
|
|
print(f" {DIM}the branch build validates the proposal. Nothing merges"
|
|
f" automatically: the incident stays human-required whatever the branch"
|
|
f" build says, because a person decides whether the fix is right.{RESET}\n")
|
|
else:
|
|
print(f" {DIM}exactly one rebuild is triggered; it never retries in a loop. This"
|
|
f" completes the operational branch: because the action was authorized and"
|
|
f" performed, neither the human-required response nor the optional source"
|
|
f" proposal is entered for this incident.{RESET}\n")
|
|
|
|
|
|
class Monitor:
|
|
"""Track which Test Automation Diagram stage the current incident has reached."""
|
|
|
|
def __init__(self) -> None:
|
|
self.done: set[str] = set()
|
|
self.incident = ""
|
|
self.summarised = False
|
|
self.last_tick = ""
|
|
|
|
def mark(self, key: str, evidence: str) -> None:
|
|
if key in self.done:
|
|
return
|
|
self.done.add(key)
|
|
banner(key)
|
|
print(f" {BOLD}what happened:{RESET} {evidence}\n")
|
|
evidence_for(key, self.incident)
|
|
|
|
def new_incident(self, incident: str) -> None:
|
|
if incident and incident != self.incident:
|
|
self.incident = incident
|
|
self.done.clear()
|
|
self.summarised = False
|
|
print(f"\n{BOLD}{YELLOW}══ incident {incident} ══{RESET}")
|
|
|
|
def checklist(self) -> None:
|
|
print(f"\n{BOLD} Test Automation Diagram progress{RESET}")
|
|
for key in ORDER:
|
|
tick = f"{GREEN}✓{RESET}" if key in self.done else f"{DIM}·{RESET}"
|
|
print(f" {tick} {key:<9} {STAGES[key][0]:<22} {DIM}{STAGES[key][2]}{RESET}")
|
|
print()
|
|
|
|
|
|
def main() -> None:
|
|
print(f"{BOLD}Hermes triage monitor — following the Test Automation Diagram"
|
|
f" (mermaid/TestAutomation.mmd){RESET}")
|
|
print(f"{DIM}Every command is echoed before it runs. Read-only. Ctrl-C to stop.{RESET}")
|
|
print(f"{YELLOW}Jenkins: {JENKINS}/job/{JOB}/{RESET}")
|
|
print(f"{YELLOW}Gitea: {GITEA}/bstein{RESET}")
|
|
print(f"{YELLOW}Grafana: {GRAFANA}/d/atlas-testing{RESET}\n")
|
|
monitor = Monitor()
|
|
while True:
|
|
states = job_states(ariadne_records())
|
|
# Replay every tick in the window, so a step that lasted one tick is
|
|
# never lost to poll timing. mark() is idempotent.
|
|
for state in states:
|
|
status = str(state.get("status") or "")
|
|
monitor.new_incident(str(state.get("incident_id") or ""))
|
|
if status in {"human_required", "awaiting_rebuild", "deduped", "failed"}:
|
|
monitor.mark("detect", f"incident {monitor.incident} opened from a terminal failure")
|
|
if status == "awaiting_rebuild":
|
|
monitor.mark("evidence", "bundle collected: console regions, tests, logs")
|
|
monitor.mark("hermes", "diagnosis returned and parsed against the frozen schema")
|
|
monitor.mark("gates", "every gate passed; the predefined action was authorized")
|
|
monitor.mark(
|
|
"route",
|
|
"policy result: authorized -> Action registry (Optional source proposal not taken)",
|
|
)
|
|
monitor.mark(
|
|
"response",
|
|
f"Scoped ConfigMap repair: {state.get('repair', 'action')} on {state.get('target', '')}",
|
|
)
|
|
monitor.mark("verify", "one rebuild triggered with seeding disabled")
|
|
if status == "human_required":
|
|
reason = str(state.get("reason") or "")
|
|
proposed = reason == "code_fix_proposed"
|
|
monitor.mark("evidence", "bundle collected")
|
|
monitor.mark(
|
|
"hermes",
|
|
"patch proposed as data: a path, an exact anchor and a replacement"
|
|
if proposed
|
|
else "diagnosis returned",
|
|
)
|
|
monitor.mark(
|
|
"gates",
|
|
"no action was authorized; a source fix is not an action, so this takes"
|
|
" the proposal branch rather than the registry"
|
|
if proposed
|
|
else f"Ariadne refused: {reason} - nothing ran",
|
|
)
|
|
monitor.mark(
|
|
"route",
|
|
"policy result: no action -> Optional source proposal -> Patch validator"
|
|
if proposed
|
|
else "policy result: refused -> Diagnosis and next checks -> opens an issue",
|
|
)
|
|
monitor.mark(
|
|
"response",
|
|
"Ariadne opens a pull request; nothing merges without a human"
|
|
if proposed
|
|
else "escalated; issue filed in the service repository",
|
|
)
|
|
if proposed:
|
|
monitor.mark(
|
|
"verify",
|
|
"the branch build validates the proposal; the incident stays"
|
|
" human-required either way",
|
|
)
|
|
if status == "healthy" and state.get("resolved") and not monitor.summarised:
|
|
monitor.mark("verify", "rebuild finished green")
|
|
monitor.mark("outputs", f"resolved: {', '.join(state['resolved'])}")
|
|
monitor.checklist()
|
|
monitor.summarised = True
|
|
status = str(states[-1].get("status") or "") if states else ""
|
|
|
|
if status and status != monitor.last_tick:
|
|
monitor.last_tick = status
|
|
print(f"{DIM}{stamp()} tick: {status}{RESET}")
|
|
sys.stdout.flush()
|
|
time.sleep(POLL_SECONDS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\nstopped")
|