my-homelab-configs/scripts/incident-commander

#!/usr/bin/env python3
"""Read-only incident classification and triage for homelab failures."""

from __future__ import annotations

import argparse
import csv
import json
import re
import sys
from dataclasses import dataclass
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
INCIDENT_DIR = REPO_ROOT / "infra" / "incident-commander"
INCIDENTS_FILE = INCIDENT_DIR / "incidents.tsv"
FIXTURES_DIR = INCIDENT_DIR / "fixtures"


@dataclass
class Incident:
    incident_id: str
    title: str
    fixture: str
    expected_class: str
    runbook: str
    next_commands: list[str]
    forbidden_actions: list[str]


RULES: list[tuple[str, str, str]] = [
    ("edge_bad_gateway", r"\b502\b|bad gateway|nginx/1\.31", "Public edge proxy cannot get a healthy upstream response."),
    ("gitea_edge_backend", r"cannot reach gitea backend|100\.85\.138\.30:3000|/git/", "Edge-to-Gitea backend path is broken or Gitea is unhealthy."),
    ("cluster_api_down", r"6443.*refused|api server.*refused|connection to the server .* was refused", "Kubernetes API is not accepting connections."),
    ("dns_failure", r"no servers could be reached|communications error .*#53|timed out", "RPi/Pi-hole DNS path is unavailable or blocked."),
]


def split_items(value: str) -> list[str]:
    return [item.strip() for item in value.split(";") if item.strip()]


def load_incidents() -> list[Incident]:
    with INCIDENTS_FILE.open(encoding="utf-8", newline="") as handle:
        rows = csv.DictReader(handle, delimiter="\t")
        return [
            Incident(
                incident_id=row["id"],
                title=row["title"],
                fixture=row["fixture"],
                expected_class=row["expected_class"],
                runbook=row["runbook"],
                next_commands=split_items(row["next_commands"]),
                forbidden_actions=split_items(row["forbidden_actions"]),
            )
            for row in rows
        ]


def fixture_text(incident: Incident) -> str:
    path = FIXTURES_DIR / incident.fixture
    if not path.is_file():
        raise SystemExit(f"Missing fixture for {incident.incident_id}: {path}")
    return path.read_text(encoding="utf-8")


def classify(text: str) -> tuple[str, str]:
    lowered = text.lower()
    for incident_class, pattern, hypothesis in RULES:
        if re.search(pattern, lowered, re.IGNORECASE | re.DOTALL):
            return incident_class, hypothesis
    return "unknown", "No known incident pattern matched. Preserve evidence and run read-only status checks."


def incident_for_class(incidents: list[Incident], incident_class: str) -> Incident | None:
    for incident in incidents:
        if incident.expected_class == incident_class:
            return incident
    return None


def render_triage(text: str, incidents: list[Incident]) -> dict[str, object]:
    incident_class, hypothesis = classify(text)
    incident = incident_for_class(incidents, incident_class)
    if incident is None:
        return {
            "class": incident_class,
            "hypothesis": hypothesis,
            "runbook": "none",
            "next_commands": ["./jeannie status", "./jeannie scorecard"],
            "forbidden_actions": ["destructive commands", "secret exposure", "unverified manual fixes"],
            "evidence": first_evidence_lines(text),
        }
    return {
        "class": incident_class,
        "hypothesis": hypothesis,
        "runbook": incident.runbook,
        "next_commands": incident.next_commands,
        "forbidden_actions": incident.forbidden_actions,
        "evidence": first_evidence_lines(text),
    }


def first_evidence_lines(text: str) -> list[str]:
    lines = [line.strip() for line in text.splitlines() if line.strip()]
    return lines[:6]


def print_triage(result: dict[str, object]) -> None:
    print("AI Incident Commander")
    print("=====================")
    print("mode: read-only")
    print()
    print(f"Class: {result['class']}")
    print(f"Hypothesis: {result['hypothesis']}")
    print(f"Runbook: {result['runbook']}")
    print()
    print("Evidence:")
    for line in result["evidence"]:
        print(f"  {line}")
    print()
    print("Next read-only commands:")
    for command in result["next_commands"]:
        print(f"  {command}")
    print()
    print("Blocked until explicitly approved:")
    for action in result["forbidden_actions"]:
        print(f"  {action}")


def print_list(incidents: list[Incident]) -> int:
    for incident in incidents:
        print(f"{incident.incident_id}\t{incident.expected_class}\t{incident.title}")
    return 0


def replay(incidents: list[Incident], incident_id: str | None, as_json: bool, details: bool) -> int:
    selected = [incident for incident in incidents if incident_id in (None, incident.incident_id)]
    if not selected:
        print(f"Unknown incident fixture: {incident_id}", file=sys.stderr)
        return 2

    results = []
    for incident in selected:
        text = fixture_text(incident)
        result = render_triage(text, incidents)
        status = "pass" if result["class"] == incident.expected_class else "fail"
        results.append({"id": incident.incident_id, "status": status, "expected": incident.expected_class, **result})

    if as_json:
        print(json.dumps(results, indent=2, sort_keys=True))
    else:
        failures = sum(1 for result in results if result["status"] != "pass")
        print("Incident Replay")
        print("===============")
        print(f"cases={len(results)} failures={failures}")
        print()
        for result in results:
            print(f"{result['status']:5} {result['id']} - {result['class']}")
            print(f"      next: {result['next_commands'][0]}")
            if details:
                print(f"      runbook: {result['runbook']}")
                print(f"      hypothesis: {result['hypothesis']}")
    return 1 if any(result["status"] != "pass" for result in results) else 0


def read_triage_input(args: argparse.Namespace) -> str:
    if args.text:
        return args.text
    if args.from_file:
        return Path(args.from_file).read_text(encoding="utf-8", errors="replace")
    if not sys.stdin.isatty():
        return sys.stdin.read()
    raise SystemExit("Provide --text, --from-file, or stdin for incident triage.")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)
    subparsers.add_parser("list")
    replay_parser = subparsers.add_parser("replay")
    replay_parser.add_argument("incident_id", nargs="?")
    replay_parser.add_argument("--json", action="store_true")
    replay_parser.add_argument("--details", action="store_true")
    triage_parser = subparsers.add_parser("triage")
    triage_parser.add_argument("--from-file")
    triage_parser.add_argument("--text")
    triage_parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    incidents = load_incidents()
    if args.command == "list":
        return print_list(incidents)
    if args.command == "replay":
        return replay(incidents, args.incident_id, args.json, args.details)
    if args.command == "triage":
        result = render_triage(read_triage_input(args), incidents)
        if args.json:
            print(json.dumps(result, indent=2, sort_keys=True))
        else:
            print_triage(result)
        return 0
    return 2


if __name__ == "__main__":
    raise SystemExit(main())