dashboards/agent/host_agent.py

#!/usr/bin/env python3
"""host_agent.py — read-only host metrics agent for the homelab dashboards.

**Stdlib only, one file, Python 3.10+.** Copy it to a host, run it, and the dashboard can
read that host. It never writes anything anywhere: every number comes from reading
/proc, /sys and statvfs, and the HTTP server only answers GET.

    GET /health      -> {"ok": true, "hostname": ..., "version": ...}
    GET /info        -> slow-changing identity (os, kernel, cpu model, cores, gpu list)
    GET /snapshot    -> everything, one JSON document (see SCHEMA below)
    GET /schema      -> this docstring's SCHEMA section, for self-documenting deploys

Design rules
------------
* **Counters stay counters.** The agent reports cumulative values (jiffies, bytes, ops)
  and the *dashboard* computes rates with a labelled window. A stateless agent cannot
  lie about an interval, survive a restart silently, or be polled twice by two dashboards
  and halve its own rate.
* **Absent is null, never 0.** A host without a GPU, or a sysfs file only root can read,
  yields `null`. The UI renders `n/a`. Turning "unknown" into 0 is the one mistake a
  monitoring agent is not allowed to make.
* **Every read is guarded.** A vanished /proc entry (process exited mid-scan) skips that
  item; the snapshot still succeeds. The poll is degraded, never failed.
* **SMART is a best-effort add-on.** The per-disk `smart` block runs `smartctl -i -A -H`
  in a subprocess: bounded 5 s per device and a pass budget of 20 s scaled to the disk
  count, cached 240 s so a fast dashboard poll cannot re-issue ATA commands every few
  seconds. Refreshing an expired
  cache happens on a background thread while the request is served the previous result —
  20 s of wedged disk must never cost a poll more than its own timeout. No binary, a denied
  device or a wedged disk degrades *that disk's* `smart` to `null` plus one `errors`
  entry (a missing binary produces exactly one entry for the whole pass) and never
  fails the snapshot or `/health`. Absence is a documented state, not a red panel:
  install `smartmontools` and give the agent user read access to the block devices
  (typically the `disk` group), or pass `--no-smart` to opt out silently.
* **The state** is a small pid->cpu-time cache so `processes.top_cpu` can answer a
  question /proc cannot (CPU needs two samples), bounded and cleared when stale, plus
  the short-TTL SMART result cache above.

SCHEMA (GET /snapshot)
----------------------
    schema           int                                 -> 1
    ts               float  unix seconds (agent clock)
    duration_ms      float  collection cost
    agent_version    str
    tls              true      only when serving with --cert-bundle
    bundle_rev       int       the `# BUNDLE-REV` of the bundle it loaded
    host             {hostname, os_pretty, kernel, arch, python, boot_ts, uptime_seconds,
                      cpu_model, cpu_cores_logical, cpu_cores_physical, container}
    cpu              {modes: {user,nice,system,idle,iowait,irq,softirq,steal,guest,
                              guest_nice,total}  [clock ticks, cumulative],
                      cores: [{core, modes: {...}}],
                      load1, load5, load15, procs_running, procs_blocked,
                      ctxt_total, processes_total,
                      freq_mhz, freq_max_mhz}
    memory           {total_bytes, available_bytes, used_bytes, free_bytes, buffers_bytes,
                      cached_bytes, dirty_bytes, swap_total_bytes, swap_used_bytes,
                      committed_bytes}
    filesystems      [{device, mountpoint, fstype, total_bytes, used_bytes, free_bytes,
                       used_pct, read_only}]
    disk_io          {devices: [{name, model, rotational, size_bytes, reads_total,
                                 writes_total, read_bytes_total, write_bytes_total,
                                 in_flight, io_time_ms_total, weighted_io_ms_total,
                                 discard_bytes_total,
                                 smart: {model, serial, overall_health, temp_c,
                                         power_on_hours, reallocated_sector_ct,
                                         current_pending_sector_ct, offline_uncorrectable,
                                         percentage_used} | null}],
                      device_names: [str]}   # whole disks from /sys/block, no partitions
                      # `smart` is `smartctl -i -A -H` parsed; null with an errors[] entry
                      # when the binary is missing or that device cannot be read. Fields
                      # the drive does not publish stay null (e.g. NVMe has no
                      # reallocated_sector_ct), and NVMe `offline_uncorrectable` carries
                      # the health log's "Media and Data Integrity Errors" counter.
    network          {interfaces: [{name, rx_bytes_total, tx_bytes_total, rx_packets_total,
                                    tx_packets_total, rx_errors_total, tx_errors_total,
                                    rx_dropped_total, tx_dropped_total, speed_mbps}]}
    gpus             [{index, card, vendor, model, pci_slot, pci_address, driver, busy_pct,
                       vram_total_bytes, vram_used_bytes, gtt_total_bytes, gtt_used_bytes,
                       power_watts, temps: {label: celsius}, clocks: {label: mhz}}]
    temperatures     [{kind: hwmon|thermal, chip, label, celsius}]
    processes        {total, top_rss: [{pid, name, rss_bytes}],
                      top_cpu: [{pid, name, cpu_pct}]}
    errors           [str]  reads that failed; non-empty means the snapshot is partial

Deployment: see deploy/install-agent.sh (systemd unit, binds 127.0.0.1 unless you pass
--host, requires a bearer token via --token or AGENT_TOKEN when bound off-loopback). The
other protection is `--cert-bundle`/AGENT_CERT_BUNDLE: one installed PEM (server cert, server
key, the dashboard's root CA, `# BUNDLE-REV <n>` and `# REVOKED <serial>` lines) turns the
server into HTTPS that answers every path but `/health` only to a client certificate signed
by that root, and refuses one on the denylist on every path — the certificate is the auth, so
no token is read on that path.
"""

from __future__ import annotations

import argparse
import contextlib
import json
import os
import re
import shutil
import socket
import ssl
import subprocess
import tempfile
import threading
import time
from collections.abc import Callable
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any

AGENT_VERSION = "1.1.0"
SCHEMA_VERSION = 1
PAGE_SIZE = 0  # resolved from sysconf at runtime

# Filesystems that are not storage. Mirrors the spirit of node_exporter's default
# --collector.filesystem.fs-types-exclude, trimmed to what actually appears here.
PSEUDO_FSTYPES = {
    "autofs", "binfmt_misc", "bpf", "cgroup", "cgroup2", "configfs", "debugfs", "devpts",
    "devtmpfs", "efivarfs", "fuse.gvfsd-fuse", "fusectl", "hugetlbfs", "iso9660", "mqueue",
    "nsfs", "overlay", "proc", "procfs", "pstore", "ramfs", "rpc_pipefs", "securityfs",
    "selinuxfs", "squashfs", "sysfs", "tmpfs", "tracefs", "fuse.lxcfs", "binfmt",
}
PSEUDO_MOUNT_PREFIXES = ("/proc", "/sys", "/dev", "/run", "/var/lib/containers/storage",
                        "/var/lib/docker", "/snap", "/boot/grub2", "/mnt/iso")

# Interface name prefixes that are containers/bridges talking to themselves.
VIRTUAL_NET_PREFIXES = ("lo", "veth", "docker", "br-", "virbr", "br", "tap", "vnet",
                        "cni", "kube", "flannel", "dummy", "ifb", "sit", "teql")

# /sys/block entries that are not physical media. dm-* is deliberately here too: an LVM
# or LUKS volume is carved out of a disk we already count, so charting both double-counts.
VIRTUAL_DISK_PREFIXES = ("loop", "ram", "zram", "sr", "dm-", "md127")

# SMART collection bounds. 5 s per device is generous for a healthy drive and caps how
# long one wedged device can hold a snapshot hostage; the pass budget is a 20 s floor
# scaled to the per-device bound (5 s x disk count), so a 16-disk shelf gets the 80 s a
# real read takes while a single-disk host keeps 20 s and a shelf of bad disks cannot
# stall the HTTP handler; 240 s is below the dashboard's 5-minute disks cadence, so the
# slow disks job gets fresh data while the fast system poll never re-runs smartctl
# every 10 s.
SMARTCTL_TIMEOUT_SECONDS = 5.0
SMARTCTL_BUDGET_SECONDS = 20.0
SMARTCTL_CACHE_SECONDS = 240.0

# The smart block's field set — the shape the `/disks` page (phase 09) consumes. A field
# the drive does not publish stays None; the page renders `n/a`, never 0.
SMART_FIELDS = ("model", "serial", "overall_health", "temp_c", "power_on_hours",
                "reallocated_sector_ct", "current_pending_sector_ct",
                "offline_uncorrectable", "percentage_used")

_SMART_HEALTH_RE = re.compile(r"^SMART overall-health self-assessment test result:\s*(\S+)")
# The number a SMART attribute RAW_VALUE column leads with. Captured from the deployed
# smartmontools 7.5 on this fleet: a Seagate IronWolf writes `35 (Min/Max 34/44)`, a
# Western Digital writes its packed hours as `43894h+43m+32.519s`, and most drives write a
# bare `30423`. In every shape the drive's own figure is the leading decimal group.
_SMART_RAW_NUMBER_RE = re.compile(r"^\d[\d,]*(?:\.\d+)?")
# ATA SMART attribute table rows, keyed by the exact attribute names the deployed
# smartmontools 7.5 prints (verified against its own drive database, not from memory).
_SMART_ATTR_FIELDS = {
    "Temperature_Celsius": "temp_c",
    "Airflow_Temperature_Cel": "temp_c",
    "Power_On_Hours": "power_on_hours",
    "Reallocated_Sector_Ct": "reallocated_sector_ct",
    "Current_Pending_Sector": "current_pending_sector_ct",
    "Offline_Uncorrectable": "offline_uncorrectable",
    "Percent_Lifetime_Used": "percentage_used",
    "Percent_Life_Used": "percentage_used",
    "Percentage_Lifetime_Use": "percentage_used",
}

