39cc744d19
Some checks failed
Tests / Declarative: Post Actions testing.tests.test_repo_structure.test_knowledge_service_mirror_matches_source failed
It read only the newest tick from a --tail=400 window. Ariadne now emits roughly two hundred lines per tick, so that window held about two ticks, and a step that lasted a single tick - the repair, most importantly - vanished if a poll landed after it. Build 30 was triaged, repaired and resolved correctly while the monitor printed nothing at all. The window is now time-based, and every tick inside it is replayed rather than just the last. mark() was already idempotent, so replaying is safe and the monitor recovers stages it was not running for. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
286 lines
11 KiB
Python
Executable File
286 lines
11 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 = 4
|
|
|
|
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",
|
|
"intake: detector -> evidence_sources",
|
|
f"{JENKINS}/job/{JOB}/ — the red build",
|
|
),
|
|
"evidence": (
|
|
"Detect and gather",
|
|
"Ariadne assembles the bounded evidence bundle",
|
|
"intake: collector -> bundle",
|
|
"",
|
|
),
|
|
"hermes": (
|
|
"Hermes analysis",
|
|
"Hermes returns a recommendation it cannot act on",
|
|
"hermes_plane: skills -> recommendation",
|
|
f"{HERMES_UI} — the agent run appears here",
|
|
),
|
|
"gates": (
|
|
"Ariadne policy gates",
|
|
"Ariadne decides on its own reading of the evidence",
|
|
"controls: response_check -> authorizer",
|
|
"",
|
|
),
|
|
"route": (
|
|
"Ariadne response",
|
|
"Which edge off the route diamond is taken",
|
|
"response: route",
|
|
"",
|
|
),
|
|
"response": (
|
|
"Ariadne response",
|
|
"Ariadne executes, proposes, or escalates",
|
|
"response: operational_path / code_path / human_path",
|
|
"",
|
|
),
|
|
"verify": (
|
|
"Ariadne response",
|
|
"One rebuild decides whether the repair held",
|
|
"response: operational_result -> validation build",
|
|
f"{JENKINS}/job/{JOB}/ — a new build starts on its own",
|
|
),
|
|
"outputs": (
|
|
"Inspectable outputs",
|
|
"Incident closed; the artifacts remain",
|
|
"outputs: audit_events, alert_output, issue_output",
|
|
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:
|
|
done = subprocess.run(cmd, capture_output=True, text=True, timeout=25)
|
|
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 = "15m") -> 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
|
|
|
|
|
|
def evidence_for(key: str) -> None:
|
|
"""Run the reads that show this stage actually happened."""
|
|
|
|
if key == "detect":
|
|
run(["kubectl", "-n", NS_DEMO, "get", "cm", "hermes-triage-demo-fixture",
|
|
"-o", "jsonpath={.data.state}"])
|
|
elif key == "evidence":
|
|
print(f" {DIM}the bundle is what Ariadne sends Hermes; the audit trail keeps it{RESET}\n")
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"printenv", "ARIADNE_HERMES_AUTOTRIAGE_JOB_ALLOWLIST"], limit=2)
|
|
elif key == "hermes":
|
|
run(["kubectl", "-n", "hermes", "get", "pods", "-l", "app=hermes", "--no-headers"], limit=3)
|
|
print(f" {DIM}Hermes holds no Git or Kubernetes write access; it returns JSON only{RESET}\n")
|
|
elif key == "gates":
|
|
for var in ("ARIADNE_HERMES_ALLOWED_ACTIONS", "ARIADNE_HERMES_MIN_CONFIDENCE"):
|
|
run(["kubectl", "-n", NS_ARIADNE, "exec", "deploy/ariadne", "-c", "ariadne", "--",
|
|
"printenv", var], limit=2)
|
|
elif key in {"response", "outputs"}:
|
|
run(["kubectl", "-n", NS_DEMO, "get", "cm", "hermes-triage-demo-fixture",
|
|
"-o", "jsonpath={.data.state}"])
|
|
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)
|
|
|
|
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]}")
|
|
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", "route --|authorized action|--> registry -> operational_result")
|
|
monitor.mark("response", f"{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", "route --|human required|--> human_required -> gitea_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")
|