#!/usr/bin/env python3
"""o7infra_agent - Oracle7-infrastructure: Oracle7 pointed at a system (Ben 2026-10-12: "it gets pointed at anything and it can
become the sensors and run it intelligently"; "we want to be infrastructure"). One file, Python standard library only (3.10+).

  DISCOVER   with no configuration: systemd services (long-running) and jobs (oneshot / timer-run), nginx sites and the TLS
             certificates they serve, listening ports, containers (if docker is present)
  SENSE      host (load, cpu, memory, swap, processes), every real disk (space, inodes), network, every service (state,
             memory, cpu, RESTART RATE), every job (last run ok), journal errors per unit, certificate DAYS LEFT, site
             reachability and latency; plus any source in /etc/o7infra/sources.json (http_json / prometheus / command).
             Every value is a FACT (time, entity, signal, value, source) - the receipt says where it was read.
  STORE      SQLite with a HARD SIZE CAP: raw 24 h, 10-minute means 30 days, pruned every cycle.
  READ       readers generated for every signal: current, trend, time-to-threshold, anomaly vs its own history, what changed
             before an event. `ask` answers plain questions from them.
  THE GATE   = THE FUTURE. Forecasts are logged when made and scored when their time arrives; a method is ADMITTED for a kind
             of signal only after >= 30 scored forecasts beat "stays the same" by >= 10%.
  ACT        actions are CHECKED OPERATIONS: precondition, predicted effect, dry run. Ladder: recommend -> dry-run -> a human
             second key (`approve <id>`) -> autonomous ONLY for an action class whose observed effect matched its prediction
             5 times, and only when O7INFRA_AUTONOMOUS=1. Default: recommend + dry-run, nothing changes the system.

usage: o7infra_agent.py run | brief | status | forecast | gate | why <entity> | signals [f] | discover | plan
                        | approve <id> | execute <id> | ask "<question>" """
import glob
import json
import os
import re
import socket
import sqlite3
import ssl
import subprocess
import sys
import time
import urllib.request

HOME = os.environ.get("O7INFRA_HOME", "/var/lib/o7infra")
DB = os.path.join(HOME, "facts.sqlite3")
SOURCES = os.environ.get("O7INFRA_SOURCES", "/etc/o7infra/sources.json")
STEP = int(os.environ.get("O7INFRA_STEP", "60"))
CAP_MB = float(os.environ.get("O7INFRA_CAP_MB", "60"))
AUTONOMOUS = os.environ.get("O7INFRA_AUTONOMOUS") == "1"
RAW_KEEP, AGG_KEEP = 24 * 3600, 30 * 86400
FORECAST_EVERY, SLOW_EVERY, DISCOVER_EVERY = 600, 300, 3600
HORIZONS = (3600, 6 * 3600)
REAL_FS = {"ext2", "ext3", "ext4", "xfs", "btrfs", "zfs", "f2fs", "vfat", "ntfs"}
# (warn, critical); "low" = lower is worse
THRESH = {"used_pct": (90.0, 97.0, "high"), "swap_used_pct": (80.0, 95.0, "high"), "mem_used_pct": (90.0, 97.0, "high"),
          "inode_pct": (85.0, 95.0, "high"), "cert_days_left": (14.0, 5.0, "low"), "restarts_per_h": (6.0, 30.0, "high"),
          "up": (0.5, 0.5, "low"), "last_run_ok": (0.5, 0.5, "low")}


# ------------------------------------------------------------------------------------------------------------------ store
def db():
    os.makedirs(HOME, exist_ok=True)
    c = sqlite3.connect(DB, timeout=30)
    c.executescript("""
    PRAGMA journal_mode=WAL;
    CREATE TABLE IF NOT EXISTS fact (ts INTEGER, entity TEXT, signal TEXT, value REAL, source TEXT);
    CREATE INDEX IF NOT EXISTS fact_es ON fact(entity, signal, ts);
    CREATE INDEX IF NOT EXISTS fact_ts ON fact(ts);
    CREATE TABLE IF NOT EXISTS agg (ts INTEGER, entity TEXT, signal TEXT, value REAL);
    CREATE INDEX IF NOT EXISTS agg_es ON agg(entity, signal, ts);
    CREATE TABLE IF NOT EXISTS event (ts INTEGER, entity TEXT, kind TEXT, detail TEXT);
    CREATE TABLE IF NOT EXISTS forecast (made INTEGER, target INTEGER, entity TEXT, signal TEXT, method TEXT, value REAL,
                                         scored INTEGER DEFAULT 0, actual REAL);
    CREATE INDEX IF NOT EXISTS forecast_t ON forecast(scored, target);
    CREATE TABLE IF NOT EXISTS action (id INTEGER PRIMARY KEY, made INTEGER, entity TEXT, kind TEXT, command TEXT,
                                       precondition TEXT, predicted TEXT, dry_run TEXT, status TEXT, approved INTEGER,
                                       executed INTEGER, executed_at INTEGER, observed TEXT, matched INTEGER);
    CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT);
    """)
    return c


