titan-iac/scripts/ops/hermes_triage_monitor.py
jenkins 7d40355ff6
Some checks failed
Tests / Declarative: Post Actions testing.tests.test_repo_structure.test_knowledge_service_mirror_matches_source failed
feat(demo): show what each source contributed to the bundle
The bundle stage printed the job allowlist, which says nothing about what was
actually sent. It now samples the bundle itself, rebuilt with the same
collector Ariadne used - durable, because a finished build's console does not
change - and trimmed hard, since the point is to show what each source
contributes rather than to reprint it:

    jenkins.console_failures : 1 region(s), truncated=False
       | ERROR: Demo fixture check failed for incident hermes-triage-demo/34
    jenkins.failed_tests     : 0
    log_evidence.records     : 35 from OpenSearch kube-*
       | [jenkins] Using /home/jenkins/agent/remoting as a remoting work directory

The diagnosis now also carries run_id and run_seconds, so the run can be
opened in the Hermes dashboard to see the prompt it was given and every tool
call it made - the request, not just the answer.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 18:57:44 -03:00

403 lines
16 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
`mermaid/TestAutomation.mmd` is reached this prints, in order:
the stage banner, naming the chart 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
chart 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 chart 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
BOLD, DIM, GREEN, CYAN, YELLOW, RED, RESET = (
"\033[1m", "\033[2m", "\033[32m", "\033[36m", "\033[33m", "\033[31m", "\033[0m"
)
# stage key -> (chart subgraph, what it means, chart 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",
"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} chart: {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=16)
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{RESET}\n")
elif key == "gates":
print(f" {DIM}authorized/authorize_reason above is the policy result: the verdict"
f" Ariadne reached on its own reading{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}only these two ids may ever be executed. Source fixes are not actions:"
f" they become pull requests and are never executed by Ariadne.{RESET}\n")
elif key == "route":
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"}:
run(["kubectl", "-n", NS_DEMO, "get", "cm", "hermes-triage-demo-fixture",
"-o", "jsonpath={.data.state}"])
if key == "response":
print(f" {DIM}Scoped ConfigMap repair on the chart: one merge patch to one"
f" ConfigMap, in process, no pod created{RESET}\n")
if key == "outputs":
print(f" {YELLOW}issues filed by triage: {GITEA}/bstein/ariadne/issues{RESET}\n")
elif key == "verify":
print(f" {DIM}exactly one rebuild is triggered; it never retries in a loop{RESET}\n")
class Monitor:
"""Track which flow-chart 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} flow chart 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 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":
monitor.mark("evidence", "bundle collected")
monitor.mark("hermes", "diagnosis returned")
monitor.mark("gates", f"Ariadne refused: {state.get('reason', '')} — nothing ran")
monitor.mark(
"route",
"policy result: refused -> Diagnosis and next checks -> Ariadne opens an issue",
)
monitor.mark("response", "escalated; issue filed in the service repository")
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")