# pci slot -> human model, filled on first look (lspci output cannot change at runtime).
_GPU_MODEL_CACHE: dict[str, str | None] = {}

VENDOR_NAMES = {"0x1002": "AMD", "0x10de": "NVIDIA", "0x8086": "Intel", "0x15ad": "VMware",
                "0x1af4": "Virtio", "0x1a03": "ASPEED", "0x10ec": "Realtek"}

# sysfs names worth showing as "the CPU temperature", in preference order.
CPU_TEMP_CHIPS = ("k10temp", "zenpower", "coretemp", "cpu_thermal", "soc_thermal", "acpitz")

# Bracket groups in lspci output that name the vendor rather than the product.
_GPU_VENDOR_TAG = re.compile(r"(?:amd|ati|nvidia|intel|apple|matrox|s3|vmware|red hat|xinxiang"
                             r"|cirrus logic)(?:\s*/\s*(?:amd|ati|nvidia|intel))?", re.I)
# "Advanced Micro Devices, Inc.", "Intel Corporation", "NVIDIA Corporation" …
# The trailing `\b` matters: without it, "VGA compatible controller" splits on the "co" of
# "compatible" and the legend reads "mpatible controller: …".
_GPU_CORP_SUFFIX = re.compile(r"\b(?:Incorporated|Incorporation|Corporation|Company|Limited"
                              r"|Inc|Corp|Ltd|Co)\b\.?,?\s*", re.I)
# "00:02.0 VGA compatible controller: Intel …" — slot and class are not part of the name.
_GPU_PCI_PREFIX = re.compile(r"^[0-9a-fA-F]{2,4}:[0-9a-fA-F]{2}\.[0-9a-fA-F]\s*[^:]{0,60}:\s*")


def short_gpu_model(name: str | None) -> str | None:
    """Turn an lspci description into something a chart legend can hold.

    lspci says `Advanced Micro Devices, Inc. [AMD/ATI] Navi 48 [Radeon AI PRO R9700]`;
    the useful product name is the last bracket group, and AMD lists APU aliases separated
    by slashes where the last one is the strongest part. The full string is still reported
    as `model_full` so nothing is lost — this is display, not identity.
    """
    if not name:
        return None
    groups = re.findall(r"\[([^\]]+)\]", name)
    # [AMD/ATI] and [NVIDIA] are vendor tags, not products — keeping them turned an iGPU
    # named Phoenix1 into a legend that said "ATI".
    products = [g for g in groups if not _GPU_VENDOR_TAG.fullmatch(g.strip())]
    if products:
        candidate = products[-1].split("/")[-1].strip()
    else:
        rest = _GPU_PCI_PREFIX.sub("", name)
        rest = re.sub(r"\[[^\]]+\]", " ", rest)                  # drop vendor tags
        candidate = _GPU_CORP_SUFFIX.split(rest, maxsplit=1)[-1].strip(" ,;-")
    candidate = candidate.strip() or name.strip()
    return candidate[:40] + "\u2026" if len(candidate) > 40 else candidate


def _to_float(text: str) -> float | None:
    try:
        value = float(text)
    except (TypeError, ValueError):
        return None
    return value if value == value and abs(value) != float("inf") else None


def _at(values: list[float | None], index: int) -> float | None:
    """Element `index`, or None. /proc and /sys files are truncated more often than you'd think.

    This was a closure over the enclosing loop variable, which is the bug pattern B023 warns
    about: escape the loop and it reads the *last* iteration's counters. It takes its data now.
    """
    return values[index] if index < len(values) else None