def meta_get(c, k, d=None):
    r = c.execute("SELECT v FROM meta WHERE k=?", (k,)).fetchone()
    return json.loads(r[0]) if r else d


def meta_set(c, k, v):
    c.execute("INSERT OR REPLACE INTO meta VALUES (?,?)", (k, json.dumps(v)))


def _read(p):
    try:
        with open(p) as f:
            return f.read()
    except OSError:
        return ""


def _run(cmd, timeout=20):
    try:
        return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout
    except (OSError, subprocess.TimeoutExpired):
        return ""


# --------------------------------------------------------------------------------------------------------------- discover
def discover():
    """what is on this system, found with no configuration"""
    d = {"services": [], "jobs": [], "sites": [], "certs": {}, "ports": [], "containers": []}
    names = [l.split()[0] for l in _run(["systemctl", "list-units", "--type=service", "--all", "--no-legend", "--plain"]).splitlines()
             if l.split() and l.split()[0].endswith(".service")]
    timers = {l.split()[-2] if len(l.split()) >= 2 else "" for l in _run(["systemctl", "list-timers", "--all", "--no-legend", "--plain"]).splitlines()}
    if names:
        txt = _run(["systemctl", "show", "-p", "Id,Type,ActiveState,LoadState,TriggeredBy"] + names, timeout=40)
        for block in txt.strip().split("\n\n"):
            kv = dict(l.split("=", 1) for l in block.splitlines() if "=" in l)
            if kv.get("LoadState") != "loaded" or kv.get("ActiveState") not in ("active", "failed", "activating", "deactivating"):
                continue
            uid = kv.get("Id", "")
            if uid.startswith(("user@", "user-runtime-dir@")):
                continue                                   # per-user session managers are not the system's services
            job = kv.get("Type") == "oneshot" or ".timer" in kv.get("TriggeredBy", "") or uid in timers
            (d["jobs"] if job else d["services"]).append(uid[:-8])
    for conf in glob.glob("/etc/nginx/sites-enabled/*") + glob.glob("/etc/nginx/conf.d/*.conf"):
        txt = _read(conf)
        for block in re.split(r"\bserver\s*\{", txt)[1:]:
            names_ = re.findall(r"\bserver_name\s+([^;]+);", block)
            cert = re.findall(r"\bssl_certificate\s+([^;]+);", block)
            for n in (names_[0].split() if names_ else []):
                if n not in ("_", "localhost") and "." in n and not n.startswith("*") and not re.fullmatch(r"[\d.]+", n):
                    if n not in d["sites"]:
                        d["sites"].append(n)
                    if cert:
                        d["certs"].setdefault(cert[0].strip(), []).append(n)
    ss = _run(["ss", "-ltnH"])
    for line in ss.splitlines():
        p = line.split()
        if len(p) >= 4:
            port = p[3].rsplit(":", 1)[-1]
            if port.isdigit() and port not in d["ports"]:
                d["ports"].append(port)
    if os.path.exists("/usr/bin/docker"):
        d["containers"] = [l.strip() for l in _run(["docker", "ps", "--format", "{{.Names}}"]).splitlines() if l.strip()]
    return d


# ---------------------------------------------------------------------------------------------------------------- sensors
def sense_host(prev):
    out = []
    la = _read("/proc/loadavg").split()
    if la:
        out += [("host", "load1", float(la[0]), "/proc/loadavg"), ("host", "load5", float(la[1]), "/proc/loadavg")]
    st = _read("/proc/stat").splitlines()
    if st:
        f = [int(x) for x in st[0].split()[1:]]
        idle, total = f[3] + (f[4] if len(f) > 4 else 0), sum(f)
        if "cpu" in prev and total > prev["cpu"][1]:
            out.append(("host", "cpu_pct", round(100.0 * (1 - (idle - prev["cpu"][0]) / (total - prev["cpu"][1])), 2), "/proc/stat"))
        prev["cpu"] = (idle, total)
    mi = {}
    for line in _read("/proc/meminfo").splitlines():
        k, _, v = line.partition(":")
        mi[k] = int(v.split()[0]) if v.split() else 0
    if mi.get("MemTotal"):
        avail = mi.get("MemAvailable", 0)
        out += [("host", "mem_used_pct", round(100.0 * (1 - avail / mi["MemTotal"]), 2), "/proc/meminfo"),
                ("host", "mem_avail_mb", round(avail / 1024, 1), "/proc/meminfo")]
        if mi.get("SwapTotal"):
            out.append(("host", "swap_used_pct", round(100.0 * (1 - mi.get("SwapFree", 0) / mi["SwapTotal"]), 2), "/proc/meminfo"))
    out.append(("host", "procs", float(len([d for d in os.listdir("/proc") if d.isdigit()])), "/proc"))
    return out


