db14a9d6a8
Some checks failed
Tests / Declarative: Post Actions testing.tests.test_repo_structure.test_knowledge_service_mirror_matches_source failed
An audience following mermaid/TestAutomation.mmd needs to know which edge off the route diamond was taken, which was the one thing the output left them to infer. It now quotes the chart's own labels - authorized action, human required, and the optional code path - and the docstring states where the reporting is coarser than the diagram. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
210 lines
8.3 KiB
Python
Executable File
210 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Stream the triage flow as it happens, in the stages of the flow chart.
|
|
|
|
Run in a second terminal alongside the demo. Every stage name below is a
|
|
subgraph in `mermaid/TestAutomation.mmd`, and the route line quotes that
|
|
chart's own edge labels, so an audience can follow along on the diagram:
|
|
|
|
Detect and gather -> Hermes analysis -> Ariadne policy gates
|
|
-> Ariadne response -> Inspectable outputs
|
|
|
|
Each stage prints the evidence that moved it, because the claim the chart
|
|
makes is that every step is answerable from data rather than asserted.
|
|
|
|
Two honest limits. The chart draws finer nodes inside each subgraph - the
|
|
collector, the response check, the marker check, the authorizer - and this
|
|
reports at subgraph granularity, not per node. And it follows one incident on
|
|
one job at a time, whereas the chart describes the system as a whole.
|
|
|
|
Read-only: it polls Kubernetes and changes nothing.
|
|
"""
|
|
|
|
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"
|
|
POLL_SECONDS = 4
|
|
|
|
STAGES = [
|
|
("detect", "Detect and gather", "Jenkins failure detector opens an incident"),
|
|
("evidence", "Detect and gather", "Ariadne collects the bounded evidence bundle"),
|
|
("hermes", "Hermes analysis", "Hermes returns a classification it cannot act on"),
|
|
("gates", "Ariadne policy gates", "Ariadne authorizes or refuses, on its own reading"),
|
|
("route", "Ariadne response", "Which branch off the route diamond was taken"),
|
|
("response", "Ariadne response", "Ariadne executes, proposes, or escalates"),
|
|
("verify", "Ariadne response", "One rebuild decides whether the repair worked"),
|
|
("outputs", "Inspectable outputs", "Incident closed; artifacts left behind"),
|
|
]
|
|
|
|
BOLD, DIM, GREEN, YELLOW, RED, RESET = (
|
|
"\033[1m", "\033[2m", "\033[32m", "\033[33m", "\033[31m", "\033[0m"
|
|
)
|
|
|
|
|
|
def sh(*args: str, timeout: int = 20) -> str:
|
|
"""Run a command and return stdout, or "" on any failure."""
|
|
|
|
try:
|
|
out = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
|
|
return out.stdout.strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def stamp() -> str:
|
|
return datetime.now(timezone.utc).strftime("%H:%M:%S")
|
|
|
|
|
|
def fixture_state() -> str:
|
|
return sh("kubectl", "-n", NS_DEMO, "get", "cm", "hermes-triage-demo-fixture",
|
|
"-o", "jsonpath={.data.state}")
|
|
|
|
|
|
def ariadne_lines(tail: int = 400) -> list[dict]:
|
|
"""Return recent Ariadne log records that parse as JSON."""
|
|
|
|
raw = sh("kubectl", "-n", NS_ARIADNE, "logs", "deploy/ariadne", "-c", "ariadne",
|
|
f"--tail={tail}", timeout=25)
|
|
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_state(records: list[dict]) -> dict:
|
|
"""Pull this job's latest auto-triage tick result out of the log."""
|
|
|
|
for record in reversed(records):
|
|
if record.get("event") == "hermes_autotriage" and record.get("jobs"):
|
|
try:
|
|
return json.loads(record["jobs"]).get(JOB) or {}
|
|
except ValueError:
|
|
return {}
|
|
return {}
|
|
|
|
|
|
class Monitor:
|
|
"""Track which flow-chart stage the current incident has reached."""
|
|
|
|
def __init__(self) -> None:
|
|
self.done: dict[str, str] = {}
|
|
self.incident = ""
|
|
self.last_line = ""
|
|
self.summarised = False
|
|
|
|
def mark(self, key: str, evidence: str) -> None:
|
|
"""Record a stage as reached and print it once, with its evidence."""
|
|
|
|
if key in self.done:
|
|
return
|
|
self.done[key] = evidence
|
|
_, subgraph, meaning = next(s for s in STAGES if s[0] == key)
|
|
print(f"{GREEN}{stamp()} ✓ {BOLD}{subgraph}{RESET}{GREEN} — {meaning}{RESET}")
|
|
print(f" {DIM}evidence:{RESET} {evidence}")
|
|
sys.stdout.flush()
|
|
|
|
def checklist(self) -> None:
|
|
"""Print the stage checklist as it currently stands."""
|
|
|
|
print(f"\n{BOLD} flow chart progress{RESET}")
|
|
for key, subgraph, _meaning in STAGES:
|
|
mark = f"{GREEN}✓{RESET}" if key in self.done else f"{DIM}·{RESET}"
|
|
label = subgraph if key in self.done else f"{DIM}{subgraph}{RESET}"
|
|
print(f" {mark} {key:<9} {label}")
|
|
print()
|
|
sys.stdout.flush()
|
|
|
|
def reset_for(self, incident: str) -> None:
|
|
if incident and incident != self.incident:
|
|
self.incident = incident
|
|
self.done.clear()
|
|
self.summarised = False
|
|
print(f"\n{BOLD}{YELLOW}{stamp()} ── incident {incident} ──{RESET}")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def main() -> None:
|
|
print(f"{BOLD}Watching {JOB}. Stages follow mermaid/TestAutomation.mmd.{RESET}")
|
|
print(f"{DIM}Read-only. Ctrl-C to stop.{RESET}\n")
|
|
monitor = Monitor()
|
|
while True:
|
|
records = ariadne_lines()
|
|
state = job_state(records)
|
|
status = str(state.get("status") or "")
|
|
incident = str(state.get("incident_id") or "")
|
|
monitor.reset_for(incident)
|
|
|
|
if incident and status in {"human_required", "awaiting_rebuild", "deduped", "failed"}:
|
|
monitor.mark("detect", f"incident {incident} opened from a terminal build failure")
|
|
|
|
for record in records:
|
|
if record.get("logger", "").endswith("hermes_autotriage") and record.get("status") == "ok":
|
|
continue
|
|
msg = str(record.get("message") or "")
|
|
if "run_triage" in msg or "hermes run" in msg.lower():
|
|
monitor.mark("hermes", msg[:110])
|
|
if record.get("event") == "hermes_code_flow":
|
|
monitor.mark("route", "code_path (optional source proposal) also taken")
|
|
monitor.mark("response", f"code proposal: {record.get('status')} -> gitea_pr")
|
|
|
|
if status == "awaiting_rebuild":
|
|
monitor.mark("evidence", "bundle collected; console regions, tests and logs attached")
|
|
monitor.mark("hermes", "diagnosis returned and parsed against the frozen schema")
|
|
monitor.mark("gates", "every gate passed; Ariadne authorized the predefined action")
|
|
# The chart labels this edge "authorized action"; naming it lets a
|
|
# viewer point at the branch being taken rather than infer it.
|
|
monitor.mark("route", "route --|authorized action|--> registry -> operational_result")
|
|
monitor.mark(
|
|
"response",
|
|
f"{state.get('repair', 'action')} on {state.get('target', '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', 'human_required')} — no action taken",
|
|
)
|
|
monitor.mark("route", "route --|human required|--> human_required -> gitea_issue")
|
|
monitor.mark("response", "escalated to a human; 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()
|
|
# The resolving tick stays newest until another incident opens, so
|
|
# the completed checklist is printed once rather than every poll.
|
|
monitor.summarised = True
|
|
|
|
line = f"{status}|{incident}|{fixture_state()}"
|
|
if line != monitor.last_line:
|
|
monitor.last_line = line
|
|
fixture = fixture_state()
|
|
colour = GREEN if fixture == "healthy" else RED
|
|
print(
|
|
f"{DIM}{stamp()}{RESET} tick: {status or 'idle':<16} "
|
|
f"fixture={colour}{fixture or '?'}{RESET} {DIM}{incident}{RESET}"
|
|
)
|
|
sys.stdout.flush()
|
|
time.sleep(POLL_SECONDS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
print("\nstopped")
|