class Collector:
    """Reads one host's /proc and /sys. `root` is injectable so tests can hand it a tree
    of fixture files instead of a live machine."""

    def __init__(self, root: str | os.PathLike[str] = "/", *, top_processes: int = 5,
                 net_all: bool = False, smartctl: str = "smartctl",
                 smart: bool = True) -> None:
        self.root = Path(root)
        self.top_processes = top_processes
        self.net_all = net_all
        self.smartctl = smartctl
        self.smart_enabled = smart
        self.smart_timeout_seconds = SMARTCTL_TIMEOUT_SECONDS
        self.smart_budget_seconds = SMARTCTL_BUDGET_SECONDS
        self.smart_cache_seconds = SMARTCTL_CACHE_SECONDS
        self._smart_lock = threading.Lock()  # ThreadingHTTPServer may poll while a pass runs
        self._smart_cache: tuple[float, dict[str, dict[str, Any] | None],
                                 list[str]] | None = None
        self._smart_refreshing = False       # at most one background refresh at a time
        self._lock = threading.Lock()  # one HTTP handler at a time may touch the pid cache
        self._pid_cpu: dict[int, tuple[float, int, str]] = {}  # pid -> (ts, jiffies, name)
        self._boot_ts: float | None = None
        self._page_size = os.sysconf("SC_PAGE_SIZE") or 4096
        self._ticks = os.sysconf("SC_CLK_TCK") or 100
        self._cpu_model_cache: str | None = None

    # -- tiny fs helpers ----------------------------------------------------- #
    def _path(self, *parts: str) -> Path:
        return self.root.joinpath(*parts)

    def read_text(self, *parts: str) -> str | None:
        try:
            return self._path(*parts).read_text(errors="replace")
        except (OSError, ValueError):
            return None

    def read_lines(self, *parts: str) -> list[str]:
        text = self.read_text(*parts)
        return text.splitlines() if text else []

    def read_number(self, *parts: str) -> float | None:
        """First number in the file.

        `/proc/uptime` is `210036.18 9613750.86`, while `/sys/.../scaling_cur_freq` is bare.
        Parsing the whole file as a float returns None for exactly the files that matter,
        which reads as "this kernel has no uptime" rather than "you forgot to split".
        """
        text = self.read_text(*parts)
        if not text:
            return None
        fields = text.split()
        return _to_float(fields[0]) if fields else None

    # -- cpu ----------------------------------------------------------------- #
    def cpu(self) -> dict[str, Any]:
        """Jiffies per mode, cumulatively since boot. Divide by SC_CLK_TCK for seconds."""
        modes: dict[str, float] = {}
        cores: list[dict[str, Any]] = []
        for line in self.read_lines("proc/stat"):
            if not line.startswith("cpu"):
                break
            fields = line.split()
            name = fields[0]
            values = [float(v) for v in fields[1:11]] if len(fields) > 1 else []
            if not values:
                continue
            while len(values) < 10:
                values.append(0.0)
            sample = dict(zip(MODE_NAMES, values[:10], strict=True))
            # guest and guest_nice are already counted inside user/nice on Linux
            sample["total"] = sum(values[:8])
            if name == "cpu":
                modes = sample
            elif name[3:].isdigit():
                cores.append({"core": int(name[3:]), "modes": sample})
        load = self.read_text("proc/loadavg")
        load_parts = load.split() if load else []
        procs_running = procs_blocked = ctxt_total = processes_total = None
        for line in self.read_lines("proc/stat"):
            if line.startswith("procs_running"):
                procs_running = _to_float(line.split()[-1])
            elif line.startswith("procs_blocked"):
                procs_blocked = _to_float(line.split()[-1])
            elif line.startswith("ctxt"):
                ctxt_total = _to_float(line.split()[-1])
            elif line.startswith("processes"):
                processes_total = _to_float(line.split()[-1])
        freq = self.cpu_frequency()
        return {
            "modes": modes or None,
            "cores": cores,
            "clock_ticks_per_second": self._ticks,
            "load1": _to_float(load_parts[0]) if len(load_parts) > 0 else None,
            "load5": _to_float(load_parts[1]) if len(load_parts) > 1 else None,
            "load15": _to_float(load_parts[2]) if len(load_parts) > 2 else None,
            "procs_running": procs_running,
            "procs_blocked": procs_blocked,
            "ctxt_total": ctxt_total,
            "processes_total": processes_total,
            "freq_mhz": freq["current"],
            "freq_max_mhz": freq["max"],
        }

    def cpu_frequency(self) -> dict[str, float | None]:
        """Average current / best-case maximum, in MHz."""
        current: list[float] = []
        maximum: list[float] = []
        base = self._path("sys/devices/system/cpu")
        if base.is_dir():
            for cpu in sorted(base.glob("cpu[0-9]*")):
                cur = self.read_number(str(cpu.relative_to(self.root)), "cpufreq/scaling_cur_freq")
                if cur is None:
                    cur = self.read_number(str(cpu.relative_to(self.root)), "cpufreq/cpuinfo_cur_freq")
                if cur is not None:
                    current.append(cur / 1000.0)
                mx = self.read_number(str(cpu.relative_to(self.root)), "cpufreq/cpuinfo_max_freq")
                if mx is not None:
                    maximum.append(mx / 1000.0)
        return {
            "current": (sum(current) / len(current)) if current else None,
            "max": max(maximum) if maximum else None,
        }

    def cpu_model(self) -> str | None:
        if self._cpu_model_cache:
            return self._cpu_model_cache
        for line in self.read_lines("proc/cpuinfo"):
            if line.lower().startswith(("model name", "cpu model", "hardware")):
                _, _, value = line.partition(":")
                if value.strip():
                    self._cpu_model_cache = value.strip()
                    return self._cpu_model_cache
        # arm64 fallback: compatible = "arm,cortex-a76"
        text = self.read_text("proc/device-tree/compatible")
        if text:
            self._cpu_model_cache = text.strip("\x00").split(",")[-1].strip()
            return self._cpu_model_cache
        return None

    def core_counts(self) -> tuple[int | None, int | None]:
        text = self.read_text("proc/cpuinfo") or ""
        logical = len(re.findall(r"^processor\s*:", text, re.M)) or (os.cpu_count() or None)
        # one block per *thread*, so dedupe (socket, core) — counting how many "core id:"
        # lines exist yields 1 for the set of colons, which is how a 24-core TR reports
        # "1 physical core" and makes a fleet table quietly meaningless.
        seen: set[tuple[str, str]] = set()
        for block in text.split("\n\n"):
            socket = re.search(r"^physical id\s*:\s*(\d+)", block, re.M)
            core = re.search(r"^core id\s*:\s*(\d+)", block, re.M)
            if socket and core:
                seen.add((socket.group(1), core.group(1)))
        physical: int | None = len(seen) or None
        if physical is None and logical:
            threads = self._threads_per_core()
            if threads:
                physical = max(1, logical // threads)
        return (logical or None), physical

    def _threads_per_core(self) -> int | None:
        """SMT width from cpu0's sibling list: `0,24` or `0-1` → 2. VMs often omit it."""
        listing = self.read_text("sys/devices/system/cpu/cpu0/topology/thread_siblings_list")
        if not listing:
            return None
        count = 0
        for part in listing.strip().split(","):
            start, dash, end = part.partition("-")
            first = _to_float(start)
            if first is None:
                continue
            last = _to_float(end) if dash else None
            count += int((last if last is not None else first) - first) + 1
        return count or None

    # -- memory -------------------------------------------------------------- #
    def memory(self) -> dict[str, float | None]:
        raw: dict[str, float] = {}
        for line in self.read_lines("proc/meminfo"):
            name, _, rest = line.partition(":")
            value = _to_float(rest.split()[0]) if rest.split() else None
            if value is not None:
                raw[name.strip()] = value * 1024.0  # meminfo is in kB
        if not raw:
            return dict.fromkeys(MEMORY_KEYS)
        total = raw.get("MemTotal")
        available = raw.get("MemAvailable", raw.get("MemFree"))
        swap_total = raw.get("SwapTotal", 0.0)
        swap_free = raw.get("SwapFree", 0.0)
        out: dict[str, float | None] = {
            "total_bytes": total,
            "free_bytes": raw.get("MemFree"),
            "available_bytes": available,
            "used_bytes": (total - available) if (total is not None and available is not None) else None,
            "buffers_bytes": raw.get("Buffers"),
            "cached_bytes": (raw.get("Cached", 0.0) + raw.get("SReclaimable", 0.0)) or None,
            "dirty_bytes": raw.get("Dirty"),
            "swap_total_bytes": swap_total,
            "swap_used_bytes": swap_total - swap_free,
            "committed_bytes": raw.get("Committed_AS"),
        }
        for key in MEMORY_KEYS:
            out.setdefault(key, None)
        return out

    # -- filesystems --------------------------------------------------------- #
    def filesystems(self) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []
        seen: set[tuple[str, str]] = set()
        for line in self.read_lines("proc/mounts"):
            parts = line.split()
            if len(parts) < 3:
                continue
            device, mountpoint, fstype = parts[0], parts[1], parts[2]
            mountpoint = mountpoint.replace("\\040", " ").replace("\\011", "\t")
            if fstype in PSEUDO_FSTYPES:
                continue
            if any(mountpoint == p or mountpoint.startswith(p + "/") for p in PSEUDO_MOUNT_PREFIXES):
                continue
            key = (device, mountpoint)
            if key in seen:
                continue  # bind mounts / btrfs subvolume aliases
            seen.add(key)
            try:
                st = os.statvfs(mountpoint)
            except OSError:
                continue
            block = st.f_frsize or st.f_bsize
            total = st.f_blocks * block
            free = st.f_bavail * block          # what an unprivileged writer can use
            used = (st.f_blocks - st.f_bfree) * block
            denominator = used + free           # df's "Use%" basis, not raw total
            out.append({
                "device": device,
                "mountpoint": mountpoint,
                "fstype": fstype,
                "total_bytes": total or None,
                "used_bytes": used,
                "free_bytes": free,
                "used_pct": (used / denominator * 100.0) if denominator > 0 else None,
                "read_only": bool(st.f_flag & os.ST_RDONLY),
            })
        out.sort(key=lambda item: item["mountpoint"])
        return out

    # -- disk io ------------------------------------------------------------- #
    def disk_io(self) -> dict[str, Any]:
        """Whole disks via /sys/block/<dev>/stat, which is documented ABI-stable.

        /proc/diskstats is NOT: the discard/flush counters were inserted at different
        offsets in different kernels, so "which index is in-flight" depends on the build.
        Measured on driveripper (kernel 7.2): its diskstats line and its /sys/block/stat
        line are the same 17 numbers — classic 11 first (in-flight at index 8, matching
        /sys/block/<dev>/inflight), then the discard/flush tail. So the classic 11 are
        read from fixed indexes and the tail only when the 17-number shape is present.

        /sys/block lists whole disks only (nvme0n1, not nvme0n1p1), which is exactly the
        set we want: partitions and dm-* layers would double-count the same bytes.
        """
        devices: list[dict[str, Any]] = []
        names: list[str] = []
        base = self._path("sys/block")
        if not base.is_dir():
            return {"devices": [], "device_names": []}
        for entry in sorted(base.iterdir(), key=lambda p: p.name):
            name = entry.name
            if name.startswith(VIRTUAL_DISK_PREFIXES):
                continue
            numbers = (self.read_text("sys/block", name, "stat") or "").split()
            if len(numbers) < 11:
                continue
            values = [_to_float(v) for v in numbers]
            rotational = self.read_number("sys/block", name, "queue/rotational")
            model = (self.read_text("sys/block", name, "device/model") or "").strip() or None
            names.append(name)
            devices.append({
                "name": name,
                "model": model,
                "rotational": None if rotational is None else rotational == 1.0,
                "size_bytes": (self.read_number("sys/block", name, "size") or 0) * 512.0 or None,
                "reads_total": _at(values, 0),
                "writes_total": _at(values, 4),
                "read_bytes_total": (_at(values, 2) or 0.0) * 512.0,
                "write_bytes_total": (_at(values, 6) or 0.0) * 512.0,
                "in_flight": _at(values, 8),
                "io_time_ms_total": _at(values, 9),
                "weighted_io_ms_total": _at(values, 10),
                # discard/flush tail: only this shape is verified (see docstring)
                "discard_bytes_total": ((_at(values, 13) or 0.0) * 512.0
                                        if len(values) >= 17 else None),
            })
        return {"devices": devices, "device_names": names}

    # -- smart (smartctl) ------------------------------------------------------ #
    def smart(self, device_names: list[str]) -> tuple[dict[str, dict[str, Any] | None],
                                                      list[str]]:
        """Best-effort `smartctl -i -A -H <device>` per whole disk; never raises.

        Returns (per-device parsed dict or None, error lines). Failure semantics (phase
        09 task 01): the binary missing produces **one** error entry for the whole pass
        and every disk `None`; one denied/wedged device produces its own error entry and
        nulls only that disk. Results are cached for `smart_cache_seconds` so the fast
        system poll cannot re-issue ATA commands every few seconds; the cache replays the
        same error lines, so a snapshot always states why `smart` is null.

        An **expired** cache is refreshed on a background thread and the request is served
        the previous result, because the pass budget (20 s floor, 5 s per disk) is five
        times the dashboard's
        `HOST_TIMEOUT_SECONDS` (4 s): run inline on the request path, one wedged disk would
        turn the *system* source red every `smart_cache_seconds` for a fault in an add-on
        (AGENTS.md invariant 4 — SMART must not blink the page that does not use it). The
        very first pass still runs inline, so a freshly started agent answers `/snapshot`
        with real SMART fields instead of a page of `n/a` until someone polls it twice.
        """
        now = time.monotonic()
        with self._smart_lock:
            cached = self._smart_cache
            if cached is not None:
                if now - cached[0] < self.smart_cache_seconds or self._smart_refreshing:
                    return dict(cached[1]), list(cached[2])
                self._smart_refreshing = True
                threading.Thread(target=self._refresh_smart, args=(list(device_names),),
                                 name="smart-refresh", daemon=True).start()
                return dict(cached[1]), list(cached[2])
        return self._smart_pass(device_names)

    def _refresh_smart(self, device_names: list[str]) -> None:
        """The expired-cache pass, off the request path (see `smart()`).

        A failure here keeps the previous cache (and its age) rather than clearing it: the
        snapshot keeps answering with the last truth it had, and the retry comes on the next
        expiry instead of hammering a disk that just failed.
        """
        try:
            self._smart_pass(device_names)
        except Exception as exc:  # noqa: BLE001 - the thread must always clear its own flag
            with self._smart_lock:
                previous = self._smart_cache
                results = dict(previous[1]) if previous else dict.fromkeys(device_names)
                errors = list(previous[2]) if previous else []
                self._smart_cache = (time.monotonic(), results,
                                     [*errors,
                                      f"smart: background refresh failed: "
                                      f"{type(exc).__name__}: {exc}"])
        finally:
            with self._smart_lock:
                self._smart_refreshing = False

    def _smart_pass(self, device_names: list[str]) -> tuple[dict[str, dict[str, Any] | None],
                                                            list[str]]:
        """One `smartctl` pass over the whole disks, and the cache write that goes with it."""
        now = time.monotonic()
        results: dict[str, dict[str, Any] | None] = {}
        errors: list[str] = []
        # 20 s floor scaled to the per-device bound: a 16-disk shelf takes more than 20 s
        # when the drives answer, and truncating the pass would null the tail drives.
        # An explicit override (the budget test sets this to 0) wins over the scale.
        budget = (max(SMARTCTL_BUDGET_SECONDS, self.smart_timeout_seconds * len(device_names))
                  if self.smart_budget_seconds == SMARTCTL_BUDGET_SECONDS
                  else self.smart_budget_seconds)
        deadline = now + budget
        for name in device_names:
            left = deadline - time.monotonic()
            if left <= 0:
                remaining = [n for n in device_names if n not in results]
                for skipped_name in remaining:
                    results[skipped_name] = None
                errors.append(f"smart: pass budget of {budget:.0f}s "
                              f"exhausted, {len(remaining)} disk(s) skipped")
                break
            try:
                proc = subprocess.run([self.smartctl, "-i", "-A", "-H", f"/dev/{name}"],
                                      capture_output=True, text=True,
                                      timeout=min(left, self.smart_timeout_seconds),
                                      check=False)
            except FileNotFoundError:
                # one entry for the pass, not one per disk: a missing binary is one fact
                errors.append(f"smart: {self.smartctl} not found — install smartmontools "
                              "and give this user device read access (disk group) for "
                              "SMART data")
                for missing_name in device_names:
                    results.setdefault(missing_name, None)
                break
            except subprocess.TimeoutExpired:
                results[name] = None
                errors.append(f"smart: {name}: smartctl timed out after "
                              f"{self.smart_timeout_seconds:.0f}s (device wedged?)")
                continue
            except OSError as exc:
                results[name] = None
                errors.append(f"smart: {name}: smartctl could not run: {exc}")
                continue
            parsed = parse_smart_output(proc.stdout)
            if any(value is not None for value in parsed.values()):
                results[name] = parsed  # partial output is still what the drive published
            else:
                results[name] = None
                errors.append(f"smart: {name}: "
                              f"{smart_failure_line(proc.returncode, proc.stdout, proc.stderr)}")
        with self._smart_lock:
            self._smart_cache = (time.monotonic(), results, errors)
        return dict(results), list(errors)

    def attach_smart(self, disk_io: dict[str, Any], errors: list[str]) -> None:
        """Put `smart` on every disk row; `--no-smart` leaves them all null silently."""
        devices = disk_io.get("devices") or []
        for device in devices:
            device["smart"] = None
        if not devices or not self.smart_enabled:
            return
        results, smart_errors = self.smart([str(device.get("name")) for device in devices])
        errors.extend(smart_errors)
        for device in devices:
            parsed = results.get(str(device.get("name")))
            if not parsed:
                continue
            smart = dict(parsed)
            if not smart.get("model"):
                # smartctl's identity block can be absent (some bridges refuse -i while
                # answering the health log); the agent's sysfs model is the fallback.
                smart["model"] = device.get("model")
            device["smart"] = smart

    # -- network ------------------------------------------------------------- #
    def network(self) -> dict[str, Any]:
        interfaces: list[dict[str, Any]] = []
        for line in self.read_lines("proc/net/dev"):
            name, sep, rest = line.partition(":")
            if not sep:
                continue
            name = name.strip()
            if not self.net_all and name.startswith(VIRTUAL_NET_PREFIXES):
                continue
            fields = [_to_float(v) for v in rest.split()]
            if len(fields) < 9:
                continue

            interfaces.append({
                "name": name,
                "rx_bytes_total": _at(fields, 0),
                "rx_packets_total": _at(fields, 1),
                "rx_errors_total": _at(fields, 2),
                "rx_dropped_total": _at(fields, 3),
                "tx_bytes_total": _at(fields, 8),
                "tx_packets_total": _at(fields, 9),
                "tx_errors_total": _at(fields, 10),
                "tx_dropped_total": _at(fields, 11),
                "speed_mbps": self._net_speed(name),
            })
        interfaces.sort(key=lambda item: item["name"])
        return {"interfaces": interfaces}

    def _net_speed(self, name: str) -> float | None:
        value = self.read_number("sys/class/net", name, "speed")
        return value if value and value > 0 else None

    # -- gpus ---------------------------------------------------------------- #
    def gpus(self) -> list[dict[str, Any]]:
        """AMD GPUs via DRM sysfs (gpu_busy_percent, mem_info_vram_*, power, hwmon temps).

        That set exists on amdgpu since kernel ~4.x and is world-readable, which is why
        this agent needs no root. NVIDIA/Intel are not wired up: the fleet has no such
        card, and guessing at sysfs paths we cannot test would be worse than `null`.
        """
        out: list[dict[str, Any]] = []
        base = self._path("sys/class/drm")
        if not base.is_dir():
            return out
        for card in sorted(base.glob("card[0-9]*")):
            name = card.name
            if "-" in name or name == "card":  # only cardN, not cardN-DP-1 connectors
                continue
            device = card / "device"
            if not device.is_dir():
                continue
            busy = self.read_number(str(device.relative_to(self.root)), "gpu_busy_percent")
            vram_total = self.read_number(str(device.relative_to(self.root)), "mem_info_vram_total")
            vram_used = self.read_number(str(device.relative_to(self.root)), "mem_info_vram_used")
            if busy is None and vram_total is None:
                continue  # not a compute-capable GPU (connector-only / virtual)
            pci_id = self._read_uevent(device, "PCI_ID")  # "1002:7551"
            slot = self._read_uevent(device, "PCI_SLOT_NAME")  # "0000:4d:00.0"
            vendor_id, _, device_id = (pci_id or "").partition(":")
            driver = self._driver_name(device)
            full_model = self._gpu_model(slot, vendor_id, device_id)
            out.append({
                "index": len(out),
                "card": name,
                "vendor": VENDOR_NAMES.get(f"0x{vendor_id}" if vendor_id else "", vendor_id or None),
                "model": short_gpu_model(full_model),
                "model_full": full_model,
                "pci_slot": slot,
                "pci_address": pci_id or None,
                "driver": driver,
                "busy_pct": busy,
                "vram_total_bytes": vram_total,
                "vram_used_bytes": vram_used,
                # On an APU (Strix Halo) the dedicated carve-out is tiny and the model
                # actually lives in GTT (shared system RAM): report both, never silently
                # add them into one number that hides which pool was used.
                "gtt_total_bytes": self.read_number(str(device.relative_to(self.root)),
                                                     "mem_info_gtt_total"),
                "gtt_used_bytes": self.read_number(str(device.relative_to(self.root)),
                                                   "mem_info_gtt_used"),
                "power_watts": self._gpu_power(device),
                "temps": self._gpu_temps(device),
                "clocks": self._gpu_clocks(device),
            })
        return out

    def _gpu_power(self, device: Path) -> float | None:
        """Board power in watts. hwmon `power*_input`/`*_average`, microWatts, PPT preferred.

        Not always present: the amdgpu build on driveripper ships power1_label=PPT with no
        matching input file, so this honestly returns None instead of inventing a number.
        """
        hwmon_dir = device / "hwmon"
        if not hwmon_dir.is_dir():
            return None
        best: float | None = None
        for hwmon in hwmon_dir.glob("hwmon*"):
            for suffix in ("input", "average", "max"):
                for candidate in sorted(hwmon.glob(f"power*_{suffix}")):
                    key = candidate.stem.split("_")[0]
                    label_file = hwmon / f"{key}_label"
                    label = (self.read_text(str(label_file.relative_to(self.root))) or "").strip().lower()
                    value = self.read_number(str(candidate.relative_to(self.root)))
                    if value is None:
                        continue
                    watts = value / 1_000_000.0
                    if best is None:
                        best = watts
                    if "ppt" in label or "package" in label:
                        return watts
        return best

    def _gpu_clocks(self, device: Path) -> dict[str, float]:
        """Clocks in MHz, keyed by their sysfs label (sclk / mclk)."""
        clocks: dict[str, float] = {}
        hwmon_dir = device / "hwmon"
        if not hwmon_dir.is_dir():
            return clocks
        for hwmon in hwmon_dir.glob("hwmon*"):
            for input_file in sorted(hwmon.glob("freq*_input")):
                key = input_file.stem.replace("_input", "")
                label_file = hwmon / f"{key}_label"
                label = (self.read_text(str(label_file.relative_to(self.root))) or key).strip() \
                    if label_file.exists() else key
                value = self.read_number(str(input_file.relative_to(self.root)))
                if value is not None:
                    clocks[label] = value / 1_000_000.0
        return clocks

    def _read_uevent(self, device: Path, key: str) -> str | None:
        for line in self.read_lines(str(device.relative_to(self.root)), "uevent"):
            name, _, value = line.partition("=")
            if name == key:
                return value.strip().lower()
        return None

    def _driver_name(self, device: Path) -> str | None:
        driver = device / "driver"
        try:
            return os.path.basename(os.readlink(driver))
        except OSError:
            return None

    def _gpu_model(self, slot: str | None, vendor_id: str | None, device_id: str | None) -> str | None:
        """Human GPU name. lspci when it exists (cached: the answer never changes at
        runtime), otherwise vendor + device id — never a bare "gpu"."""
        cache_key = slot or f"{vendor_id}:{device_id}"
        if cache_key in _GPU_MODEL_CACHE:
            return _GPU_MODEL_CACHE[cache_key]
        model: str | None = None
        if slot:
            found = self._lspci_name(slot.split(":", 1)[-1] if slot.count(":") == 2 else slot)
            if found:
                model = found
        if model is None:
            vendor = VENDOR_NAMES.get(f"0x{vendor_id}" if vendor_id else "")
            if vendor and device_id:
                model = f"{vendor} GPU ({vendor_id}:{device_id})"
            elif vendor:
                model = vendor
        _GPU_MODEL_CACHE[cache_key] = model
        return model

    def _lspci_name(self, slot: str) -> str | None:
        """`lspci -nns <slot>` -> the text after the class, e.g. 'Navi 48 [Radeon ...]'.

        Optional: lspci is not guaranteed to be installed, so the caller must cope with
        None. The subprocess is bounded and never raises into the snapshot.
        """
        try:
            import subprocess  # noqa: PLC0415 - only needed for one lspci call

            raw = subprocess.run(["lspci", "-nns", slot], capture_output=True, text=True,
                                 timeout=3, check=False)
        except Exception:  # noqa: BLE001 - missing lspci / no permissions is normal
            return None
        if raw.returncode != 0 or not raw.stdout.strip():
            return None
        line = raw.stdout.strip().splitlines()[0]
        _, _, rest = line.partition(":")
        _, _, rest = rest.partition(":")
        rest = re.sub(r"\s*\[[0-9a-fA-F]{4}:[0-9a-fA-F]{4}\]", "", rest)
        rest = re.sub(r"\s*\(rev [0-9a-fA-F]+\)\s*$", "", rest)
        return rest.strip() or None

    def _gpu_temps(self, device: Path) -> dict[str, float]:
        temps: dict[str, float] = {}
        hwmon_dir = device / "hwmon"
        if not hwmon_dir.is_dir():
            return temps
        for hwmon in hwmon_dir.glob("hwmon*"):
            for input_file in sorted(hwmon.glob("temp*_input")):
                label = input_file.stem.replace("_input", "")
                named = hwmon / f"{label}_label"
                if named.exists():
                    label = (self.read_text(str(named.relative_to(self.root))) or label).strip()
                value = self.read_number(str(input_file.relative_to(self.root)))
                if value is not None:
                    temps[label] = value / 1000.0
        return temps

    # -- temperatures -------------------------------------------------------- #
    def temperatures(self) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []
        hwmon = self._path("sys/class/hwmon")
        if hwmon.is_dir():
            for unit in sorted(hwmon.glob("hwmon*")):
                chip = (self.read_text(str(unit.relative_to(self.root)), "name") or unit.name).strip()
                for input_file in sorted(unit.glob("temp*_input")):
                    key = input_file.stem.replace("_input", "")
                    label_file = unit / f"{key}_label"
                    label = chip
                    if label_file.exists():
                        label = (self.read_text(str(label_file.relative_to(self.root))) or chip).strip()
                    value = self.read_number(str(input_file.relative_to(self.root)))
                    if value is not None:
                        out.append({"kind": "hwmon", "chip": chip, "label": label,
                                    "celsius": value / 1000.0})
        thermal = self._path("sys/class/thermal")
        if thermal.is_dir():
            for zone in sorted(thermal.glob("thermal_zone*")):
                ztype = (self.read_text(str(zone.relative_to(self.root)), "type") or zone.name).strip()
                value = self.read_number(str(zone.relative_to(self.root)), "temp")
                if value is not None:
                    out.append({"kind": "thermal", "chip": ztype, "label": ztype,
                                "celsius": value / 1000.0})
        return out

    def cpu_temperature(self, temps: list[dict[str, Any]]) -> float | None:
        for chip in CPU_TEMP_CHIPS:
            picks = [t["celsius"] for t in temps if t["chip"] == chip and t["celsius"] is not None]
            if picks:
                return max(picks)  # Tctl/Tdie style: the hottest sensor on that chip
        return None

    # -- processes ----------------------------------------------------------- #
    def processes(self) -> dict[str, Any]:
        """Top-N by RSS (instant) and by CPU (needs the pid cache — see module docstring)."""
        now = time.time()
        seen: set[int] = set()
        rss_rows: list[dict[str, Any]] = []
        cpu_rows: list[dict[str, Any]] = []
        total = 0
        proc = self._path("proc")
        try:
            entries = list(proc.iterdir())
        except OSError:
            return {"total": None, "top_rss": [], "top_cpu": []}
        for entry in entries:
            if not entry.name.isdigit():
                continue
            pid = int(entry.name)
            stat = self.read_text("proc", entry.name, "stat")
            if not stat:
                continue
            fields = _split_stat(stat)
            if not fields:
                continue
            total += 1
            seen.add(pid)
            name = fields[0]
            utime = _to_float(fields[11]) if len(fields) > 11 else None
            stime = _to_float(fields[12]) if len(fields) > 12 else None
            jiffies = (utime + stime) if (utime is not None and stime is not None) else None
            rss_pages = self._rss_pages(entry.name)
            if rss_pages is not None:
                rss_rows.append({"pid": pid, "name": name, "rss_bytes": rss_pages * self._page_size})
            previous = self._pid_cpu.get(pid)
            if jiffies is not None:
                if previous is not None and previous[0] < now:
                    elapsed = now - previous[0]
                    used = (jiffies - previous[1]) / self._ticks
                    if elapsed > 0 and used >= 0:
                        cpu_rows.append({"pid": pid, "name": name,
                                         "cpu_pct": used / elapsed * 100.0})
                self._pid_cpu[pid] = (now, jiffies, name)
        with self._lock:
            for pid in list(self._pid_cpu):
                if pid not in seen:
                    del self._pid_cpu[pid]
        rss_rows.sort(key=lambda row: row["rss_bytes"], reverse=True)
        cpu_rows.sort(key=lambda row: row["cpu_pct"], reverse=True)
        return {"total": total, "top_rss": rss_rows[: self.top_processes],
                "top_cpu": cpu_rows[: self.top_processes]}

    def _rss_pages(self, pid_dir: str) -> float | None:
        """statm field 2 is resident set size in pages."""
        parts = (self.read_text("proc", pid_dir, "statm") or "").split()
        return _to_float(parts[1]) if len(parts) > 1 else None

    # -- host identity ------------------------------------------------------- #
    def host_info(self) -> dict[str, Any]:
        uptime = self.read_number("proc/uptime")
        if uptime is not None and self._boot_ts is None:
            self._boot_ts = time.time() - uptime
        release: dict[str, str] = {}
        for line in self.read_lines("etc/os-release"):
            key, sep, value = line.partition("=")
            if sep:
                release[key] = value.strip('"')
        logical, physical = self.core_counts()
        meminfo = self.memory()
        try:
            uname = os.uname()
        except AttributeError:  # pragma: no cover - non-POSIX
            uname = None
        return {
            "hostname": socket.gethostname(),
            "os_pretty": release.get("PRETTY_NAME") or " ".join(
                v for v in (release.get("NAME"), release.get("VERSION")) if v) or None,
            "os_id": release.get("ID"),
            "kernel": uname.release if uname else None,
            "arch": uname.machine if uname else None,
            "python": f"{os.sys.version_info.major}.{os.sys.version_info.minor}.{os.sys.version_info.micro}",
            "cpu_model": self.cpu_model(),
            "cpu_cores_logical": logical,
            "cpu_cores_physical": physical,
            "memory_total_bytes": meminfo.get("total_bytes"),
            "boot_ts": self._boot_ts,
            "uptime_seconds": uptime,
            "container": self._in_container(),
            "agent_version": AGENT_VERSION,
        }

    def _in_container(self) -> bool:
        for marker in ("run/.containerenv", ".dockerenv"):
            if self._path(marker).exists():
                return True
        cgroup = self.read_text("proc/1/cgroup") or ""
        return bool(re.search(r"(docker|containerd|kubepods|libpod)", cgroup))

    # -- everything ---------------------------------------------------------- #
    def snapshot(self, *, top: int | None = None) -> dict[str, Any]:
        started = time.perf_counter()
        errors: list[str] = []
        info = self._safe(self.host_info, "host", errors)
        temps = self._safe(self.temperatures, "temperatures", errors) or []
        if top is not None:
            self.top_processes = top
        # no lock here: processes() takes it itself, and threading.Lock is not reentrant
        processes = self._safe(self.processes, "processes", errors) or {"total": None,
                                                                       "top_rss": [],
                                                                       "top_cpu": []}
        disk_io = self._safe(self.disk_io, "disk_io", errors) or {"devices": [], "device_names": []}
        # best-effort, after the cheap reads: a wedged disk can cost the pass budget, not
        # the snapshot (see the module docstring and smart() for the bounds)
        self._safe(lambda: self.attach_smart(disk_io, errors), "smart", errors)
        return {
            "schema": SCHEMA_VERSION,
            "agent_version": AGENT_VERSION,
            "ts": time.time(),
            "duration_ms": round((time.perf_counter() - started) * 1000.0, 2),
            "host": info or {},
            "cpu": self._safe(self.cpu, "cpu", errors) or {},
            "memory": self._safe(self.memory, "memory", errors) or {},
            "filesystems": self._safe(self.filesystems, "filesystems", errors) or [],
            "disk_io": disk_io,
            "network": self._safe(self.network, "network", errors) or {"interfaces": []},
            "gpus": self._safe(self.gpus, "gpus", errors) or [],
            "temperatures": temps,
            "cpu_temperature_celsius": self.cpu_temperature(temps),
            "processes": processes,
            "errors": errors,
        }

    def _safe(self, fn: Callable[[], Any], name: str, errors: list[str]) -> Any:
        """One failing collector degrades its own section to empty; never the snapshot."""
        try:
            return fn()
        except Exception as exc:  # noqa: BLE001 - a bad /proc read must not fail the poll
            errors.append(f"{name}: {type(exc).__name__}: {exc}")
            return None


MODE_NAMES = ("user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal",
              "guest", "guest_nice")
MEMORY_KEYS = ("total_bytes", "free_bytes", "available_bytes", "used_bytes", "buffers_bytes",
               "cached_bytes", "dirty_bytes", "swap_total_bytes", "swap_used_bytes",
               "committed_bytes")

_SMART_FAILURE_HINT_RE = re.compile(r"failed|denied|error|unable|unsupported|not supported", re.I)


def smart_failure_line(returncode: int, stdout: str | None, stderr: str | None) -> str:
    """The sentence that says *why* smartctl answered nothing.

    Never its first line: smartctl opens every run with its version banner, and errors
    land on stdout, on stderr, or both. Take the first line that reads like a failure,
    else the last non-empty line, else the exit status — an error entry that quotes the
    copyright line is worse than no error entry.
    """
    blobs = [stderr or "", stdout or ""]
    for blob in blobs:
        for line in blob.splitlines():
            if line.strip() and _SMART_FAILURE_HINT_RE.search(line):
                return line.strip()
    for blob in blobs:  # nothing that reads like a failure: last real line of stderr first
        lines = [line.strip() for line in blob.splitlines() if line.strip()]
        if lines:
            return lines[-1]
    return f"exit status {returncode}"


def _smart_number(text: str) -> float | None:
    """`8,760` and `38` are numbers; a hex-packed vendor raw is not one we can decode
    without a drive-specific table, so it stays None — an invented sector count is the
    exact failure mode this agent exists to avoid."""
    return _to_float(text.strip().replace(",", ""))


def _smart_raw_number(raw: str) -> float | None:
    """The value of a SMART attribute row's RAW_VALUE column (column 9, `raw`).

    smartctl prints one bare number for most attributes and the shapes above for the rest,
    all of which are read here; `0x...` is a packed raw, and a packed raw read as its
    leading `0` would report a healthy-looking zero for a value we never decoded, so that
    one answers None."""
    token = raw.strip().split(" ")[0]
    if not token or token.lower().startswith("0x"):
        return None
    match = _SMART_RAW_NUMBER_RE.match(token)
    return _smart_number(match.group(0)) if match else None


def parse_smart_output(text: str | None) -> dict[str, Any]:
    """Parse `smartctl -i -A -H` stdout (ATA, NVMe and SCSI share this entry point).

    Only labels the deployed smartmontools 7.5 actually prints are matched (they were
    verified against that binary's own format strings and drive database, and are pinned
    by tests/fixtures/smartctl_*.txt). Values the drive or the parser cannot supply stay
    None. `overall_health` keeps smartctl's own verdict word (`PASSED`, `FAILED`,
    `UNKNOWN`, trailing `!` stripped); NVMe `offline_uncorrectable` carries the health
    log's `Media and Data Integrity Errors` counter — same meaning, different name, and
    the ATA `Offline_Uncorrectable` attribute when both exist wins because they are the
    drive's own words for it. ATA rows are read from the RAW_VALUE column rather than the
    end of the line, because this binary annotates that column — `35 (Min/Max 34/44)` on a
    Seagate, `43894h+43m+32.519s` on a Western Digital — and the last token of such a row
    is not the value.
    """
    out: dict[str, Any] = dict.fromkeys(SMART_FIELDS)
    if not text:
        return out
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped:
            continue
        label, sep, value = stripped.partition(":")
        if sep:
            label, value = label.strip(), value.strip()
            if label in ("Device Model", "Model Number"):
                if value:
                    out["model"] = value
            elif label == "Serial Number":
                if value:
                    out["serial"] = value
            elif label == "Temperature":  # NVMe health log: `38 Celsius`
                number = _smart_number(value.split()[0]) if value.split() else None
                if number is not None:
                    out["temp_c"] = number
            elif label == "Power On Hours":
                out["power_on_hours"] = _smart_number(value)
            elif label == "Percentage Used":
                out["percentage_used"] = _smart_number(value.rstrip("%"))
            elif label == "Media and Data Integrity Errors" and out["offline_uncorrectable"] is None:
                out["offline_uncorrectable"] = _smart_number(value)
        health = _SMART_HEALTH_RE.match(stripped)
        if health:
            out["overall_health"] = health.group(1).rstrip("!")
            continue
        fields = stripped.split()
        if fields and fields[0].isdigit() and len(fields) >= 6:
            attribute = _SMART_ATTR_FIELDS.get(fields[1])
            if attribute is not None:
                # ID NAME FLAG VALUE WORST THRESH TYPE UPDATED WHEN_FAILED RAW_VALUE —
                # the raw is column 9, and reading the *last* token instead loses every
                # drive that annotates it: `35 (Min/Max 34/44)` ends in `44)`.
                number = _smart_raw_number(fields[9] if len(fields) > 9 else fields[-1])
                if number is not None:
                    out[attribute] = number
    return out


# /proc/<pid>/stat comm is in parens and may itself contain spaces and parens.
_STAT_COMM_RE = re.compile(r"^\d+ \((?P<comm>.*)\) (?P<rest>.*)$", re.S)


def _split_stat(stat: str) -> list[str] | None:
    match = _STAT_COMM_RE.match(stat.strip())
    if not match:
        return None
    return [match.group("comm"), *match.group("rest").split()]


# --------------------------------------------------------------------------- #
# TLS bundle: what the dashboard signs, what the installer places
# --------------------------------------------------------------------------- #

#: The two metadata lines inside a bundle, spelled exactly as `src/dashboards/certs.py` writes
#: them. The agent stays stdlib (no `cryptography` on a monitored host), so the format is agreed
#: here by text rather than imported, and `tests/test_agent.py` parses a bundle that `certs.py`
#: actually produced — the only way the two halves can stay honest.
BUNDLE_REV_LINE = "# BUNDLE-REV "
REVOKED_LINE = "# REVOKED "

#: Where the bundle is unpacked when systemd gave the unit a `RuntimeDirectory` (task 06): tmpfs,
#: private to the service user, gone when the unit stops. Otherwise a `mkdtemp` directory that is
#: removed as soon as the SSL context holds the material.
BUNDLE_RUNTIME_DIR = Path("/run/dashboards-agent")

_PEM_BLOCK_RE = re.compile(r"-----BEGIN [A-Z0-9 ]+-----.*?-----END [A-Z0-9 ]+-----\n", re.DOTALL)
_CERT_HEADER = "-----BEGIN CERTIFICATE-----"
#: The unpacked file names, in the order `_bundle_blocks` hands the blocks back.
_BUNDLE_FILES = ("agent-cert.pem", "agent-key.pem", "root-ca.pem")


def _line_value(line: str, prefix: str) -> str:
    """The first token after a bundle line's prefix, or "" when the line carries nothing."""
    rest = line[len(prefix):].strip()
    fields = rest.split()
    return fields[0] if fields else ""


def bundle_revision(text: str) -> int | None:
    """The `# BUNDLE-REV <n>` number, or None when the line is absent or is not a number.

    Absent is an error rather than a default of 0: the revision is how the dashboard tells a stale
    bundle from the one it just issued, so an agent serving "rev 0" would be reporting a number it
    never had.
    """
    for line in text.splitlines():
        stripped = line.strip()
        if not stripped.startswith(BUNDLE_REV_LINE):
            continue
        token = _line_value(stripped, BUNDLE_REV_LINE)
        try:
            return int(token)
        except ValueError:
            return None
    return None


def bundle_revoked(text: str) -> frozenset[str]:
    """Every `# REVOKED <serial>` value in a bundle, kept exactly as written."""
    serials: set[str] = set()
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith(REVOKED_LINE):
            serial = _line_value(stripped, REVOKED_LINE)
            if serial:
                serials.add(serial)
    return frozenset(serials)


def _bundle_blocks(text: str) -> tuple[str, str, str]:
    """(agent certificate, agent key, root certificate) — positionally, because that is the format.

    Two certificates and exactly one private key is the shape of a bundle. A bundle with the root
    key in it is the failure this catches: an exporter needs to *verify* its clients, which is a
    public-key operation, so a stolen agent install must never be a stolen CA.
    """
    blocks = _PEM_BLOCK_RE.findall(text)
    certs = [block for block in blocks if block.startswith(_CERT_HEADER)]
    keys = [block for block in blocks if not block.startswith(_CERT_HEADER)]
    if len(certs) != 2 or len(keys) != 1:
        raise ValueError("expected the agent certificate, the agent key and the root certificate")
    return certs[0], keys[0], certs[1]


def _file_stamp(path: str) -> tuple[int, int] | None:
    """(mtime_ns, size) for the denylist re-read, or None when the file cannot be stated.

    Nanoseconds rather than the float `st_mtime`: two writes in the same millisecond are a real
    redeploy, and a change this agent misses is a revoked certificate that keeps working.
    """
    try:
        info = os.stat(path)
    except OSError:
        return None
    return (info.st_mtime_ns, info.st_size)


def _material_dir() -> Path:
    """The private directory to unpack into: the unit's RuntimeDirectory when usable, else tmp."""
    if BUNDLE_RUNTIME_DIR.is_dir() and os.access(BUNDLE_RUNTIME_DIR, os.W_OK | os.X_OK):
        return BUNDLE_RUNTIME_DIR
    directory = Path(tempfile.mkdtemp(prefix="dashboards-agent-"))
    os.chmod(directory, 0o700)  # mkdtemp is 0700 already; the claim is this agent's, not the umask's
    return directory


def _unlink(path: Path) -> None:
    with contextlib.suppress(FileNotFoundError):
        path.unlink()


def _write_private(path: Path, text: str) -> None:
    """Create `path` at 0600 in one `open()`, so a private key never exists at the umask default."""
    _unlink(path)  # O_EXCL would refuse a leftover from a killed run, and a symlink would be worse
    handle = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(handle, "w", encoding="ascii") as stream:
        stream.write(text)


def resolve_bundle(path: str) -> tuple[ssl.SSLContext | None, int, frozenset[str], str | None]:
    """(context, bundle_rev, revoked serials, error) for one installed bundle.

    The `resolve_token` idiom: no globals, no server state, every failure an error string the
    caller prints and exits on. A bundle this agent cannot load must stop the startup rather than
    answer plain HTTP behind a token nobody configured — "degraded to no TLS" is the one answer
    that is worse than a crash.

    The three PEM blocks are grepped out by text and written into a private directory, because
    `ssl` loads a chain from paths; they are unlinked as soon as the context has them, so the key
    then lives in this process's memory and nowhere else (and there is no shutdown path that could
    forget to remove it). The root CA goes in as the *verify* list, which is the whole of what an
    exporter needs: it authenticates its clients, it is never authenticated by them.
    """
    try:
        text = Path(path).read_text(encoding="utf-8", errors="replace")
    except OSError as exc:
        return None, 0, frozenset(), f"cannot read --cert-bundle {path}: {exc}"
    if not text.strip():
        return None, 0, frozenset(), f"--cert-bundle {path} is empty"
    revision = bundle_revision(text)
    if revision is None:
        return None, 0, frozenset(), (
            f"--cert-bundle {path} has no readable `# BUNDLE-REV <n>` line — the revision is how "
            "the dashboard tells this bundle from a stale one, so a bundle without it is not one "
            "this app signed")
    try:
        server_cert, server_key, root_cert = _bundle_blocks(text)
    except ValueError as exc:
        return None, 0, frozenset(), f"--cert-bundle {path} is not a bundle: {exc}"
    directory: Path | None = None
    ephemeral = False
    try:
        directory = _material_dir()
        ephemeral = directory != BUNDLE_RUNTIME_DIR
        written = [directory / name for name in _BUNDLE_FILES]
        for target, body in zip(written, (server_cert, server_key, root_cert), strict=True):
            _write_private(target, body)
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        context.load_cert_chain(str(written[0]), str(written[1]))
        context.load_verify_locations(str(written[2]))
        # Optional at the handshake, enforced by the handler: `CERT_REQUIRED` would make the
        # installer's `--status` and systemd's probe fail, because neither holds a client
        # certificate the installer was never given, and a service that cannot health-check is a
        # dead install. See `Handler._certificate_allows` for who may read what.
        context.verify_mode = ssl.CERT_OPTIONAL
    except (OSError, ValueError, ssl.SSLError) as exc:
        return None, 0, frozenset(), f"cannot load --cert-bundle {path}: {exc}"
    finally:
        if directory is not None:
            for name in _BUNDLE_FILES:
                _unlink(directory / name)
            if ephemeral:
                shutil.rmtree(directory, ignore_errors=True)
    return context, revision, bundle_revoked(text), None


class TlsBundle:
    """One loaded bundle: the SSL context, the revision it carries, and the live denylist.

    The context and the revision are fixed for the process's life — `ssl` cannot swap a chain
    under a live socket, so a new certificate needs a restart (and says so on `/certs`). The
    `# REVOKED` list is not fixed: re-reading one small text file when its stamp changes costs
    nothing per poll and is the difference between "revoked" and "revoked on the next deploy",
    which is the only reason the lines live in the bundle instead of in the dashboard alone.
    """

    def __init__(self, path: str, context: ssl.SSLContext, bundle_rev: int,
                 revoked: frozenset[str]) -> None:
        self.path = path
        self.context = context
        self.bundle_rev = bundle_rev
        self._lock = threading.Lock()  # ThreadingHTTPServer may re-read on several requests
        self._revoked = revoked
        self._stamp = _file_stamp(path)

    @classmethod
    def load(cls, path: str) -> tuple[TlsBundle | None, str | None]:
        """`resolve_bundle` plus the file stamp the denylist re-read compares against."""
        context, revision, revoked, error = resolve_bundle(path)
        if error is not None or context is None:
            return None, error
        return cls(path, context, revision, revoked), None

    def revoked_serials(self) -> frozenset[str]:
        """The denylist as the file holds it right now, re-read only when the file changed."""
        stamp = _file_stamp(self.path)
        with self._lock:
            if stamp is None or stamp == self._stamp:
                return self._revoked
            try:
                text = Path(self.path).read_text(encoding="utf-8", errors="replace")
            except OSError:
                # Unreadable mid-redeploy: keep the list this agent already had. Forgetting a
                # revocation because a file vanished for one second is the wrong direction.
                return self._revoked
            self._revoked = bundle_revoked(text)
            self._stamp = stamp
            return self._revoked


def _peer_certificate(connection: Any) -> dict[str, Any]:
    """The verified client certificate of this connection, or {} when there is none.

    `getpeercert()` is only there on a TLS socket and only answers after a handshake, so a
    plain-HTTP connection to a bundle-configured agent (a scanner, a port probe) reads as "no
    certificate" and gets the same 403 as a certless TLS client rather than a traceback.
    """
    getter = getattr(connection, "getpeercert", None)
    if getter is None:
        return {}
    try:
        peer = getter()
    except (OSError, ValueError):  # ValueError is `ssl` for "the handshake has not run"
        return {}
    return peer or {}


def _serial_forms(raw: str) -> frozenset[str]:
    """Every spelling one certificate serial can be written in.

    `getpeercert()` renders a serial as **hexadecimal** text while `certs.py` writes the **decimal**
    number into the bundle's `# REVOKED` line, and an agent that compared one spelling to the other
    would accept a revoked client forever — a revocation that silently does nothing is worse than
    no revocation list, because the operator believes it happened. Accepting both spellings on both
    sides can only ever deny a certificate that is not revoked (two different serials whose hex and
    decimal renderings coincide), never allow one that is.
    """
    token = (raw or "").strip()
    if not token:
        return frozenset()
    forms = {token}
    for base in (10, 16):
        try:
            value = int(token, base)
        except ValueError:
            continue
        forms.add(str(value))
        forms.add(format(value, "X"))
    return frozenset(forms)


def serial_is_revoked(cert: dict[str, Any], revoked: frozenset[str]) -> bool:
    """Is this peer certificate on the bundle's denylist?"""
    peer = _serial_forms(str(cert.get("serialNumber") or ""))
    if not peer:
        return False
    return any(_serial_forms(serial) & peer for serial in revoked)


# --------------------------------------------------------------------------- #
# HTTP server
# --------------------------------------------------------------------------- #
STATUS_PAGE = """<!doctype html>
<meta charset=utf-8><title>{hostname} — host agent</title>
<style>
 body{{font:14px/1.5 ui-monospace,SFMono-Regular,Menlo,monospace;background:#0f1115;color:#e6e8ec;
       margin:2rem;max-width:46rem}}
 h1{{font-size:1.1rem;letter-spacing:.02em}} a{{color:#f7931e}}
 table{{border-collapse:collapse}} td{{padding:.15rem .8rem .15rem 0}}
 .k{{color:#8b93a7}} code{{background:#1a1e26;padding:.1rem .35rem;border-radius:4px}}
</style>
<h1>{hostname}</h1>
<table>
 <tr><td class=k>agent</td><td>v{version}</td></tr>
 <tr><td class=k>os</td><td>{os_pretty}</td></tr>
 <tr><td class=k>kernel</td><td>{kernel}</td></tr>
 <tr><td class=k>cpu</td><td>{cpu_model}</td></tr>
 <tr><td class=k>uptime</td><td>{uptime}</td></tr>
</table>
<p><a href="/snapshot">/snapshot</a> · <a href="/info">/info</a> ·
   <a href="/health">/health</a> · <a href="/schema">/schema</a></p>
"""


class Handler(BaseHTTPRequestHandler):
    server_version = f"host_agent/{AGENT_VERSION}"
    collector: Collector
    token: str | None = None
    bundle: TlsBundle | None = None

    def do_GET(self) -> None:  # the BaseHTTPRequestHandler API name — keep it
        path = self.path.split("?", 1)[0].rstrip("/") or "/"
        bundle = self.bundle
        if bundle is not None:
            # The handshake is the auth on this path, so the bearer check below never runs: a
            # bundle *and* a token would let a stolen token read a host whose certificate the
            # operator just revoked, which is the one thing the denylist exists to prevent.
            if not self._certificate_allows(bundle, path):
                return
        elif self.token and path != "/health":
            given = self.headers.get("Authorization", "")
            if given != f"Bearer {self.token}":
                self._send(401, {"ok": False, "detail": "missing or bad bearer token"})
                return
        if path == "/":
            self._status_page()
        elif path == "/health":
            self._send(200, {"ok": True, "hostname": socket.gethostname(),
                             "version": AGENT_VERSION, "ts": time.time()})
        elif path == "/info":
            self._send(200, {"ok": True, "info": self.collector.host_info(),
                             "gpus": self.collector.gpus()})
        elif path == "/schema":
            self._send(200, {"ok": True, "schema": SCHEMA_VERSION, "doc": __doc__})
        elif path == "/snapshot":
            query = self.path.split("?", 1)[1] if "?" in self.path else ""
            top = None
            for part in query.split("&"):
                if part.startswith("top="):
                    try:
                        top = max(0, int(part[4:]))
                    except ValueError:
                        top = None
            body: dict[str, Any] = {"ok": True, **self.collector.snapshot(top=top)}
            if bundle is not None:
                # Additive, and only in TLS mode: the dashboard (task 03) reads the revision to
                # tell a stale installed bundle from the one `/certs` just issued, and an absent
                # field means "this agent predates certificates" rather than "revision 0".
                body["tls"] = True
                body["bundle_rev"] = bundle.bundle_rev
            self._send(200, body)
        else:
            self._send(404, {"ok": False, "detail": f"no such path: {path}"})

    def _certificate_allows(self, bundle: TlsBundle, path: str) -> bool:
        """Does this connection's client certificate authorise `path`? Sends the refusal if not.

        Two answers, because they are two different operator actions: 403 "required" is a client
        that never proved who it is (nothing installed, or the wrong CA), 403 "revoked" is a client
        that proved it and was withdrawn.

        The order is the design. A revoked identity is refused on **every** path, `/health` first:
        the exemption is for the probe that holds the root CA and no client certificate at all, not
        for a certificate the operator has just taken away. Then `/health` passes a certless probe,
        because `install-agent.sh --status` and systemd's probe were never handed a client
        certificate and a service that cannot health-check is a dead install.
        """
        peer = _peer_certificate(self.connection)
        present = bool(peer and peer.get("subject"))
        if present and serial_is_revoked(peer, bundle.revoked_serials()):
            self._send(403, {"ok": False, "detail": "client certificate revoked"})
            return False
        if path == "/health":
            return True
        if not present:
            self._send(403, {"ok": False, "detail": "client certificate required"})
            return False
        return True

    def _status_page(self) -> None:
        info = self.collector.host_info()
        uptime = info.get("uptime_seconds")
        body = STATUS_PAGE.format(
            hostname=info.get("hostname") or "unknown",
            version=AGENT_VERSION,
            os_pretty=info.get("os_pretty") or "unknown",
            kernel=info.get("kernel") or "unknown",
            cpu_model=info.get("cpu_model") or "unknown",
            uptime=f"{uptime / 86400:.1f} days" if uptime else "unknown",
        )
        data = body.encode()
        self.send_response(200)
        self.send_header("content-type", "text/html; charset=utf-8")
        self.send_header("content-length", str(len(data)))
        self.send_header("cache-control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def _send(self, status: int, body: dict[str, Any]) -> None:
        data = json.dumps(body, default=str).encode()
        self.send_response(status)
        self.send_header("content-type", "application/json")
        self.send_header("content-length", str(len(data)))
        self.send_header("cache-control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, fmt: str, *args: Any) -> None:
        # One line per poll, journald-friendly, and never with query strings (a token
        # pasted by mistake in a URL should not live in the journal).
        if "%s" in fmt:
            fmt = fmt % args  # the log_message idiom from BaseHTTPRequestHandler
        print(f"[agent] {self.address_string()} {fmt}", flush=True)


def serve(host: str, port: int, collector: Collector, token: str | None,
          bundle: TlsBundle | None = None) -> None:
    """Serve forever on host:port — plain HTTP with an optional token, or HTTPS with a bundle.

    All three class attributes are assigned on every call: `Handler` is process-wide state, so a
    server started without a bundle must not inherit one from a server that had it (that would
    answer a plain-HTTP socket with 403 on every path, or worse, accept a client certificate the
    operator revoked).
    """
    Handler.collector = collector
    Handler.token = token
    Handler.bundle = bundle
    httpd = ThreadingHTTPServer((host, port), Handler)
    if bundle is not None:
        httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
        print(f"[agent] listening on https://{host}:{port} with tls "
              f"(client cert required, rev {bundle.bundle_rev}); pid {os.getpid()}", flush=True)
    else:
        print(f"[agent] listening on http://{host}:{port} "
              f"({'token required' if token else 'no token'}); pid {os.getpid()}", flush=True)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("[agent] shutting down", flush=True)
    finally:
        httpd.server_close()


def resolve_token(token: str, token_file: str) -> tuple[str | None, str | None]:
    """(token, error). A file wins over a flag because `ps` shows argv to every user on the
    box, and this token guards hostname, mounts, GPU model and process names."""
    if not token_file:
        return (token or None), None
    try:
        value = Path(token_file).read_text(encoding="utf-8").strip()
    except OSError as exc:
        return None, f"cannot read --token-file {token_file}: {exc}"
    if not value:
        return None, f"--token-file {token_file} is empty"
    return value, None


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0],
                                     formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument("--host", default=os.environ.get("AGENT_HOST", "127.0.0.1"),
                        help="bind address; 0.0.0.0 exposes host metrics to the network")
    parser.add_argument("--port", type=int, default=int(os.environ.get("AGENT_PORT", "9120")))
    parser.add_argument("--token", default=os.environ.get("AGENT_TOKEN", ""),
                        help="bearer token required on every request except /health")
    parser.add_argument("--token-file", default=os.environ.get("AGENT_TOKEN_FILE", ""),
                        help="read the bearer token from a file (keeps it out of ps); wins "
                             "over --token")
    parser.add_argument("--cert-bundle", default=os.environ.get("AGENT_CERT_BUNDLE", ""),
                        help="one installed PEM from the dashboard: this agent's server "
                             "certificate and key, the dashboard's root CA, a `# BUNDLE-REV` "
                             "line and the `# REVOKED` denylist. Serves https and answers "
                             "every path but /health only to a client certificate signed by "
                             "that root and not on the list; on that path the token is not read")
    parser.add_argument("--root", default="/", help="filesystem root to read (for tests)")
    parser.add_argument("--top", type=int, default=5, help="processes per top-N list")
    parser.add_argument("--net-all", action="store_true", help="include veth/bridge interfaces")
    parser.add_argument("--smartctl", default="smartctl",
                        help="smartctl binary; missing is a documented degradation "
                             "(every disk's smart is null, one errors entry)")
    parser.add_argument("--no-smart", action="store_true",
                        help="skip the smartctl pass entirely (disks report smart: null, "
                             "no error entries)")
    parser.add_argument("--once", action="store_true", help="print one snapshot as JSON and exit")
    args = parser.parse_args(argv)

    collector = Collector(args.root, top_processes=args.top, net_all=args.net_all,
                          smartctl=args.smartctl, smart=not args.no_smart)
    if args.once:
        print(json.dumps({"ok": True, **collector.snapshot()}, indent=2, default=str))
        return 0
    token, error = resolve_token(args.token, args.token_file)
    if error:
        print(f"[agent] {error}", flush=True)
        return 2
    bundle: TlsBundle | None = None
    if args.cert_bundle:
        bundle, bundle_error = TlsBundle.load(args.cert_bundle)
        if bundle_error is not None:
            print(f"[agent] {bundle_error}", flush=True)
            return 2
    if args.host not in ("127.0.0.1", "localhost", "::1") and not token and bundle is None:
        print("[agent] refusing to bind off-loopback without --token or --cert-bundle (host "
              "metrics are sensitive); pass --token, --token-file, --cert-bundle or set "
              "AGENT_TOKEN / AGENT_CERT_BUNDLE", flush=True)
        return 2
    serve(args.host, args.port, collector, token, bundle)
    return 0


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