def sense_disks():
    out, seen = [], set()
    for line in _read("/proc/mounts").splitlines():
        p = line.split()
        if len(p) < 3 or p[2] not in REAL_FS or p[0] in seen:
            continue
        seen.add(p[0])
        try:
            s = os.statvfs(p[1])
        except OSError:
            continue
        if not s.f_blocks:
            continue
        ent = "disk:" + p[1]
        out += [(ent, "used_pct", round(100.0 * (1 - s.f_bavail / s.f_blocks), 2), "statvfs " + p[1]),
                (ent, "free_gb", round(s.f_bavail * s.f_frsize / 1e9, 3), "statvfs " + p[1])]
        if s.f_files:
            out.append((ent, "inode_pct", round(100.0 * (1 - s.f_favail / s.f_files), 2), "statvfs " + p[1]))
    return out


def sense_net(prev, dt):
    out = []
    for line in _read("/proc/net/dev").splitlines()[2:]:
        name, _, rest = line.partition(":")
        name = name.strip()
        if name == "lo" or not rest.split():
            continue
        f = rest.split()
        rx, tx = int(f[0]), int(f[8])
        key = "net:" + name
        if key in prev and dt > 0:
            out += [(key, "rx_kbps", round((rx - prev[key][0]) / dt / 1024, 2), "/proc/net/dev"),
                    (key, "tx_kbps", round((tx - prev[key][1]) / dt / 1024, 2), "/proc/net/dev")]
        prev[key] = (rx, tx)
    return out


def sense_units(prev, dt, disc):
    out, events = [], []
    jobs = set(disc.get("jobs", []))
    names = [n + ".service" for n in disc.get("services", []) + disc.get("jobs", [])]
    if not names:
        return out, events
    txt = _run(["systemctl", "show", "-p", "Id,ActiveState,NRestarts,MemoryCurrent,CPUUsageNSec,Result"] + names, timeout=40)
    now = time.time()
    for block in txt.strip().split("\n\n"):
        kv = dict(l.split("=", 1) for l in block.splitlines() if "=" in l)
        uid = kv.get("Id", "")[:-8]
        if not uid:
            continue
        ent = ("job:" if uid in jobs else "unit:") + uid
        state = kv.get("ActiveState", "")
        if uid in jobs:
            ok = 0.0 if (state == "failed" or kv.get("Result", "success") not in ("success", "")) else 1.0
            out.append((ent, "last_run_ok", ok, "systemctl show (Result=%s)" % kv.get("Result", "")))
        else:
            out.append((ent, "up", 1.0 if state == "active" else 0.0, "systemctl show (ActiveState=%s)" % state))
        last = prev.get("state:" + ent)
        if last and last != state:
            events.append((ent, "state", "%s -> %s" % (last, state)))
        prev["state:" + ent] = state
        try:
            r = int(kv.get("NRestarts", "0"))
            hist = prev.setdefault("rh:" + ent, [])
            hist.append((now, r))
            while hist and hist[0][0] < now - 3600:
                hist.pop(0)
            if len(hist) >= 2 and hist[-1][0] > hist[0][0]:
                rate = (hist[-1][1] - hist[0][1]) / (hist[-1][0] - hist[0][0]) * 3600
                out.append((ent, "restarts_per_h", round(rate, 2), "systemctl show NRestarts=%d" % r))
            if "restarts:" + ent in prev and r > prev["restarts:" + ent] and r - prev["restarts:" + ent] < 1000:
                events.append((ent, "restart", "NRestarts %d -> %d" % (prev["restarts:" + ent], r)))
            prev["restarts:" + ent] = r
        except ValueError:
            pass
        try:
            mem = int(kv.get("MemoryCurrent", ""))
            if mem < 1 << 60:
                out.append((ent, "mem_mb", round(mem / 1048576, 1), "systemctl show"))
        except ValueError:
            pass
        try:
            cpu = int(kv.get("CPUUsageNSec", ""))
            if cpu < 1 << 62 and ("cpu:" + ent) in prev and dt > 0:
                out.append((ent, "cpu_pct", round(100.0 * (cpu - prev["cpu:" + ent]) / 1e9 / dt, 2), "systemctl show"))
            if cpu < 1 << 62:
                prev["cpu:" + ent] = cpu
        except ValueError:
            pass
    return out, events


def sense_journal(since):
    counts = {}
    for line in _run(["journalctl", "--since", "@%d" % since, "-p", "3", "-o", "json", "--no-pager", "-q"], timeout=30).splitlines():
        try:
            j = json.loads(line)
        except ValueError:
            continue
        u = j.get("_SYSTEMD_UNIT") or j.get("SYSLOG_IDENTIFIER") or "kernel"
        counts[u] = counts.get(u, 0) + 1
    out = [("unit:" + (u[:-8] if u.endswith(".service") else u), "errors", float(n), "journalctl -p 3") for u, n in counts.items()]
    out.append(("host", "errors", float(sum(counts.values())), "journalctl -p 3"))
    return out


