titan-iac/scripts/ops/hermes_triage_monitor.py

#!/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. Each stage below is a subgraph in
`mermaid/TestAutomation.mmd`, so what prints here can be followed on the chart:

    Detect and gather -> Hermes analysis -> Ariadne policy gates
                      -> Ariadne response -> Inspectable outputs

Every stage prints the evidence that moved it, because the point of the chart
is that each step is answerable from data rather than asserted.

Read-only: it polls Kubernetes, Jenkins and Gitea 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"),
    ("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("response", f"code proposal: {record.get('status')}")

        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")
            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("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")