FabledSteward/steward/timeline/catalog.py

"""What can appear on the timeline: each entry type, the kind of event it is,
and the area it sorts under until Settings → Timeline says otherwise.

An area is what part of the infrastructure was affected (Service, Network,
Hardware …); a kind is how the event came about (a change someone made, an
automation run, an alert …). Kinds are fixed here. Areas, and each type's
area and visibility, are settings (steward.timeline.config), so they can be
changed without a release and the change re-sorts entries already written.

A plugin adds its own types with register_type() from its setup().
"""
from __future__ import annotations

from dataclasses import dataclass

from steward.core.settings import DEFAULTS

# How an event came about. Fixed: the page's "Kind" filter and the MCP
# tool's `kind` argument take these keys.
KINDS: dict[str, str] = {
    "change": "Change",          # a person or an agent did it
    "automation": "Automation",  # a playbook or maintenance run
    "alert": "Alert",
    "anomaly": "Anomaly",
    "ticket": "Ticket",
    "observed": "Observed",      # Steward saw it happen: a reboot, a crash-loop
    "steward": "Steward",        # Steward's own settings and access
}

# The starting areas, [key, label]: the timeline.areas setting's default.
# Settings → Timeline renames, adds and removes them.
DEFAULT_AREAS: list[list[str]] = DEFAULTS["timeline.areas"]
# Where an entry goes whose area is unknown: removed without a destination,
# or named by a plugin that is no longer loaded.
OTHER = ("other", "Other")


@dataclass(frozen=True)
class EntryType:
    key: str
    label: str        # how Settings → Timeline names it
    kind: str         # a KINDS key
    area: str         # the area its entries are written with, unless the writer names one
    shown: bool       # whether the timeline shows it until Settings says otherwise
    about: str        # one line for Settings → Timeline


TYPES: dict[str, EntryType] = {}


def register_type(key: str, label: str, kind: str, area: str, shown: bool, about: str) -> None:
    if kind not in KINDS:
        raise ValueError(f"timeline type {key!r}: unknown kind {kind!r}")
    TYPES[key] = EntryType(key, label, kind, area, shown, about)


# The full records an entry can point to: its ref_type and the model whose
# id it names, so the timeline can tell a record still kept from one aged out.
# A plugin adds its own with register_ref() from its setup().
REFS: dict[str, object] = {}


def register_ref(ref_type: str, model) -> None:
    REFS[ref_type] = model


for _t in (
    ("alert.fired", "Alert fired", "alert", "monitoring", True,
     "An alert rule's condition held long enough to fire."),
    ("alert.resolved", "Alert resolved", "alert", "monitoring", True,
     "A firing alert cleared, or its rule was changed."),
    ("alert.acknowledged", "Alert acknowledged", "alert", "monitoring", True,
     "Someone took a firing alert."),
    ("alert.silent", "Readings stopped", "alert", "monitoring", True,
     "What an alert rule watches stopped reporting."),
    ("alert.resumed", "Readings resumed", "alert", "monitoring", True,
     "What an alert rule watches reported again after a silence."),
    ("ansible.changed", "Playbook changed something", "automation", "host_os", True,
     "A playbook run finished and changed at least one host."),
    ("ansible.unchanged", "Playbook changed nothing", "automation", "host_os", False,
     "A playbook run finished with nothing to change."),
    ("ansible.failed", "Playbook failed", "automation", "host_os", True,
     "A playbook run failed, was cancelled, or was stopped by a restart."),
    ("ansible.check", "Playbook dry run", "automation", "host_os", False,
     "A playbook run in check mode: it showed what would change."),
    ("steward.config", "Steward settings changed", "steward", "monitoring", True,
     "Monitors, alert rules, maintenance windows, inventory, plugins, settings."),
    ("steward.access", "Users and access", "steward", "security", True,
     "Users, roles, passwords and AI access tokens."),
    ("steward.signin", "Sign-ins", "steward", "security", False,
     "Someone signed in, or failed to."),
    ("steward.backup", "Steward backups", "steward", "storage", True,
     "Steward backed itself up, or was restored from a backup."),
    # An anomaly's entry changes type as it is reviewed; its area is what the
    # metric is about.
    ("anomaly.to_review", "Anomaly to review", "anomaly", "monitoring", False,
     "A metric left its normal and nobody has reviewed it yet."),
    ("anomaly.real", "Anomaly judged real", "anomaly", "monitoring", True,
     "An anomaly someone reviewed and marked a real problem."),
    ("anomaly.normal", "Anomaly accepted as normal", "anomaly", "monitoring", False,
     "An anomaly someone reviewed and marked normal: its level is learned."),
    ("anomaly.lapsed", "Anomaly not reviewed", "anomaly", "monitoring", False,
     "An anomaly nobody reviewed within the review window."),
):
    register_type(*_t)


# Which area an alert's entry is written with, from what its rule watches.
# A source not listed here is Monitoring.
_SOURCE_AREAS = {
    "icmp": "network", "dns": "network", "tcp": "service", "http": "service",
    "docker": "service", "prometheus": "service",
    "traefik": "network", "unifi": "network",
    "snmp": "hardware",
}
# host_agent metrics by prefix, first match wins; the rest are Host OS.
_HOST_METRIC_AREAS = (
    ("updates_security", "security"),
    ("temp_", "hardware"), ("drive", "hardware"),
    ("disk_", "storage"), ("inodes_", "storage"), ("mounts_", "storage"), ("nfs_", "storage"),
    ("net_", "network"),
)


def area_of_metric(source: str, metric: str) -> str:
    """The area a reading of `metric` from `source` is about."""
    if source == "host_agent":
        for prefix, area in _HOST_METRIC_AREAS:
            if metric.startswith(prefix):
                return area
        return "host_os"
    if source == "snmp" and any(w in metric for w in ("volume", "disk_used", "storage")):
        return "storage"
    return _SOURCE_AREAS.get(source, "monitoring")


# Audit actions and the type each is entered as. The first prefix that
# matches wins; None leaves the action off the timeline, because another
# writer records the event itself (a run, an acknowledgement) or it is about
# one person's own preferences.
_AUDIT_TYPES = (
    ("ansible.run.", None), ("ansible.host_run", None),
    ("alert_rule.acknowledged", None), ("alert_rule.tested", None),
    ("settings.notification.tested", None),
    ("account.time_zone", None), ("account.email", None),
    ("host_agent.maintenance_plan", "steward.config"), ("host_agent.maintenance", None),
    ("docker.prune.", None),
    ("auth.login", "steward.signin"), ("auth.setup", "steward.config"),
    ("account.", "steward.access"), ("user.", "steward.access"), ("api_token.", "steward.access"),
    ("backup.", "steward.backup"),
)


def audit_type(action: str) -> str | None:
    for prefix, type_ in _AUDIT_TYPES:
        if action.startswith(prefix):
            return type_
    return "steward.config"