def sense_certs(disc):
    out = []
    now = time.time()
    for path, sites in disc.get("certs", {}).items():
        txt = _run(["openssl", "x509", "-enddate", "-noout", "-in", path])
        m = re.search(r"notAfter=(.+)", txt)
        if not m:
            continue
        try:
            end = time.mktime(time.strptime(m.group(1).strip().replace(" GMT", ""), "%b %d %H:%M:%S %Y")) - time.timezone
        except ValueError:
            continue
        out.append(("cert:" + sites[0], "cert_days_left", round((end - now) / 86400, 2), "openssl x509 %s" % path))
    return out


def sense_sites(disc):
    out = []
    ctx = ssl.create_default_context()
    for name in disc.get("sites", [])[:60]:
        t0 = time.time()
        code = 0
        for method in ("HEAD", "GET"):                    # a server that does not implement HEAD (405 / 501) is asked by GET
            try:
                req = urllib.request.Request("https://%s/" % name, method=method, headers={"User-Agent": "o7infra/2"})
                with urllib.request.urlopen(req, timeout=8, context=ctx) as r:
                    code = r.status
            except urllib.error.HTTPError as e:
                code = e.code
            except Exception:  # noqa: BLE001 - unreachable, TLS failure, timeout
                code = 0
            if code not in (405, 501):
                break
        ms = (time.time() - t0) * 1000
        up = 1.0 if 0 < code < 500 else 0.0
        out += [("site:" + name, "up", up, "HTTPS HEAD status %d" % code), ("site:" + name, "latency_ms", round(ms, 1), "HTTPS HEAD")]
    return out


def sense_sources():
    """any system: /etc/o7infra/sources.json = [{"type": "http_json", "url", "entity", "signals": {"name": "a.b.c"}},
    {"type": "prometheus", "url", "entity", "match": "regex"}, {"type": "command", "cmd": [...], "entity", "regex": "name=(value)"}]"""
    out = []
    try:
        srcs = json.load(open(SOURCES))
    except (OSError, ValueError):
        return out
    for s in srcs:
        ent = s.get("entity", "source")
        try:
            if s["type"] == "http_json":
                d = json.loads(urllib.request.urlopen(s["url"], timeout=8).read())
                for name, path in s.get("signals", {}).items():
                    v = d
                    for k in path.split("."):
                        v = v[int(k)] if isinstance(v, list) else v[k]
                    out.append((ent, name, float(v), "GET %s %s" % (s["url"], path)))
            elif s["type"] == "prometheus":
                rx = re.compile(s.get("match", ".*"))
                for line in urllib.request.urlopen(s["url"], timeout=8).read().decode().splitlines():
                    if line.startswith("#") or not line.strip():
                        continue
                    name, _, val = line.rpartition(" ")
                    if rx.search(name):
                        out.append((ent, name[:120], float(val), "GET %s" % s["url"]))
            elif s["type"] == "command":
                txt = _run(s["cmd"], timeout=20)
                for name, val in re.findall(s.get("regex", r"(\w+)=([-\d.]+)"), txt):
                    out.append((ent, name, float(val), "command %s" % " ".join(s["cmd"])[:80]))
        except Exception:  # noqa: BLE001 - one broken source never stops the others
            out.append((ent, "source_ok", 0.0, "source %s" % s.get("url", s.get("cmd", "?"))))
    return out


# --------------------------------------------------------------------------------------------------------------- readers
def series(c, entity, signal, since):
    rows = c.execute("SELECT ts, value FROM fact WHERE entity=? AND signal=? AND ts>=? ORDER BY ts", (entity, signal, since)).fetchall()
    if not rows:
        rows = c.execute("SELECT ts, value FROM agg WHERE entity=? AND signal=? AND ts>=? ORDER BY ts", (entity, signal, since)).fetchall()
    return rows


def slope(rows):
    if len(rows) < 6:
        return None
    pts = rows[:: max(1, len(rows) // 40)]
    ss = sorted((b[1] - a[1]) / (b[0] - a[0]) for i, a in enumerate(pts) for b in pts[i + 1:] if b[0] > a[0])
    return ss[len(ss) // 2] if ss else None


def thresholds(signal):
    for k in sorted(THRESH, key=len, reverse=True):
        if signal == k or signal.endswith(k):
            return THRESH[k]
    return None


def level(signal, v):
    th = thresholds(signal)
    if not th:
        return None
    w, cr, d = th
    if d == "high":
        return "critical" if v >= cr else ("warning" if v >= w else None)
    return "critical" if v < cr or (signal in ("up", "last_run_ok") and v < 0.5) else ("warning" if v < w else None)


# ------------------------------------------------------------------------------------------------- forecasting + the gate
def make_forecasts(c, now):
    for ent, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=? AND (signal LIKE '%used_pct' OR signal LIKE '%inode_pct' "
                            "OR signal IN ('mem_mb','mem_avail_mb','free_gb','cert_days_left'))", (now - 3600,)).fetchall():
        rows = series(c, ent, s, now - 6 * 3600)
        if len(rows) < 6:
            continue
        last = rows[-1][1]
        # a trend is extrapolated only from >= 3 h of history (from less it measured worse than "nothing changes" everywhere)
        sl = slope(rows) if rows[-1][0] - rows[0][0] >= 3 * 3600 else None
        for h in HORIZONS:
            c.execute("INSERT INTO forecast (made, target, entity, signal, method, value) VALUES (?,?,?,?,?,?)", (now, now + h, ent, s, "naive", last))
            if sl is not None:
                for m, k in METHODS.items():
                    c.execute("INSERT INTO forecast (made, target, entity, signal, method, value) VALUES (?,?,?,?,?,?)",
                              (now, now + h, ent, s, m, last + k * sl * h))


def score_forecasts(c, now):
    for rid, target, ent, s in c.execute("SELECT rowid, target, entity, signal FROM forecast WHERE scored=0 AND target<=?", (now,)).fetchall():
        r = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts BETWEEN ? AND ? ORDER BY ABS(ts-?) LIMIT 1",
                      (ent, s, target - STEP * 3, target + STEP * 3, target)).fetchone()
        if r:
            c.execute("UPDATE forecast SET scored=1, actual=? WHERE rowid=?", (r[0], rid))
        elif target < now - 3600:
            c.execute("UPDATE forecast SET scored=2 WHERE rowid=?", (rid,))


METHODS = {"trend6h": 1.0, "damped": 0.5}       # candidate methods; the gate admits whichever beats "nothing changes"


def kind_of(signal):
    return signal                                   # each signal keeps its own record (disk is not judged with memory)


def gate(c):
    acc = {}
    for s, h, m, e in c.execute("SELECT signal, target-made, method, ABS(value-actual) FROM forecast WHERE scored=1").fetchall():
        acc.setdefault((kind_of(s), int(h)), {}).setdefault(m, []).append(e)
    out = {}
    for k, ms in acc.items():
        nv = ms.get("naive", [])
        if not nv:
            continue
        mn = sum(nv) / len(nv)
        rec = {"n_naive": len(nv), "mae_naive": round(mn, 4), "methods": {}, "admitted": None}
        for m in METHODS:
            tr = ms.get(m, [])
            if tr:
                mt = sum(tr) / len(tr)
                ok = len(tr) >= 30 and mt <= 0.9 * mn
                rec["methods"][m] = {"n": len(tr), "mae": round(mt, 4), "admitted": ok}
                if ok and (rec["admitted"] is None or mt < rec["methods"][rec["admitted"]]["mae"]):
                    rec["admitted"] = m
        out["%s@%dh" % (k[0], k[1] // 3600)] = rec
    return out


# ---------------------------------------------------------------------------------------------------------------- actions
SAFE_DISK = [   # (kind, command, dry-run command reporting reclaimable MB)
    ("journal_vacuum", ["journalctl", "--vacuum-size=200M"], ["journalctl", "--disk-usage"]),
    ("apt_clean", ["apt-get", "clean"], ["du", "-sm", "/var/cache/apt/archives"]),
]


def plan_actions(c, now):
    """CHECKED OPERATIONS proposed from the current facts; dry-run results attached; nothing executes here"""
    t = c.execute("SELECT MAX(ts) FROM fact").fetchone()[0]
    if not t:
        return []
    _, rows = latest_values(c)
    rows = [(e, s, v) for e, s, v, _, _ in rows]
    loops = {e: v for e, s, v in rows if s == "restarts_per_h"}
    existing = {r[0] for r in c.execute("SELECT entity FROM action WHERE status IN ('proposed','approved')").fetchall()}
    proposals = []
    for e, s, v in rows:
        lv = level(s, v)
        if not lv or e in existing:                    # one open proposal per entity; dry runs only for new ones
            continue
        if e.startswith("disk:") and s == "used_pct":
            for kind, cmd, dry in SAFE_DISK:
                out = _run(dry, timeout=30).strip().splitlines()
                proposals.append((e, kind, " ".join(cmd), "%s used %.1f%% (%s)" % (e, v, lv),
                                  "free_gb rises by the reclaimable amount; used_pct falls", out[-1][:160] if out else "no output"))
        elif e.startswith("unit:") and s == "restarts_per_h" and lv == "critical":
            u = e[5:]
            tail = _run(["journalctl", "-u", u, "-n", "3", "--no-pager", "-q", "-o", "cat"], timeout=15).strip().replace("\n", " | ")
            proposals.append((e, "investigate_crash_loop", "systemctl status %s; journalctl -u %s -n 50" % (u, u),
                              "%s restarting %.0f times per hour" % (u, v), "read-only: shows the failing cause", tail[:200]))
        elif e.startswith("unit:") and s == "up" and v == 0 and loops.get(e, 0) < 6:
            u = e[5:]                                      # a crash-looping service is investigated, never just restarted
            proposals.append((e, "restart_service", "systemctl restart %s" % u, "%s is not active" % u,
                              "up becomes 1 within 60 s and restarts_per_h stays < 6", "dry run: unit file %s" %
                              ("present" if _run(["systemctl", "cat", u + ".service"], timeout=10) else "missing")))
        elif e.startswith("cert:") and s == "cert_days_left":
            proposals.append((e, "renew_certificate", "certbot renew --cert-name %s" % e[5:], "%s expires in %.1f days" % (e[5:], v),
                              "cert_days_left jumps to ~89", (_run(["certbot", "renew", "--dry-run", "--cert-name", e[5:]], timeout=120) or "certbot not runnable")[-200:]))
        elif e.startswith("job:") and s == "last_run_ok" and v == 0:
            u = e[4:]
            tail = _run(["journalctl", "-u", u, "-n", "3", "--no-pager", "-q", "-o", "cat"], timeout=15).strip().replace("\n", " | ")
            proposals.append((e, "investigate_job", "journalctl -u %s -n 50" % u, "job %s last run failed" % u, "read-only", tail[:200]))
    for e, kind, cmd, pre, pred, dry in proposals:
        c.execute("INSERT INTO action (made, entity, kind, command, precondition, predicted, dry_run, status, approved, executed) "
                  "VALUES (?,?,?,?,?,?,?, 'proposed', 0, 0)", (now, e, kind, cmd, pre, pred, dry))
    return proposals


EFFECT = {   # action kind -> (signal, test on (before, after)) - what "the effect matched the prediction" means
    "restart_service": ("up", lambda b, a: a == 1.0),
    "journal_vacuum": ("free_gb", lambda b, a: b is not None and a > b),
    "apt_clean": ("free_gb", lambda b, a: b is not None and a >= b),
    "renew_certificate": ("cert_days_left", lambda b, a: a > 30),
}


def verify_actions(c, now):
    """an executed action's predicted effect is checked against the readings that follow it (>= 2 minutes later)"""
    for aid, ent, kind, exe, obs in c.execute("SELECT id, entity, kind, executed_at, observed FROM action "
                                              "WHERE executed=1 AND matched IS NULL").fetchall():
        if kind not in EFFECT or not exe or now - exe < 120:
            continue
        sig, test = EFFECT[kind]
        b = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts<? ORDER BY ts DESC LIMIT 1", (ent, sig, exe)).fetchone()
        a = c.execute("SELECT value FROM fact WHERE entity=? AND signal=? AND ts>? ORDER BY ts DESC LIMIT 1", (ent, sig, exe + 60)).fetchone()
        if a is None:
            continue
        ok = bool(test(b[0] if b else None, a[0]))
        c.execute("UPDATE action SET observed=?, matched=? WHERE id=?",
                  (json.dumps({"signal": sig, "before": b[0] if b else None, "after": a[0]}), 1 if ok else 0, aid))


def admitted_actions(c):
    """an action CLASS is admitted for autonomy after 5 executions whose observed effect matched the prediction, none failed"""
    out = {}
    for kind, n, ok in c.execute("SELECT kind, COUNT(*), SUM(matched) FROM action WHERE executed=1 GROUP BY kind").fetchall():
        out[kind] = {"executed": n, "matched": ok or 0, "admitted": n >= 5 and (ok or 0) == n}
    return out


# -------------------------------------------------------------------------------------------------------------- retention
def retain(c, now):
    cut = now - RAW_KEEP
    last_agg = meta_get(c, "last_agg", 0)
    for b in range(max(last_agg, cut - RAW_KEEP) // 600 * 600, cut // 600 * 600, 600):
        c.execute("INSERT INTO agg SELECT ?, entity, signal, AVG(value) FROM fact WHERE ts>=? AND ts<? GROUP BY entity, signal", (b, b, b + 600))
        meta_set(c, "last_agg", b + 600)
    c.execute("DELETE FROM fact WHERE ts<?", (cut,))
    c.execute("DELETE FROM agg WHERE ts<?", (now - AGG_KEEP,))
    c.execute("DELETE FROM forecast WHERE made<?", (now - AGG_KEEP,))
    c.execute("DELETE FROM event WHERE ts<?", (now - AGG_KEEP,))
    if os.path.exists(DB) and os.path.getsize(DB) / 1e6 > CAP_MB:
        c.execute("DELETE FROM agg WHERE ts < (SELECT MIN(ts) + (MAX(ts)-MIN(ts))/4 FROM agg)")
        c.execute("DELETE FROM fact WHERE ts < (SELECT MIN(ts) + (MAX(ts)-MIN(ts))/4 FROM fact)")
        c.commit()
        c.execute("VACUUM")


# ------------------------------------------------------------------------------------------------------------------- loop
def run():
    c = db()
    prev, last_t = {}, time.time()
    last_fc = last_slow = last_disc = 0
    disc = {}
    slow = []
    while True:
        t0 = time.time()
        now, dt = int(t0), t0 - last_t
        if now - last_disc >= DISCOVER_EVERY or not disc:
            disc = discover()
            meta_set(c, "discovery", disc)
            last_disc = now
        facts = sense_host(prev) + sense_disks() + sense_net(prev, dt)
        uf, ev = sense_units(prev, dt, disc)
        facts += uf + sense_journal(int(last_t))
        if now - last_slow >= SLOW_EVERY:
            slow = sense_certs(disc) + sense_sites(disc) + sense_sources()
            facts += slow
            last_slow = now
        c.executemany("INSERT INTO fact VALUES (?,?,?,?,?)", [(now, e, s, v, src) for e, s, v, src in facts])
        c.executemany("INSERT INTO event VALUES (?,?,?,?)", [(now, e, k, d) for e, k, d in ev])
        if now - last_fc >= FORECAST_EVERY:
            score_forecasts(c, now)
            make_forecasts(c, now)
            plan_actions(c, now)
            verify_actions(c, now)
            retain(c, now)
            last_fc = now
        c.commit()
        last_t = t0
        time.sleep(max(5, STEP - (time.time() - t0)))


# ------------------------------------------------------------------------------------------------------------------ query
def latest_values(c):
    """the newest value of every (entity, signal) in the last 15 minutes (slow sensors run every 5)"""
    t = c.execute("SELECT MAX(ts) FROM fact").fetchone()[0]
    if not t:
        return None, []
    rows = c.execute("SELECT f.entity, f.signal, f.value, f.source, f.ts FROM fact f JOIN (SELECT entity, signal, MAX(ts) m FROM fact "
                     "WHERE ts>=? GROUP BY entity, signal) x ON f.entity=x.entity AND f.signal=x.signal AND f.ts=x.m", (t - 900,)).fetchall()
    return t, rows


def status(c):
    t, rows = latest_values(c)
    if not t:
        return {"facts": 0}
    crit, warn = [], []
    for e, s, v, src, ts in rows:
        lv = level(s, v)
        if lv:
            (crit if lv == "critical" else warn).append({"entity": e, "signal": s, "value": v, "receipt": src})
    ev = c.execute("SELECT ts, entity, kind, detail FROM event WHERE ts>=? ORDER BY ts DESC LIMIT 20", (t - 86400,)).fetchall()
    return {"at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t)), "critical": crit, "warning": warn,
            "events_24h": [{"at": time.strftime("%H:%M", time.gmtime(a)), "entity": b, "kind": k, "detail": d} for a, b, k, d in ev]}


def forecast_view(c):
    now = int(time.time())
    g = gate(c)
    out = []
    for ent, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=? AND (signal LIKE '%used_pct' OR signal LIKE '%inode_pct')",
                            (now - 3600,)).fetchall():
        rows = series(c, ent, s, now - 6 * 3600)
        sl = slope(rows)
        th = thresholds(s)
        if not rows or sl is None or sl <= 0 or not th or rows[-1][1] >= th[1]:
            continue
        rec = g.get("%s@6h" % kind_of(s), {})
        m = rec.get("admitted")
        k = METHODS.get(m, 1.0)
        out.append({"entity": ent, "signal": s, "now": rows[-1][1], "per_hour": round(k * sl * 3600, 4), "reaches": th[1],
                    "in_hours": round((th[1] - rows[-1][1]) / (k * sl) / 3600, 1), "admitted": bool(m), "method": m or "trend6h",
                    "record": rec or "not yet scored"})
    out.sort(key=lambda x: x["in_hours"])
    return out


def why(c, entity):
    t, _ = latest_values(c)
    if not t:
        return []
    out = []
    for e, s in c.execute("SELECT DISTINCT entity, signal FROM fact WHERE ts>=?", (t - 3600,)).fetchall():
        a = [v for _, v in series(c, e, s, t - 3600)]
        if len(a) < 20:
            continue
        x, y = a[:len(a) // 2], a[len(a) // 2:]
        mx, my = sum(x) / len(x), sum(y) / len(y)
        sd = (sum((v - mx) ** 2 for v in x) / len(x)) ** 0.5 or 1e-9
        z = (my - mx) / sd
        if abs(z) >= 3:
            out.append({"entity": e, "signal": s, "before": round(mx, 3), "after": round(my, 3), "shift_sd": round(z, 1)})
    out.sort(key=lambda d: (d["entity"] != entity, -abs(d["shift_sd"])))
    return out[:15]


def ask(c, q):
    """plain questions answered by the generated readers; every answer cites the readers' facts"""
    ql = q.lower()
    if re.search(r"\b(fill|full|space|disk)\b", ql):
        return {"answer": forecast_view(c) or "no disk is trending toward its threshold", "reader": "forecast"}
    if re.search(r"\b(cert|certificate|tls|ssl|expire)", ql):
        _, rows = latest_values(c)
        return {"answer": sorted([{"site": e[5:], "days_left": v, "receipt": src} for e, s, v, src, _ in rows if s == "cert_days_left"],
                                 key=lambda x: x["days_left"]), "reader": "cert_days_left"}
    m = re.search(r"\bwhy\b.*?\b([a-z0-9][\w.:-]+)\??$", ql)
    if m:
        return {"answer": why(c, m.group(1)), "reader": "why"}
    if re.search(r"\b(wrong|broken|fail|down|problem|status|health)", ql):
        return {"answer": status(c), "reader": "status"}
    if re.search(r"\b(do|fix|action|should)\b", ql):
        return {"answer": [dict(zip(("id", "entity", "kind", "command", "precondition", "predicted", "dry_run", "status"), r))
                           for r in c.execute("SELECT id, entity, kind, command, precondition, predicted, dry_run, status FROM action "
                                              "WHERE status='proposed' ORDER BY id DESC LIMIT 20").fetchall()], "reader": "plan"}
    return {"answer": "ask about disks, certificates, failures, why <service>, or what to do", "reader": None}


def main():
    cmd = sys.argv[1] if len(sys.argv) > 1 else "brief"
    if cmd == "run":
        return run()
    c = db()
    if cmd == "status":
        print(json.dumps(status(c), indent=1))
    elif cmd == "forecast":
        print(json.dumps(forecast_view(c), indent=1))
    elif cmd == "gate":
        print(json.dumps({"forecasts": gate(c), "actions": admitted_actions(c)}, indent=1))
    elif cmd == "why":
        print(json.dumps(why(c, sys.argv[2] if len(sys.argv) > 2 else "host"), indent=1))
    elif cmd == "discover":
        d = discover()
        print(json.dumps({k: (v if k != "certs" else {p: s for p, s in v.items()}) for k, v in d.items()}, indent=1))
    elif cmd == "plan":
        plan_actions(c, int(time.time()))
        c.commit()
        for r in c.execute("SELECT id, entity, kind, command, precondition, predicted, dry_run FROM action WHERE status='proposed' ORDER BY id").fetchall():
            print("#%d %s %s\n   do: %s\n   because: %s\n   expect: %s\n   dry run: %s" % r)
    elif cmd == "approve":
        c.execute("UPDATE action SET status='approved', approved=? WHERE id=? AND status='proposed'", (int(time.time()), int(sys.argv[2])))
        c.commit()
        print("approved #%s (executes only with `execute %s`)" % (sys.argv[2], sys.argv[2]))
    elif cmd == "execute":
        r = c.execute("SELECT id, kind, command, status FROM action WHERE id=?", (int(sys.argv[2]),)).fetchone()
        if not r or r[3] != "approved":
            print("refused: action #%s is not approved" % sys.argv[2])
            return 1
        if r[1].startswith("investigate"):
            print(_run(["sh", "-c", r[2]], timeout=60))
        else:
            res = subprocess.run(["sh", "-c", r[2]], capture_output=True, text=True, timeout=600)
            print(res.stdout[-2000:], res.stderr[-1000:])
        c.execute("UPDATE action SET status='executed', executed=1, executed_at=? WHERE id=?", (int(time.time()), r[0]))
        c.commit()
    elif cmd == "ask":
        print(json.dumps(ask(c, " ".join(sys.argv[2:])), indent=1, default=str))
    elif cmd == "signals":
        f = sys.argv[2] if len(sys.argv) > 2 else ""
        _, rows = latest_values(c)
        print(json.dumps([{"entity": e, "signal": s, "value": v, "receipt": src} for e, s, v, src, _ in rows if f in e + s], indent=1))
    elif cmd == "brief":
        s = status(c)
        n = c.execute("SELECT COUNT(*) FROM fact").fetchone()[0]
        disc = meta_get(c, "discovery", {})
        print("%s  at %s  facts %d  store %.1f MB  | discovered: %d services, %d jobs, %d sites, %d certs, %d ports" % (
            socket.gethostname(), s.get("at"), n, os.path.getsize(DB) / 1e6, len(disc.get("services", [])), len(disc.get("jobs", [])),
            len(disc.get("sites", [])), len(disc.get("certs", {})), len(disc.get("ports", []))))
        for x in s.get("critical", []):
            print("  CRITICAL %-34s %-15s %-8s [%s]" % (x["entity"], x["signal"], x["value"], x["receipt"][:60]))
        for x in s.get("warning", [])[:10]:
            print("  warning  %-34s %-15s %-8s [%s]" % (x["entity"], x["signal"], x["value"], x["receipt"][:60]))
        for x in forecast_view(c)[:3]:
            print("  forecast %-34s %-15s %s now, %+.3f/h -> %s in %sh [%s]" % (x["entity"], x["signal"], x["now"], x["per_hour"],
                  x["reaches"], x["in_hours"], "ADMITTED" if x["admitted"] else "unproven"))
        g = gate(c)
        print("  gate: %d signal forecasts scored, %d with an admitted method; actions proposed %d, autonomy %s" % (
            len(g), sum(1 for v in g.values() if v.get("admitted")),
            c.execute("SELECT COUNT(*) FROM action WHERE status='proposed'").fetchone()[0], "ON" if AUTONOMOUS else "off"))
    else:
        print(__doc__)


if __name__ == "__main__":
    main()