FabledSteward/plugins/host_agent/agent.py

# plugins/host_agent/agent.py
"""Steward host agent — pushes resource metrics to a Steward instance.

Python 3.8+ stdlib only. Target ~300 lines. Served to targets at
GET /plugins/host_agent/agent.py.
"""
from __future__ import annotations

import json
import os
import re
import shutil
import signal
import socket
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
from collections import deque
from concurrent.futures import ThreadPoolExecutor, wait as wait_futures
from datetime import datetime, timezone

AGENT_VERSION = "1.22.0"

# Default path to the local Docker Engine socket. Overridable via the
# `docker_socket` config key; collection is silently skipped if it's absent or
# unreadable, so this never needs setting on a Docker-less host (zero-config).
DEFAULT_DOCKER_SOCKET = "/var/run/docker.sock"


class ConfigError(Exception):
    pass


REQUIRED_KEYS = ("url", "token")
INT_KEYS = ("interval_seconds",)
LIST_KEYS = ("mounts", "docker_log_exclude")
# Truthy strings for bool keys; anything else (incl. empty) is False.
BOOL_KEYS = ("docker_logs_enabled",)
_BOOL_TRUE = ("1", "true", "yes", "on")


def read_config(path: str) -> dict:
    """Parse a flat `key = value` config file."""
    cfg: dict = {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            for lineno, raw in enumerate(f, 1):
                line = raw.strip()
                if not line or line.startswith("#"):
                    continue
                if "=" not in line:
                    raise ConfigError(f"{path}:{lineno}: expected 'key = value'")
                key, _, value = line.partition("=")
                key = key.strip()
                value = value.strip()
                if key in INT_KEYS:
                    try:
                        cfg[key] = int(value)
                    except ValueError:
                        raise ConfigError(f"{path}:{lineno}: {key} must be int")
                elif key in LIST_KEYS:
                    cfg[key] = [v.strip() for v in value.split(",") if v.strip()]
                elif key in BOOL_KEYS:
                    cfg[key] = value.lower() in _BOOL_TRUE
                else:
                    cfg[key] = value
    except FileNotFoundError:
        raise ConfigError(f"{path}: not found")

    missing = [k for k in REQUIRED_KEYS if k not in cfg]
    if missing:
        raise ConfigError(f"{path}: missing required key(s): {', '.join(missing)}")

    cfg.setdefault("interval_seconds", 30)
    cfg.setdefault("docker_socket", DEFAULT_DOCKER_SOCKET)
    # Container-log collection is on by default (operator preference); a host can
    # opt out with `docker_logs_enabled = false` or thin it with docker_log_exclude.
    cfg.setdefault("docker_logs_enabled", True)
    cfg.setdefault("docker_log_exclude", [])
    return cfg


# ─── collectors ──────────────────────────────────────────────────────────────

STAT_PATH = "/proc/stat"
MEMINFO_PATH = "/proc/meminfo"
LOADAVG_PATH = "/proc/loadavg"
UPTIME_PATH = "/proc/uptime"
OS_RELEASE_PATH = "/etc/os-release"
NETDEV_PATH = "/proc/net/dev"
DISKSTATS_PATH = "/proc/diskstats"
MOUNTSTATS_PATH = "/proc/self/mountstats"
HWMON_DIR = "/sys/class/hwmon"
# Where the kernel lists virtual network devices (bridges, veth, docker0,
# vxlan, tun, VLANs, bonds) and the block devices (a disk has a `device`).
SYS_VIRTUAL_NET_DIR = "/sys/devices/virtual/net"
SYS_BLOCK_DIR = "/sys/block"
PSI_DIR = "/proc/pressure"

# Partition names to drop from disk I/O (we report whole-disk throughput only).
_PARTITION_RE = re.compile(
    r"^(?:sd[a-z]+\d+|vd[a-z]+\d+|hd[a-z]+\d+|nvme\d+n\d+p\d+|mmcblk\d+p\d+)$"
)


def _read_file(path: str) -> str:
    with open(path, "r", encoding="utf-8") as f:
        return f.read()


def _parse_cpu_lines(stat_text: str) -> dict:
    """Return {name: (total_jiffies, idle_jiffies)} for every cpu line.

    'cpu' is the aggregate; 'cpu0', 'cpu1', … are per-core.
    """
    out: dict = {}
    for line in stat_text.splitlines():
        if not line.startswith("cpu"):
            continue
        parts = line.split()
        try:
            fields = [int(x) for x in parts[1:]]
        except ValueError:
            continue
        if len(fields) < 4:
            continue
        idle = fields[3] + (fields[4] if len(fields) > 4 else 0)  # idle + iowait
        out[parts[0]] = (sum(fields), idle)
    return out


def collect_cpu(sample_window: float = 0.2) -> tuple[float, list]:
    """Return (aggregate_pct, [per_core_pct, …]) over sample_window seconds."""
    a = _parse_cpu_lines(_read_file(STAT_PATH))
    time.sleep(sample_window)
    b = _parse_cpu_lines(_read_file(STAT_PATH))

    def _pct(name: str):
        if name not in a or name not in b:
            return None
        dt = b[name][0] - a[name][0]
        di = b[name][1] - a[name][1]
        if dt <= 0:
            return 0.0
        return round(100.0 * (dt - di) / dt, 2)

    agg = _pct("cpu") or 0.0
    cores: list = []
    i = 0
    while f"cpu{i}" in b:
        cores.append(_pct(f"cpu{i}"))
        i += 1
    return agg, cores


def collect_memory() -> dict:
    info: dict[str, int] = {}
    for line in _read_file(MEMINFO_PATH).splitlines():
        if ":" not in line:
            continue
        key, _, rest = line.partition(":")
        parts = rest.strip().split()
        if not parts:
            continue
        try:
            value_kb = int(parts[0])
        except ValueError:
            continue
        info[key] = value_kb * 1024  # bytes
    total = info.get("MemTotal", 0)
    available = info.get("MemAvailable", 0)
    swap_total = info.get("SwapTotal", 0)
    swap_free = info.get("SwapFree", 0)
    return {
        "total_bytes": total,
        "used_bytes": max(total - available, 0),
        "available_bytes": available,
        "swap_used_bytes": max(swap_total - swap_free, 0),
        "cached_bytes": info.get("Cached", 0),
        "buffers_bytes": info.get("Buffers", 0),
    }


# A mount that hasn't answered in this long is reported stalled (#5760): a
# hard-mounted NFS share whose server stops blocks statvfs in the kernel with
# no timeout, and read inline that froze the whole agent, silent for a day on
# a host that still answered ping.
MOUNT_DEADLINE_SECONDS = 5.0
# Mounts whose last probe is still blocked. They aren't probed again until it
# returns, so a dead share costs one stuck thread, not one per sample.
_stuck_mounts: dict = {}


def _read_mount(m: str) -> dict:
    u = shutil.disk_usage(m)
    row = {"mount": m, "total_bytes": u.total, "used_bytes": u.used}
    # Inodes too (1.13.0): a filesystem can run out of them with space
    # to spare, and every create then fails with ENOSPC. Some (btrfs)
    # report none, and are left without.
    try:
        st = os.statvfs(m)
    except OSError:
        st = None
    if st is not None and st.f_files:
        row["inodes_total"] = st.f_files
        row["inodes_used"] = st.f_files - st.f_ffree
    return row


def collect_storage(mounts: list[str], deadline: float = MOUNT_DEADLINE_SECONDS) -> list[dict]:
    """Each mount's space and inodes, read side by side in daemon threads under
    one deadline; a mount that doesn't answer in time is {"mount", "stalled"}."""
    probes: dict = {}
    for m in mounts:
        stuck = _stuck_mounts.get(m)
        if stuck is not None and stuck.is_alive():
            continue
        _stuck_mounts.pop(m, None)
        box: dict = {}

        def probe(m=m, box=box):
            try:
                box["row"] = _read_mount(m)
            except OSError:
                pass       # gone, not ours to read, or a stale handle: left out
        t = threading.Thread(target=probe, name=f"mount {m}", daemon=True)
        t.start()
        probes[m] = (t, box)
    end = time.monotonic() + deadline
    out: list[dict] = []
    for m in mounts:
        if m not in probes:
            out.append({"mount": m, "stalled": True})
            continue
        t, box = probes[m]
        t.join(max(0.0, end - time.monotonic()))
        if t.is_alive():
            _stuck_mounts[m] = t
            out.append({"mount": m, "stalled": True})
        elif "row" in box:
            out.append(box["row"])
    return out


def collect_load() -> dict:
    parts = _read_file(LOADAVG_PATH).split()
    return {"1m": float(parts[0]), "5m": float(parts[1]), "15m": float(parts[2])}


def collect_uptime() -> int:
    return int(float(_read_file(UPTIME_PATH).split()[0]))


def detect_primary_ip() -> str | None:
    """Best-effort primary non-loopback IPv4 of this host.

    connect() on a SOCK_DGRAM socket only sets the default peer and selects the
    outbound interface — no packets are sent — so getsockname() reveals the
    source IP the kernel would use to reach the internet. This is the host's own
    view of its address, so it survives reverse proxies / NAT (unlike the
    server reading request.remote_addr). Returns None on any error (offline / no
    route) and skips loopback.
    """
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
    except OSError:
        return None
    finally:
        s.close()
    if not ip or ip.startswith("127."):
        return None
    return ip


def collect_metadata() -> dict:
    u = os.uname()
    distro = "unknown"
    try:
        for line in _read_file(OS_RELEASE_PATH).splitlines():
            if line.startswith("PRETTY_NAME="):
                distro = line.split("=", 1)[1].strip().strip('"')
                break
    except (OSError, FileNotFoundError):
        pass
    return {
        "kernel": u.release,
        "distro": distro,
        "arch": u.machine,
        "host_ip": detect_primary_ip(),
        "maint_helper": maint_helper_installed(),
    }


def default_mounts() -> list[str]:
    """Return real mount points from /proc/mounts, skipping pseudo filesystems."""
    skip_types = {"tmpfs", "devtmpfs", "proc", "sysfs", "cgroup", "cgroup2",
                  "overlay", "squashfs", "ramfs", "devpts", "mqueue", "pstore",
                  "securityfs", "debugfs", "tracefs", "hugetlbfs", "fusectl",
                  "configfs", "bpf", "autofs", "efivarfs", "binfmt_misc"}
    mounts: list[str] = []
    try:
        with open("/proc/mounts", "r") as f:
            for line in f:
                parts = line.split()
                if len(parts) < 3:
                    continue
                if parts[2] in skip_types:
                    continue
                mounts.append(parts[1])
    except OSError:
        return ["/"]
    return mounts or ["/"]


def _physical_only(found: dict, is_virtual) -> dict:
    """The entries the kernel says are physical, so a host total counts each
    byte once (a bridge's or an LVM volume's traffic is also counted on the
    NIC or disk beneath it). When /sys calls every one of them virtual (an
    agent inside LXC, where eth0 is a veth), they are all kept: a host
    reporting no traffic at all would be wrong, not filtered."""
    physical = {k: v for k, v in found.items() if not is_virtual(k)}
    return physical or found


def _virtual_net(name: str) -> bool:
    """Whether the kernel lists this interface as virtual; False if /sys can't say."""
    return os.path.isdir(SYS_VIRTUAL_NET_DIR) and os.path.exists(
        os.path.join(SYS_VIRTUAL_NET_DIR, name))


def _virtual_disk(name: str) -> bool:
    """Whether this block device has no disk behind it (dm, md, zram, nbd);
    False if /sys can't say."""
    return os.path.isdir(SYS_BLOCK_DIR) and not os.path.exists(
        os.path.join(SYS_BLOCK_DIR, name, "device"))


def collect_net_raw() -> dict:
    """Return {iface: (rx_bytes, tx_bytes)} cumulative counters from /proc/net/dev.

    Physical interfaces only (see _physical_only); loopback and veth* are
    skipped by name too, for a host whose /sys can't be read. Counters are
    monotonic; per-second rates are derived from deltas in build_sample().
    """
    out: dict = {}
    try:
        text = _read_file(NETDEV_PATH)
    except OSError:
        return out
    for line in text.splitlines():
        if ":" not in line:
            continue
        name, _, rest = line.partition(":")
        name = name.strip()
        if name == "lo" or name.startswith("veth"):
            continue
        f = rest.split()
        if len(f) < 16:
            continue
        try:
            out[name] = (int(f[0]), int(f[8]))  # rx_bytes, tx_bytes
        except ValueError:
            continue
    return _physical_only(out, _virtual_net)


def collect_diskio_raw() -> dict:
    """Return {device: (read_bytes, write_bytes)} cumulative from /proc/diskstats.

    Physical whole disks only (see _physical_only): loop/ram/sr/fd and
    partitions are skipped by name too. Sectors are 512 bytes. Counters are
    monotonic; rates derived from deltas.
    """
    out: dict = {}
    try:
        text = _read_file(DISKSTATS_PATH)
    except OSError:
        return out
    for line in text.splitlines():
        f = line.split()
        if len(f) < 10:
            continue
        name = f[2]
        if name.startswith(("loop", "ram", "sr", "fd")) or _PARTITION_RE.match(name):
            continue
        try:
            out[name] = (int(f[5]) * 512, int(f[9]) * 512)  # sectors_read, sectors_written
        except ValueError:
            continue
    return _physical_only(out, _virtual_disk)


# ─── NFS delay per mount (#5996) ─────────────────────────────────────────────
#
# The kernel's own count of every NFS request each mount has finished, read
# from /proc/self/mountstats. Reading it never touches the share, so it answers
# while one is stuck. Each op's line is cumulative: ops, transmissions, major
# timeouts, bytes sent, bytes received, then queue, round-trip and execute
# time in ms. A request is counted when it finishes, so a share that is stuck
# shows nothing until it comes back, and then its long requests at once
# (mounts_stalled says it is stuck meanwhile).


def read_mountstats(text: str) -> dict:
    """{mount point: {op: (ops, transmissions, timeouts, queue_ms, rtt_ms,
    execute_ms)}} for each NFS mount in a mountstats text."""
    out: dict = {}
    mount = None
    in_ops = False
    for line in text.splitlines():
        if line.startswith("device "):
            mount, in_ops = None, False
            parts = line.split()
            try:
                at = parts.index("on")
                fstype = parts[parts.index("fstype") + 1]
            except (ValueError, IndexError):
                continue
            if fstype.startswith("nfs") and at + 1 < len(parts):
                mount = parts[at + 1].replace("\\040", " ")
                out[mount] = {}
        elif mount is None:
            continue
        elif line.strip() == "per-op statistics":
            in_ops = True
        elif in_ops:
            name, sep, rest = line.strip().partition(":")
            nums = rest.split()
            if sep and len(nums) >= 8 and all(n.isdigit() for n in nums[:8]):
                ops, trans, timeouts, _sent, _recv, queue, rtt, exe = (int(n) for n in nums[:8])
                out[mount][name] = (ops, trans, timeouts, queue, rtt, exe)
    return out


def collect_nfs_raw() -> dict:
    try:
        return read_mountstats(_read_file(MOUNTSTATS_PATH))
    except OSError:
        return {}


def nfs_delays(cur: dict, prev: dict, dt: float) -> list:
    """Each NFS mount's requests since the last sample: how many a second,
    their average round trip and execute time (queueing included, what a
    program waits), READ and WRITE on their own, and the retransmissions and
    major timeouts among them. A mount new since the last sample, or whose
    counters went back (remounted), is left for the next one."""
    out: list = []
    if dt <= 0:
        return out
    for mount, ops in cur.items():
        before = prev.get(mount)
        if before is None:
            continue
        d = {op: tuple(a - b for a, b in zip(v, before.get(op, (0,) * 6))) for op, v in ops.items()}
        if any(x < 0 for v in d.values() for x in v):
            continue
        n = sum(v[0] for v in d.values())
        row = {"mount": mount, "ops_per_sec": round(n / dt, 2),
               "retrans": max(0, sum(v[1] - v[0] for v in d.values())),
               "timeouts": sum(v[2] for v in d.values())}
        if n:
            row["rtt_ms"] = round(sum(v[4] for v in d.values()) / n, 2)
            row["exe_ms"] = round(sum(v[5] for v in d.values()) / n, 2)
        for op, key in (("READ", "read_ms"), ("WRITE", "write_ms")):
            v = d.get(op)
            if v and v[0]:
                row[key] = round(v[5] / v[0], 2)
        out.append(row)
    return out


def collect_temps() -> list:
    """Return [{label, celsius}] from /sys/class/hwmon (best-effort, may be empty)."""
    out: list = []
    try:
        chips = os.listdir(HWMON_DIR)
    except OSError:
        return out
    for hw in chips:
        d = os.path.join(HWMON_DIR, hw)
        chip = ""
        try:
            chip = _read_file(os.path.join(d, "name")).strip()
        except OSError:
            pass
        try:
            files = os.listdir(d)
        except OSError:
            continue
        for fn in files:
            if not (fn.startswith("temp") and fn.endswith("_input")):
                continue
            idx = fn[: -len("_input")]  # e.g. "temp1"
            try:
                milli = int(_read_file(os.path.join(d, fn)).strip())
            except (OSError, ValueError):
                continue
            label = ""
            try:
                label = _read_file(os.path.join(d, idx + "_label")).strip()
            except OSError:
                pass
            name = label or (f"{chip}_{idx}" if chip else idx)
            out.append({"label": name, "celsius": round(milli / 1000.0, 1)})
    return out


def _parse_psi(text: str) -> dict:
    """Parse a /proc/pressure/* file into {some_avg10, some_avg60, full_avg10, …}."""
    res: dict = {}
    for line in text.splitlines():
        parts = line.split()
        if not parts:
            continue
        kind = parts[0]  # "some" | "full"
        for p in parts[1:]:
            for prefix, suffix in (("avg10=", "avg10"), ("avg60=", "avg60")):
                if p.startswith(prefix):
                    try:
                        res[f"{kind}_{suffix}"] = float(p[len(prefix):])
                    except ValueError:
                        pass
    return res


def collect_psi() -> dict:
    """Return pressure-stall (PSI) gauges, prefixed by resource (mem/cpu/io).

    Absent on kernels without CONFIG_PSI — returns only what exists.
    """
    out: dict = {}
    for res_name, fname in (("mem", "memory"), ("cpu", "cpu"), ("io", "io")):
        try:
            text = _read_file(os.path.join(PSI_DIR, fname))
        except OSError:
            continue
        for k, v in _parse_psi(text).items():
            out[f"{res_name}_{k}"] = v
    return out


def _rates(cur: dict, prev: dict, dt: float) -> dict:
    """Per-second rates for counters present in both snapshots over dt seconds.

    Skips names missing from prev and negative deltas (counter reset / iface or
    disk hot-plug), so a reboot never emits a bogus spike.
    """
    out: dict = {}
    if dt <= 0:
        return out
    for name, cur_vals in cur.items():
        if name not in prev:
            continue
        prev_vals = prev[name]
        deltas = [c - p for c, p in zip(cur_vals, prev_vals)]
        if any(d < 0 for d in deltas):
            continue
        out[name] = tuple(d / dt for d in deltas)
    return out


# ─── docker ──────────────────────────────────────────────────────────────────
#
# Per-host container collection over the local Docker socket. stdlib-only: a
# minimal HTTP/1.1 client over an AF_UNIX socket. We send `Connection: close`
# so the daemon closes the socket at end of response and we can read to EOF;
# Docker still frames the body with Transfer-Encoding: chunked, so we de-chunk
# when that header is present rather than assume Content-Length.

DOCKER_API_TIMEOUT = 5.0
# All of one sample's per-container Docker work (inspect + stats) must finish
# inside this; past it the sample says Docker did not answer in time rather
# than holding up the host's own metrics (#5154).
DOCKER_DEADLINE_SECONDS = 20.0
# /system/df walks every image, container and volume: asked this often, not
# every sample. Steward keeps the last answer between reports.
DOCKER_DF_INTERVAL_SECONDS = 600.0

# Container-log collection (m79). Logs ride in the same push as metrics, so one
# interval's batch is capped to keep a chatty container from bloating a POST (and
# the backoff ring): once the cap is hit a marker line is emitted and collection
# stops — the deferred lines come on the next interval. On a container's first
# interval we seed from a short tail, then switch to an incremental since-cursor
# kept per container in the agent's rate-state.
DOCKER_LOG_BATCH_MAX_BYTES = 262144   # 256 KiB of log text per push, all containers
DOCKER_LOG_FIRST_TAIL = 50            # lines seeded on a container's first interval
DOCKER_LOG_FINAL_TAIL = 200           # lines read once from a container found already crashed
DOCKER_LOG_FINAL_WINDOW_SECONDS = 600  # ... when it stopped this recently
# A container that wrote more than this since its cursor is read from its last
# DOCKER_LOG_MAX_LINES lines, and the older ones are skipped, not re-read every
# interval (#5764). One log answer over DOCKER_LOG_RESPONSE_MAX_BYTES is dropped
# and the container skipped ahead to now.
DOCKER_LOG_MAX_LINES = 2000
DOCKER_LOG_RESPONSE_MAX_BYTES = 4 * 1024 * 1024
# All of one sample's log reads must finish inside this; the containers not
# reached are read next interval.
DOCKER_LOG_DEADLINE_SECONDS = 10.0
# Any one Docker answer: past this size or this long, it is cut off rather than
# waited on. A body that keeps trickling in never reaches EOF, and a per-read
# timeout alone would wait on it forever (#5764).
DOCKER_RESPONSE_MAX_BYTES = 32 * 1024 * 1024
DOCKER_RESPONSE_DEADLINE_SECONDS = 15.0


class DockerAnswerTooLarge(OSError):
    """A Docker answer longer than the reader would hold."""


def _dechunk(body: bytes) -> bytes:
    """Decode an HTTP/1.1 chunked-transfer body into the raw payload."""
    out = bytearray()
    while body:
        line, sep, rest = body.partition(b"\r\n")
        if not sep:
            break
        try:
            size = int(line.split(b";", 1)[0], 16)  # ignore chunk extensions
        except ValueError:
            break
        if size == 0:
            break
        out += rest[:size]
        body = rest[size + 2:]  # skip the chunk data and its trailing CRLF
    return bytes(out)


def _docker_get(socket_path: str, path: str, accept: str, timeout: float,
                deadline: float, max_bytes: int) -> tuple:
    """GET `path` from the Docker Engine API over a Unix socket.

    Returns (status, body), the body de-chunked. Each read waits at most
    `timeout`, the whole answer at most `deadline`, and an answer longer than
    `max_bytes` raises DockerAnswerTooLarge without being held whole: a slow or
    endless body is cut off, never waited on (#5764). Raises OSError on any
    transport problem.
    """
    end = time.monotonic() + deadline
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(min(timeout, deadline))
    try:
        sock.connect(socket_path)
        req = (
            "GET " + path + " HTTP/1.1\r\n"
            "Host: docker\r\n"
            "Accept: " + accept + "\r\n"
            "Connection: close\r\n"
            "\r\n"
        )
        sock.sendall(req.encode("ascii"))
        chunks = []
        size = 0
        while True:
            left = end - time.monotonic()
            if left <= 0:
                raise socket.timeout(f"docker API {path} still answering after {deadline:g} s")
            sock.settimeout(min(timeout, left))
            buf = sock.recv(65536)
            if not buf:
                break
            size += len(buf)
            if size > max_bytes:
                raise DockerAnswerTooLarge(f"docker API {path} answered more than {max_bytes} bytes")
            chunks.append(buf)
    finally:
        sock.close()

    raw = b"".join(chunks)
    head, _, body = raw.partition(b"\r\n\r\n")
    header_text = head.decode("latin-1")
    status_line = header_text.split("\r\n", 1)[0]
    parts = status_line.split(None, 2)  # "HTTP/1.1 200 OK"
    status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
    if "transfer-encoding: chunked" in header_text.lower():
        body = _dechunk(body)
    return status, body


def _docker_request(socket_path: str, path: str, timeout: float = DOCKER_API_TIMEOUT):
    """GET `path` from the Docker Engine API over a Unix socket; return JSON.

    Raises OSError on any socket/transport problem (absent socket, permission
    denied, non-200, an answer too slow or too large) and ValueError on a
    non-JSON body — both are caught by the caller and treated as "no docker
    here", so collection degrades silently.
    """
    status, body = _docker_get(socket_path, path, "application/json", timeout,
                               DOCKER_RESPONSE_DEADLINE_SECONDS, DOCKER_RESPONSE_MAX_BYTES)
    if status != 200:
        raise OSError(f"docker API {path} returned {status}")
    return json.loads(body.decode("utf-8"))


def _docker_cpu_pct(stats: dict) -> float:
    """CPU % from a Docker stats snapshot (delta of container vs system CPU).

    Ported verbatim from the old central scraper's math; correct as long as both
    cpu_stats and precpu_stats are populated (we request without `one-shot` so
    the daemon fills precpu over one cycle).
    """
    try:
        cpu = stats.get("cpu_stats", {})
        precpu = stats.get("precpu_stats", {})
        cpu_delta = (
            cpu["cpu_usage"]["total_usage"] - precpu["cpu_usage"]["total_usage"]
        )
        sys_delta = cpu.get("system_cpu_usage", 0) - precpu.get("system_cpu_usage", 0)
        num_cpus = cpu.get("online_cpus") or len(
            cpu["cpu_usage"].get("percpu_usage") or [None]
        )
        if sys_delta <= 0 or cpu_delta < 0:
            return 0.0
        return round((cpu_delta / sys_delta) * num_cpus * 100.0, 2)
    except (KeyError, TypeError, ZeroDivisionError):
        return 0.0


def _docker_mem(stats: dict):
    """Return (usage_bytes, limit_bytes, mem_pct); working set excludes cache."""
    try:
        mem = stats.get("memory_stats", {})
        usage = mem.get("usage", 0)
        limit = mem.get("limit", 0)
        # Subtract page cache for working set (cgroup v1 "cache", v2 "inactive_file").
        cache = mem.get("stats", {}).get("cache", 0) or \
                mem.get("stats", {}).get("inactive_file", 0)
        actual = max(0, usage - cache)
        pct = round(actual / limit * 100.0, 2) if limit > 0 else 0.0
        return actual, limit, pct
    except (KeyError, TypeError):
        return 0, 0, 0.0


def _docker_ports(ports: list) -> list:
    """Normalise Docker port bindings to a compact published-ports list."""
    out = []
    for p in (ports or []):
        if p.get("PublicPort"):
            out.append({
                "host_port": p["PublicPort"],
                "container_port": p.get("PrivatePort"),
                "protocol": p.get("Type", "tcp"),
            })
    return out


def _docker_grouping(labels: dict) -> dict:
    """Pull compose project + swarm service/task/node out of container labels.

    These are set by Docker itself (compose / swarm), so reading them off the
    container costs nothing extra and lets Steward group + place containers
    without a manager-node query.
    """
    labels = labels or {}
    return {
        "compose_project": labels.get("com.docker.compose.project"),
        "service_name": labels.get("com.docker.swarm.service.name"),
        "task_id": labels.get("com.docker.swarm.task.id"),
        "node_id": labels.get("com.docker.swarm.node.id"),
    }


def _docker_inspect_fields(insp: dict):
    """Return (health, restart_count, exit_code, oom_killed) from an inspect.

    health is None for containers without a HEALTHCHECK (no State.Health).
    """
    state = (insp or {}).get("State") or {}
    health = (state.get("Health") or {}).get("Status")  # healthy|unhealthy|starting
    restart_count = insp.get("RestartCount", 0) if isinstance(insp, dict) else 0
    exit_code = state.get("ExitCode")
    oom_killed = bool(state.get("OOMKilled", False))
    return health, restart_count, exit_code, oom_killed


def _docker_started_at(insp: dict, key: str = "StartedAt"):
    """State.StartedAt as ISO-8601, or None for a container never started
    (Docker's zero time) or an inspect that failed. When the container last
    started, not when it was created: a restarted container's uptime counts
    from its restart. key="FinishedAt" reads when it last stopped."""
    raw = (((insp or {}).get("State") or {}).get(key) or "")
    if not raw or raw.startswith("0001-"):
        return None
    ts = _parse_log_ts(raw)
    return ts.isoformat() if ts else None


def _docker_net_io(stats: dict):
    """Return cumulative (net_rx, net_tx, blk_read, blk_write) bytes from stats.

    Counters are cumulative since container start (rates can be derived later);
    block I/O comes from the cgroup io_service_bytes list, absent on some setups.
    """
    net_rx = net_tx = blk_read = blk_write = 0
    for iface in (stats.get("networks") or {}).values():
        net_rx += iface.get("rx_bytes", 0) or 0
        net_tx += iface.get("tx_bytes", 0) or 0
    for e in ((stats.get("blkio_stats") or {}).get("io_service_bytes_recursive") or []):
        op = (e.get("op") or "").lower()
        if op == "read":
            blk_read += e.get("value", 0) or 0
        elif op == "write":
            blk_write += e.get("value", 0) or 0
    return net_rx, net_tx, blk_read, blk_write


def _collect_one_container(socket_path: str, c: dict) -> dict:
    """Build one container's enriched record (runs in a worker thread).

    Does a per-container inspect (health/restart/exit/oom — only available there)
    plus a stats read for running containers (cpu/mem/net/io). All best-effort:
    a failed call just leaves the affected fields at their defaults.
    """
    cid = c.get("Id", "") or ""
    names = c.get("Names") or []
    name = names[0].lstrip("/") if names else cid[:12]
    state = c.get("State", "unknown")

    # Inspect every container (incl. stopped) — exit codes + restart counts only
    # live here, and stopped containers are exactly where exit_code matters.
    health = exit_code = insp = None
    restart_count = 0
    oom_killed = False
    try:
        insp = _docker_request(socket_path, f"/containers/{cid}/json")
        health, restart_count, exit_code, oom_killed = _docker_inspect_fields(insp)
    except (OSError, ValueError):
        pass

    cpu_pct = mem_usage = mem_limit = mem_pct = None
    net_rx = net_tx = blk_read = blk_write = None
    if state == "running":
        try:
            stats = _docker_request(socket_path, f"/containers/{cid}/stats?stream=false")
            cpu_pct = _docker_cpu_pct(stats)
            mem_usage, mem_limit, mem_pct = _docker_mem(stats)
            net_rx, net_tx, blk_read, blk_write = _docker_net_io(stats)
        except (OSError, ValueError):
            pass

    started_at = _docker_started_at(insp)
    # When it last stopped (a running container's is its previous stop, or
    # None): dates a die event, and tells the log collector a stop is new.
    finished_at = None if state == "running" else _docker_started_at(insp, "FinishedAt")
    record = {
        "name": name,
        "container_id": cid[:12],
        "image": c.get("Image", ""),
        "status": state,
        "cpu_pct": cpu_pct,
        "mem_usage_bytes": mem_usage,
        "mem_limit_bytes": mem_limit,
        "mem_pct": mem_pct,
        "ports": _docker_ports(c.get("Ports", [])),
        "started_at": started_at,
        "finished_at": finished_at,
        "health": health,
        "restart_count": restart_count,
        "exit_code": exit_code,
        "oom_killed": oom_killed,
        "net_rx_bytes": net_rx,
        "net_tx_bytes": net_tx,
        "blk_read_bytes": blk_read,
        "blk_write_bytes": blk_write,
    }
    record.update(_docker_grouping(c.get("Labels")))
    return record


def _docker_failure(socket_path: str, exc: Exception) -> tuple:
    """(status, detail) for a failed Docker call: "unavailable" when there is no
    Docker here to ask, "error" when there is one and the call went wrong."""
    if isinstance(exc, FileNotFoundError):
        return "unavailable", "no Docker socket at " + socket_path
    if isinstance(exc, ConnectionRefusedError):
        return "unavailable", "Docker is not accepting connections on " + socket_path
    if isinstance(exc, PermissionError):
        return "error", "permission denied on " + socket_path
    if isinstance(exc, socket.timeout):
        return "error", "Docker did not answer within %.0f s" % DOCKER_API_TIMEOUT
    return "error", (str(exc) or type(exc).__name__)[:200]


def collect_docker(socket_path: str) -> tuple:
    """(status, detail, containers) from the local Docker socket.

    status is "ok" when Docker answered (containers may then be [] on a host
    with none, which tells Steward to clear its list), "unavailable" when there
    is no Docker here (no socket, or nothing listening), and "error" when Docker
    is here but the listing failed. detail says why, for the last two. A
    Docker-less host still needs no configuration: it reports "unavailable".
    """
    try:
        containers = _docker_request(socket_path, "/containers/json?all=true")
    except (OSError, ValueError) as exc:
        return _docker_failure(socket_path, exc) + ([],)
    if not isinstance(containers, list):
        return "error", "unexpected answer from /containers/json", []
    if not containers:
        return "ok", None, []

    # Each container needs an inspect (+ a stats read if running), and the stats
    # call blocks ~1s while the daemon computes the cpu delta. Done serially that
    # stretches the whole sample on a busy host, so fan the per-container work out
    # over a small bounded thread pool (I/O-bound → threads are enough).
    # A deadline on the whole fan-out: one wedged container must not stall the
    # sample. Unfinished work is reported as Docker not answering in time, not
    # as a partial list: a container left out would read as removed, and one
    # without its inspect as restarted or changed health.
    ex = ThreadPoolExecutor(max_workers=min(8, len(containers)))
    try:
        futures = [ex.submit(_collect_one_container, socket_path, c) for c in containers]
        done, pending = wait_futures(futures, timeout=DOCKER_DEADLINE_SECONDS)
    finally:
        try:
            ex.shutdown(wait=False, cancel_futures=True)
        except TypeError:          # Python < 3.9: no cancel_futures
            ex.shutdown(wait=False)
    if pending:
        return ("error", "%d of %d containers not read within %g s"
                % (len(pending), len(containers), DOCKER_DEADLINE_SECONDS), [])
    return "ok", None, [f.result() for f in futures]


# ─── container logs (m79) ─────────────────────────────────────────────────────
#
# The logs endpoint returns a raw multiplexed byte stream, not JSON, so it needs
# a sibling of _docker_request that hands back bytes. Non-TTY containers frame
# stdout/stderr with an 8-byte header (stream byte + 4-byte big-endian length);
# TTY containers emit unframed bytes. We request timestamps=1, so every line is
# prefixed with an RFC3339Nano timestamp we parse for the since-cursor.

_LOG_STREAM_NAMES = {0: "stdout", 1: "stdout", 2: "stderr"}


def _docker_request_raw(socket_path: str, path: str,
                        timeout: float = DOCKER_API_TIMEOUT) -> bytes:
    """GET `path` from the Docker API over the Unix socket; return the raw body.

    Sibling of `_docker_request` for non-JSON endpoints (container logs). The
    whole answer must arrive within `timeout` and fit in
    DOCKER_LOG_RESPONSE_MAX_BYTES (#5764); raises OSError on any transport
    problem or non-2xx status so callers can silent-skip, and
    DockerAnswerTooLarge past the size. A 200 with an empty body (no new lines
    since the cursor) returns b"".
    """
    status, body = _docker_get(socket_path, path, "application/octet-stream", timeout,
                               timeout, DOCKER_LOG_RESPONSE_MAX_BYTES)
    if not (200 <= status < 300):
        raise OSError(f"docker API {path} returned {status}")
    return body


def _demux_docker_logs(raw: bytes) -> list:
    """Split a Docker logs stream into [(stream_name, payload_bytes), …].

    Non-TTY containers multiplex with an 8-byte frame header (stream byte in
    {0,1,2} + 4-byte big-endian length). The stream is treated as framed only if
    the whole buffer parses as clean back-to-back frames; anything else (a TTY
    container's raw stream, or a malformed header) falls back to a single stdout
    blob so no bytes are lost.
    """
    if not raw:
        return []
    frames = []
    i, n = 0, len(raw)
    while i + 8 <= n:
        stream = raw[i]
        if stream > 2 or raw[i + 1] or raw[i + 2] or raw[i + 3]:
            break  # not a valid frame header → not framed
        size = int.from_bytes(raw[i + 4:i + 8], "big")
        if i + 8 + size > n:
            break  # frame overruns the buffer → not framed
        frames.append((_LOG_STREAM_NAMES.get(stream, "stdout"),
                       raw[i + 8:i + 8 + size]))
        i += 8 + size
    if frames and i == n:
        return frames
    return [("stdout", raw)]


def _parse_log_ts(token: str):
    """Parse a Docker RFC3339Nano timestamp token into a datetime, or None.

    Normalises `Z` → `+00:00` and trims Docker's nanosecond precision to the
    microseconds datetime.fromisoformat accepts (stdlib, Python 3.8+ safe).
    """
    t = token.strip()
    if not t:
        return None
    if t.endswith("Z"):
        t = t[:-1] + "+00:00"
    m = re.match(r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(\.\d+)?(.*)$", t)
    if m:
        frac = m.group(2) or ""
        if len(frac) > 7:               # "." + up to 6 fractional digits
            frac = frac[:7]
        t = m.group(1) + frac + m.group(3)
    try:
        return datetime.fromisoformat(t)
    except ValueError:
        return None


def _parse_container_logs(raw: bytes) -> list:
    """Demux + line-split a logs stream → [(dt|None, stream, line), …] in order.

    Each line is timestamp-prefixed (we request timestamps=1); an unparseable
    prefix yields dt=None with the full raw line kept.
    """
    out = []
    for stream, payload in _demux_docker_logs(raw):
        text = payload.decode("utf-8", "replace")
        for raw_line in text.split("\n"):
            if not raw_line:
                continue
            token, _, msg = raw_line.partition(" ")
            dt = _parse_log_ts(token)
            if dt is not None:
                out.append((dt, stream, msg))
            else:
                out.append((None, stream, raw_line))
    return out


def _log_note(line: str) -> dict:
    """A line Steward's own log collection adds to the batch, shown under
    the "_steward" container."""
    return {"container": "_steward", "stream": "stderr",
            "ts": datetime.now(timezone.utc).isoformat(), "line": "[steward] " + line}


def collect_docker_logs(socket_path: str, containers: list, state: dict,
                        exclude=None,
                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES,
                        now: float | None = None,
                        deadline: float = DOCKER_LOG_DEADLINE_SECONDS) -> list:
    """New container-log lines since the last interval.

    Keeps a per-container since-cursor in `state["docker_log_cursors"]`
    (name → unix ts) so each interval fetches only what's new; a container's
    first interval seeds from a short tail. A running container is read every
    interval. A stopped one is read once per stop (`finished_at`, remembered in
    `state["docker_log_finals"]`), so the lines it wrote on its way down reach
    Steward (#5049): since its cursor if it was being read, or, for one first
    seen already stopped with a non-zero exit in the last
    DOCKER_LOG_FINAL_WINDOW_SECONDS, its last DOCKER_LOG_FINAL_TAIL lines. That
    is how a task that crashes between two samples is explained by its own
    output. Both maps forget a container once it is no longer listed.

    The whole batch is capped at `max_bytes` of log text — once exceeded a
    truncation marker is appended and collection stops (logs must never balloon
    a push). `exclude` container names are skipped entirely (a per-host
    bandwidth opt-out). Best-effort: a per-container transport error just
    contributes nothing this interval.

    What is read is bounded too (#5764): at most DOCKER_LOG_MAX_LINES lines
    per container (the newest; a note says older ones were skipped), at most
    DOCKER_LOG_RESPONSE_MAX_BYTES per answer (past it the container skips
    ahead to now), and all of it within `deadline` seconds. A container that
    writes faster than it can be read costs one bounded read per interval,
    never a backlog re-read every time.
    Returns [{"container", "stream", "ts", "line"}, …].
    """
    exclude_set = set(exclude or ())
    cursors = state.setdefault("docker_log_cursors", {})
    finals = state.setdefault("docker_log_finals", {})
    now = time.time() if now is None else now
    listed = {c.get("name"): c for c in (containers or []) if c.get("name")}
    # Forget containers no longer listed so state can't grow without bound
    # over the agent's lifetime.
    for book in (cursors, finals):
        for gone in [k for k in book if k not in listed]:
            del book[gone]

    out: list = []
    total = 0
    truncated = False
    end = time.monotonic() + deadline
    # since + tail: of the lines since the cursor, the newest MAX_LINES.
    since_tail = f"&tail={DOCKER_LOG_MAX_LINES}"
    for name, c in listed.items():
        if name in exclude_set:
            continue
        left = end - time.monotonic()
        if left <= 0:
            out.append(_log_note(f"log reading stopped at its {deadline:g} s limit; "
                                 "the other containers are read next interval"))
            break
        prev = cursors.get(name)
        base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
        if c.get("status") == "running":
            # since accepts a fractional Unix ts; we still dedup the boundary line.
            path = (f"{base}&tail={DOCKER_LOG_FIRST_TAIL}" if prev is None
                    else f"{base}&since={prev:.9f}{since_tail}")
        else:
            stopped = c.get("finished_at")
            if not stopped or finals.get(name) == stopped:
                continue
            if prev is not None:
                path = f"{base}&since={prev:.9f}{since_tail}"
            else:
                at = _parse_log_ts(stopped)
                if (c.get("exit_code") in (None, 0) or at is None
                        or now - at.timestamp() > DOCKER_LOG_FINAL_WINDOW_SECONDS):
                    continue
                path = f"{base}&tail={DOCKER_LOG_FINAL_TAIL}"
            finals[name] = stopped
        try:
            raw = _docker_request_raw(socket_path, path, timeout=min(DOCKER_API_TIMEOUT, left))
        except DockerAnswerTooLarge:
            # More than one answer can hold: skip to now rather than re-asking
            # for the same backlog next interval.
            cursors[name] = now
            out.append(_log_note(f"{name} wrote more than {DOCKER_LOG_RESPONSE_MAX_BYTES} "
                                 "bytes of log since the last read; skipped ahead to now"))
            continue
        except (OSError, ValueError):
            continue
        parsed = _parse_container_logs(raw)
        if prev is not None and len(parsed) >= DOCKER_LOG_MAX_LINES:
            out.append(_log_note(f"{name} wrote more than {DOCKER_LOG_MAX_LINES} lines "
                                 "since the last read; older ones skipped"))
        newest = prev
        for dt, stream, line in parsed:
            ts_unix = dt.timestamp() if dt is not None else None
            if prev is not None and ts_unix is not None and ts_unix <= prev:
                continue  # boundary line already sent last interval
            if total >= max_bytes:
                truncated = True
                break
            out.append({"container": name, "stream": stream,
                        "ts": dt.isoformat() if dt is not None else None,
                        "line": line})
            total += len(line)
            if ts_unix is not None and (newest is None or ts_unix > newest):
                newest = ts_unix
        if newest is not None:
            cursors[name] = newest
        if truncated:
            break

    if truncated:
        out.append(_log_note(f"log batch truncated at {max_bytes} bytes; "
                             "remaining lines deferred to the next interval"))
    return out


# ─── swarm (manager-only) ────────────────────────────────────────────────────


def _swarm_is_manager(info: dict) -> bool:
    """True only on a reachable Swarm manager.

    `/services`, `/tasks`, `/nodes` are manager-only — a worker (or a non-swarm
    daemon) returns 503 for them. `Swarm.ControlAvailable` in `/info` is exactly
    "this node can serve the control-plane API", so we gate on it and skip the
    topology query everywhere else (no errors, near-zero cost on workers).
    """
    return bool((info or {}).get("Swarm", {}).get("ControlAvailable"))


# Newest tasks kept per service. Swarm retains about 5 per slot by default, so
# this covers a replicated service's recent history and a crash loop's deaths
# without shipping a global service's whole fleet every sample.
SWARM_TASKS_PER_SERVICE = 10
_TASK_TEXT_CHARS = 300


def _swarm_task(t: dict) -> dict:
    """One task as Steward stores it: where it ran, what Swarm wants of it,
    what it is doing, and why it stopped (Err/Message, exit code)."""
    status = t.get("Status") or {}
    cstatus = status.get("ContainerStatus") or {}
    exit_code = cstatus.get("ExitCode")
    return {
        "task_id": t.get("ID", "") or "",
        "slot": t.get("Slot"),
        "node_id": t.get("NodeID", "") or "",
        "desired_state": (t.get("DesiredState") or "").lower(),
        "state": (status.get("State") or "").lower(),
        "message": (status.get("Message") or "")[:_TASK_TEXT_CHARS],
        "error": (status.get("Err") or "")[:_TASK_TEXT_CHARS],
        "exit_code": exit_code if isinstance(exit_code, int) else None,
        "at": status.get("Timestamp") or t.get("UpdatedAt") or "",
    }


def _swarm_services(services: list, tasks: list) -> list:
    """Roll desired-vs-running replicas per service up from the task list.

    Replica health isn't on the service object — it's derived by counting tasks:
    `running` = tasks currently in state running; `desired` = the service's
    configured replica count (replicated) or the count of tasks the scheduler
    wants running (global, which has no fixed number). Each service also
    carries its newest tasks (`tasks`), so Steward can tell a task Swarm
    replaced from one that crashed, and say why the last one failed.
    """
    tasks_by_svc: dict = {}
    for t in (tasks or []):
        if t.get("ServiceID"):
            tasks_by_svc.setdefault(t["ServiceID"], []).append(_swarm_task(t))
    running_by_svc: dict = {}
    desired_by_svc: dict = {}  # only used for global-mode services
    placement: dict = {}       # svc_id -> {node_id: running_count}
    for t in (tasks or []):
        sid = t.get("ServiceID")
        if not sid:
            continue
        if ((t.get("Status") or {}).get("State") or "").lower() == "running":
            running_by_svc[sid] = running_by_svc.get(sid, 0) + 1
            node = t.get("NodeID")
            if node:
                placement.setdefault(sid, {})[node] = \
                    placement.setdefault(sid, {}).get(node, 0) + 1
        if (t.get("DesiredState") or "").lower() == "running":
            desired_by_svc[sid] = desired_by_svc.get(sid, 0) + 1

    out = []
    for s in (services or []):
        sid = s.get("ID", "") or ""
        spec = s.get("Spec") or {}
        name = spec.get("Name") or sid[:12]
        mode_obj = spec.get("Mode") or {}
        if "Global" in mode_obj:
            mode = "global"
            desired = desired_by_svc.get(sid, 0)
        else:
            mode = "replicated"
            desired = (mode_obj.get("Replicated") or {}).get("Replicas", 0) or 0
        image = (((spec.get("TaskTemplate") or {}).get("ContainerSpec") or {})
                 .get("Image") or "")
        image = image.split("@", 1)[0]  # drop the @sha256:… digest suffix
        out.append({
            "service_name": name, "mode": mode,
            "desired": desired, "running": running_by_svc.get(sid, 0),
            "image": image,
            # task→node placement of the running replicas (cross-node — a manager
            # sees every task, even ones whose containers live on other hosts).
            "placement": [{"node_id": n, "running": cnt}
                          for n, cnt in sorted(placement.get(sid, {}).items())],
            "tasks": sorted(tasks_by_svc.get(sid, []), key=lambda t: t["at"],
                            reverse=True)[:SWARM_TASKS_PER_SERVICE],
        })
    return out


def _swarm_nodes(nodes: list) -> list:
    """Normalise node role / availability / status (+ manager leader flag and,
    from 1.19.0, whether the swarm can reach a manager: "reachable",
    "unreachable" or "unknown"; "" for a worker), so Steward can count the
    managers the swarm has lost (#5788)."""
    out = []
    for n in (nodes or []):
        spec = n.get("Spec") or {}
        desc = n.get("Description") or {}
        status = n.get("Status") or {}
        mgr = n.get("ManagerStatus") or {}
        out.append({
            "node_id": n.get("ID", "") or "",
            "hostname": desc.get("Hostname", "") or "",
            "role": (spec.get("Role") or "worker").lower(),
            "availability": (spec.get("Availability") or "active").lower(),
            "status": (status.get("State") or "unknown").lower(),
            "leader": bool(mgr.get("Leader", False)),
            "reachability": (mgr.get("Reachability") or "").lower(),
        })
    return out


def collect_swarm(socket_path: str) -> tuple:
    """(is_manager, topology) from the local Docker socket.

    Self-detects manager via `/info` (one cheap call); only then queries the
    manager-only endpoints. is_manager is None when /info itself failed, so a
    transient error is never mistaken for a node that stopped being a manager;
    False tells Steward to drop the swarm rows this host reported as one.
    topology is None unless this is a manager and every call answered.
    """
    try:
        info = _docker_request(socket_path, "/info")
    except (OSError, ValueError):
        return None, None
    if not _swarm_is_manager(info):
        return False, None
    try:
        services = _docker_request(socket_path, "/services")
        tasks = _docker_request(socket_path, "/tasks")
        nodes = _docker_request(socket_path, "/nodes")
    except (OSError, ValueError):
        return True, None
    return True, {
        "services": _swarm_services(
            services if isinstance(services, list) else [],
            tasks if isinstance(tasks, list) else []),
        "nodes": _swarm_nodes(nodes if isinstance(nodes, list) else []),
    }


# ─── disk usage (image storage) ──────────────────────────────────────────────


def _disk_images(images: list) -> list:
    """Normalise /system/df image rows → compact per-image storage records.

    Sorted largest-first and capped so a host with a huge image cache doesn't
    bloat the payload; the panel only needs the heavy hitters. `containers` is
    how many containers reference the image (0 ⇒ reclaimable).
    """
    out = []
    for im in images:
        if not isinstance(im, dict):
            continue
        tags = im.get("RepoTags") or []
        # Untagged/dangling images report ["<none>:<none>"] or null.
        tag = next((t for t in tags if t and t != "<none>:<none>"), None)
        raw_id = (im.get("Id") or "").split(":", 1)[-1]
        out.append({
            "image_id": raw_id[:12],
            "repo_tag": tag or "<none>",
            "size": int(im.get("Size") or 0),
            "shared_size": int(im.get("SharedSize") or 0) if im.get("SharedSize", -1) >= 0 else 0,
            "containers": int(im.get("Containers") or 0) if (im.get("Containers") or 0) > 0 else 0,
        })
    out.sort(key=lambda r: r["size"], reverse=True)
    return out[:50]


def collect_disk_usage(socket_path: str):
    """Docker disk usage from `/system/df`, or None if unavailable.

    One bounded API call (the daemon walks images/layers/volumes). Reclaimable =
    bytes held by images no container references — the same intent as
    `docker system df`'s RECLAIMABLE column. Silent-skip (None) on any transport
    failure, same contract as collect_docker/collect_swarm.
    """
    try:
        df = _docker_request(socket_path, "/system/df")
    except (OSError, ValueError):
        return None
    if not isinstance(df, dict):
        return None

    images = [im for im in (df.get("Images") or []) if isinstance(im, dict)]
    containers = [c for c in (df.get("Containers") or []) if isinstance(c, dict)]
    volumes = [v for v in (df.get("Volumes") or []) if isinstance(v, dict)]
    build_cache = [b for b in (df.get("BuildCache") or []) if isinstance(b, dict)]

    def _vol_size(v):
        return int((v.get("UsageData") or {}).get("Size") or 0)

    return {
        "layers_size": int(df.get("LayersSize") or 0),
        "images_total": len(images),
        "images_active": sum(1 for im in images if (im.get("Containers") or 0) > 0),
        "images_size": sum(int(im.get("Size") or 0) for im in images),
        "images_reclaimable": sum(
            int(im.get("Size") or 0) for im in images if (im.get("Containers") or 0) <= 0),
        "containers_count": len(containers),
        "containers_size": sum(int(c.get("SizeRw") or 0) for c in containers),
        "volumes_count": len(volumes),
        "volumes_size": sum(_vol_size(v) for v in volumes if _vol_size(v) > 0),
        "build_cache_size": sum(int(b.get("Size") or 0) for b in build_cache),
        "images": _disk_images(images),
    }


# ─── ring buffer ─────────────────────────────────────────────────────────────


class RingBuffer:
    def __init__(self, maxlen: int = 20) -> None:
        self._dq: deque = deque(maxlen=maxlen)

    def push(self, item) -> None:
        self._dq.append(item)

    def drain(self) -> list:
        out = list(self._dq)
        self._dq.clear()
        return out

    def __len__(self) -> int:
        return len(self._dq)


# ─── pending updates and reboots (#5765, #5803) ──────────────────────────────
#
# What the package manager has waiting, read once an hour in a background
# thread so a slow apt never holds up a sample, and whether the host needs a
# reboot, read every sample (a file test). Nothing here needs root. apt-check
# and `apt list` read the lists apt-daily already fetched; dnf, yum and zypper
# run from their caches (-C, --no-refresh). pacman has no cache to ask without
# root, so checkupdates (pacman-contrib) syncs its own copy of the package
# lists, in the agent's private /tmp, and does reach the mirrors.

UPDATES_INTERVAL_SECONDS = 3600.0
UPDATES_DEADLINE_SECONDS = 120.0
REBOOT_REQUIRED = "/var/run/reboot-required"
REBOOT_PACKAGES_MAX = 20
KERNEL_MODULES = "/usr/lib/modules"


def _run(cmd: list, deadline: float = UPDATES_DEADLINE_SECONDS):
    """(returncode, stdout, stderr) of `cmd`, or None when it isn't installed.
    Raises subprocess.TimeoutExpired past the deadline (the child is killed)."""
    if shutil.which(cmd[0]) is None and not os.path.exists(cmd[0]):
        return None
    env = dict(os.environ, LANG="C", LC_ALL="C")
    done = subprocess.run(cmd, capture_output=True, text=True, timeout=deadline, env=env)
    return done.returncode, done.stdout, done.stderr


def _failed(manager: str, got, what: str, at: str) -> dict:
    return {"manager": manager, "error": ((got[2] or got[1]) or what + " failed").strip()[:200],
            "checked_at": at}


def parse_apt_check(stderr: str):
    """apt-check writes "<pending>;<security>" to stderr; (pending, security) or None."""
    m = re.search(r"(\d+);(\d+)", stderr or "")
    return (int(m.group(1)), int(m.group(2))) if m else None


def parse_apt_list(stdout: str) -> tuple:
    """(pending, security) from `apt list --upgradable`: one line per package,
    "name/suite[,suite…] version arch [upgradable from: …]"; a security update
    comes from a "-security" suite."""
    pending = security = 0
    for line in (stdout or "").splitlines():
        if "upgradable from" not in line:
            continue
        pending += 1
        suites = line.split("/", 1)[1].split(" ", 1)[0] if "/" in line else ""
        if "-security" in suites:
            security += 1
    return pending, security


def parse_dnf_check_update(stdout: str) -> int:
    """Packages `dnf check-update` (or yum's) lists, up to its "Obsoleting
    Packages" part. yum also prints "Loaded plugins:" lines first."""
    count = 0
    for line in (stdout or "").splitlines():
        if line.startswith("Obsoleting"):
            break
        if line.strip() and not line.startswith((" ", "Last metadata", "Loaded plugins",
                                                 "Security:")):
            count += 1
    return count


def parse_checkupdates(stdout: str) -> int:
    """Packages pacman-contrib's checkupdates lists: "name old -> new"."""
    return sum(1 for line in (stdout or "").splitlines() if " -> " in line)


def _zypper_rows(stdout: str) -> list:
    """The data rows of a zypper table: lines with "|" after the "---+---" rule."""
    rows, seen_rule = [], False
    for line in (stdout or "").splitlines():
        if re.match(r"^-+\+[-+]+$", line.strip()):
            seen_rule = True
        elif seen_rule and "|" in line:
            rows.append([c.strip() for c in line.split("|")])
    return rows


def parse_zypper_updates(stdout: str) -> int:
    """Packages `zypper list-updates` lists (one table row each)."""
    return len(_zypper_rows(stdout))


def parse_zypper_patches(stdout: str) -> int:
    """Security patches `zypper list-patches --category security` says are
    still needed (its Status column), not those already applied."""
    return sum(1 for row in _zypper_rows(stdout) if "needed" in row)


def _os_release_id(path: str = "/etc/os-release") -> str:
    try:
        with open(path) as fh:
            for line in fh:
                if line.startswith("ID="):
                    return line.split("=", 1)[1].strip().strip('"')
    except OSError:
        pass
    return ""


def _check_apt(at: str):
    got = _run(["/usr/lib/update-notifier/apt-check"])
    if got is not None:
        counts = parse_apt_check(got[2])
        if counts:
            return {"manager": "apt", "pending": counts[0], "security": counts[1], "checked_at": at}
    got = _run(["apt", "list", "--upgradable"])
    if got is not None and got[0] == 0:
        pending, security = parse_apt_list(got[1])
        return {"manager": "apt", "pending": pending, "security": security, "checked_at": at}
    return None


def _needs_restarting(out: dict) -> dict:
    """dnf and yum hosts: needs-restarting -r exits 1 when a reboot is needed."""
    restart = _run(["needs-restarting", "-r"])
    if restart is not None and restart[0] in (0, 1):
        out["reboot_required"] = restart[0] == 1
    return out


def _check_dnf_like(manager: str, at: str):
    """dnf, and yum where there is no dnf: the same output and exit codes."""
    got = _run([manager, "-q", "-C", "check-update"])
    if got is None:
        return None
    if got[0] not in (0, 100):        # 100: updates are waiting
        return _failed(manager, got, manager + " check-update", at)
    out = {"manager": manager, "pending": parse_dnf_check_update(got[1]),
           "security": None, "checked_at": at}
    sec = _run([manager, "-q", "-C", "updateinfo", "list", "--security"] if manager == "dnf"
               else [manager, "-q", "-C", "updateinfo", "list", "security"])
    if sec is not None and sec[0] == 0:
        out["security"] = len({ln.split()[-1] for ln in sec[1].splitlines()
                               if ln.strip() and not ln.startswith("Loaded plugins")})
    return _needs_restarting(out)


def _check_zypper(at: str):
    """openSUSE / SLES. Tumbleweed upgrades with dup, so its pending list is
    dup's, and it has no patches to call security."""
    tumbleweed = _os_release_id() == "opensuse-tumbleweed"
    got = _run(["zypper", "--no-refresh", "-q", "list-updates"] + (["--dup"] if tumbleweed else []))
    if got is None:
        return None
    if got[0] != 0:
        return _failed("zypper", got, "zypper list-updates", at)
    out = {"manager": "zypper", "pending": parse_zypper_updates(got[1]), "security": None,
           "checked_at": at}
    if not tumbleweed:
        sec = _run(["zypper", "--no-refresh", "-q", "list-patches", "--category", "security"])
        if sec is not None and sec[0] in (0, 100, 101):
            out["security"] = parse_zypper_patches(sec[1])
    restart = _run(["zypper", "needs-rebooting"])
    if restart is not None and restart[0] in (0, 102):     # 102: a reboot is needed
        out["reboot_required"] = restart[0] == 102
    return out


def _check_pacman(at: str):
    """Arch and its family (CachyOS, Manjaro, EndeavourOS). pacman keeps no
    security list, so security is None. Without pacman-contrib there is no
    safe check: `pacman -Sy` needs root and leaves a partial upgrade."""
    if shutil.which("pacman") is None:
        return None
    got = _run(["checkupdates"])
    if got is None:
        return {"manager": "pacman", "error": "checkupdates isn't installed (pacman-contrib)",
                "checked_at": at}
    if got[0] not in (0, 2):          # 2: nothing to update
        return _failed("pacman", got, "checkupdates", at)
    return {"manager": "pacman", "pending": parse_checkupdates(got[1]), "security": None,
            "checked_at": at}


# Asked in this order; the first that answers is the host's. dnf comes before
# yum because dnf hosts often keep a yum alias that is dnf underneath.
UPDATE_CHECKS = (
    ("apt", _check_apt),
    ("dnf", lambda at: _check_dnf_like("dnf", at)),
    ("yum", lambda at: _check_dnf_like("yum", at)),
    ("zypper", _check_zypper),
    ("pacman", _check_pacman),
)


def collect_updates() -> dict:
    """{"manager", "pending", "security", "checked_at"} (security None when the
    manager can't say), {"error"} when the check failed, or {} on a host with
    no package manager we know. dnf, yum and zypper hosts also say whether a
    reboot is required."""
    at = _now_iso()
    try:
        for _name, check in UPDATE_CHECKS:
            got = check(at)
            if got is not None:
                return got
    except subprocess.TimeoutExpired:
        return {"error": "the package manager didn't answer in %.0f s" % UPDATES_DEADLINE_SECONDS,
                "checked_at": at}
    except OSError as exc:
        return {"error": (str(exc) or type(exc).__name__)[:200], "checked_at": at}
    return {}


def refresh_updates(state: dict, now_mono: float | None = None) -> None:
    """Start an update check in the background when one is due (hourly, and at
    start). Its answer lands in state["updates"] for the samples that follow;
    one check runs at a time. state["updates_due"] = True asks for one now."""
    now_mono = time.monotonic() if now_mono is None else now_mono
    if state.get("updates_running"):
        return
    last = state.get("updates_mono")
    if not state.pop("updates_due", False) and last is not None \
            and now_mono - last < UPDATES_INTERVAL_SECONDS:
        return
    state["updates_running"], state["updates_mono"] = True, now_mono

    def work():
        try:
            state["updates"] = collect_updates()
        except Exception as exc:              # never take the thread down silently
            state["updates"] = {"error": (str(exc) or type(exc).__name__)[:200],
                                "checked_at": _now_iso()}
        finally:
            state["updates_running"] = False

    threading.Thread(target=work, name="updates", daemon=True).start()


def collect_reboot(path: str = REBOOT_REQUIRED, modules: str = KERNEL_MODULES):
    """{"required", "since", "packages"} from Debian's reboot-required marker;
    on pacman hosts, whether the running kernel's modules are gone (an upgrade
    replaced the kernel); None where neither can be asked (dnf, yum and zypper
    say it with the hourly update check)."""
    try:
        st = os.stat(path)
    except FileNotFoundError:
        if shutil.which("apt-get"):
            return {"required": False}
        if shutil.which("pacman"):
            return {"required": not os.path.isdir(os.path.join(modules, os.uname().release))}
        return None
    packages: list = []
    try:
        with open(path + ".pkgs") as fh:
            packages = sorted({ln.strip() for ln in fh if ln.strip()})[:REBOOT_PACKAGES_MAX]
    except OSError:
        pass
    return {"required": True,
            "since": datetime.fromtimestamp(st.st_mtime, timezone.utc).isoformat(),
            "packages": packages}


# ── maintenance helper (#5766) ───────────────────────────────────────────────
# The agent can't update or reboot its host: it runs unprivileged under
# NoNewPrivileges. It leaves a request in MAINT_REQUESTS, the one directory its
# unit lets it write; steward-maint.path starts the root helper (maint.py),
# which does only update / reboot / drain / undrain and writes each outcome to
# MAINT_RESULTS. The agent reads those back and reports each new state once.

MAINT_REQUESTS = "/var/lib/steward-agent/requests"
MAINT_RESULTS = "/var/lib/steward-maint/results"
MAINT_PATH_UNIT = "/etc/systemd/system/steward-maint.path"
MAINT_ACTIONS = ("update", "reboot", "drain", "undrain", "drives")
MAINT_RESULT_MAX_BYTES = 8192
# The helper keeps results a week; only this recent a one is reported, so a
# job a reboot interrupted still reaches Steward from the agent's next start
# and old ones don't repeat on every start.
MAINT_REPORT_WINDOW_SECONDS = 86400
_MAINT_ID = re.compile(r"^[A-Za-z0-9-]{1,64}$")
_MAINT_FIELDS = ("action", "state", "ok", "packages", "reboot_required", "node",
                 "detail", "started_at", "finished_at")


def maint_helper_installed(unit: str = MAINT_PATH_UNIT,
                           requests_dir: str = MAINT_REQUESTS) -> bool:
    """The helper's path unit is installed and the agent may leave it a request."""
    return os.path.exists(unit) and os.access(requests_dir, os.W_OK)


def request_maintenance(job_id: str, action: str, node: str | None = None,
                        requests_dir: str = MAINT_REQUESTS) -> None:
    """Leave the helper a request, whole: written to a dot-file the path unit
    doesn't match, then renamed to `<id>.json`."""
    if action not in MAINT_ACTIONS or not _MAINT_ID.match(job_id or ""):
        raise ValueError(f"not a maintenance request: {action!r} {job_id!r}")
    req = {"id": job_id, "action": action}
    if node:
        req["node"] = node
    tmp = os.path.join(requests_dir, f".{job_id}.tmp")
    with open(tmp, "w") as fh:
        json.dump(req, fh)
    os.replace(tmp, os.path.join(requests_dir, f"{job_id}.json"))


def take_maintenance(steps, requested: set, requests_dir: str = MAINT_REQUESTS) -> list:
    """Leave the helper each step Steward handed over that this agent hasn't
    already left it; returns the ids left. Steward hands a step on every post
    until its result arrives, so a step is taken once here, and the helper
    also refuses an id it has a result for (after a reboot this set is empty)."""
    taken = []
    for step in steps if isinstance(steps, list) else []:
        if not isinstance(step, dict) or step.get("id") in requested:
            continue
        try:
            request_maintenance(str(step.get("id") or ""), str(step.get("action") or ""),
                                step.get("node"), requests_dir=requests_dir)
        except (ValueError, OSError) as exc:
            _log("WARN", f"maintenance step {step.get('id')!r} not taken: {exc}")
            continue
        requested.add(step["id"])
        taken.append(step["id"])
        _log("INFO", f"maintenance: {step['action']} requested ({step['id']})")
    return taken


def maintenance_results(seen: set, results_dir: str = MAINT_RESULTS,
                        now: float | None = None) -> tuple:
    """(events, keys): a `maintenance` event for each result whose (id, state)
    isn't in `seen`. The caller adds `keys` to `seen` once Steward has them."""
    now = time.time() if now is None else now
    try:
        names = sorted(n for n in os.listdir(results_dir) if n.endswith(".json"))
    except OSError:
        return [], []
    events, keys = [], []
    for name in names:
        path = os.path.join(results_dir, name)
        try:
            st = os.stat(path)
            if st.st_size > MAINT_RESULT_MAX_BYTES:
                continue
            with open(path) as fh:
                res = json.load(fh)
        except (OSError, ValueError):
            continue                        # mid-write or gone: next interval
        if not isinstance(res, dict) or not _MAINT_ID.match(str(res.get("id", ""))):
            continue
        key = (res["id"], res.get("state"))
        if key in seen:
            continue
        if now - st.st_mtime > MAINT_REPORT_WINDOW_SECONDS:
            continue                        # long finished: not news to anyone
        event = {"kind": "maintenance", "id": res["id"]}
        event.update({k: res[k] for k in _MAINT_FIELDS if k in res})
        events.append(event)
        keys.append(key)
    return events, keys


# ── drive health (#5789) ─────────────────────────────────────────────────────
# Reading SMART needs root, so the helper does it: hourly (and at start) the
# agent leaves it a `drives` request, and the helper writes each drive's
# figures to DRIVES_PATH. The last answer rides every sample, like updates.

DRIVES_PATH = "/var/lib/steward-maint/drives.json"
DRIVES_INTERVAL_SECONDS = 3600
DRIVES_MAX_BYTES = 65536


def refresh_drives(state: dict, now_mono: float | None = None,
                   requests_dir: str = MAINT_REQUESTS, drives_path: str = DRIVES_PATH,
                   unit: str = MAINT_PATH_UNIT) -> None:
    """Ask the helper for a drive read when one is due, and keep its newest
    answer in state["drives"]."""
    now_mono = time.monotonic() if now_mono is None else now_mono
    last = state.get("drives_asked_mono")
    if last is None or now_mono - last >= DRIVES_INTERVAL_SECONDS:
        state["drives_asked_mono"] = now_mono
        if not maint_helper_installed(unit, requests_dir):
            state["drives"] = {"error": "needs the maintenance helper (Update agent installs it)"}
        else:
            try:
                request_maintenance(f"drives-{int(time.time())}", "drives", requests_dir=requests_dir)
            except (ValueError, OSError) as exc:
                _log("WARN", f"drive health not requested: {exc}")
    try:
        st = os.stat(drives_path)
    except OSError:
        return
    if st.st_mtime == state.get("drives_mtime") or st.st_size > DRIVES_MAX_BYTES:
        return
    try:
        with open(drives_path) as fh:
            got = json.load(fh)
    except (OSError, ValueError):
        return                              # mid-write: the next sample reads it
    if isinstance(got, dict):
        state["drives"], state["drives_mtime"] = got, st.st_mtime


def collect_self(state: dict) -> dict:
    """The agent's own footprint: resident memory and CPU % of one core since
    the last sample (#5764). A 1.13.0 agent once grew to 2.1 GB and spun for a
    day before going silent; this is how Steward sees the next one coming."""
    out: dict = {}
    for line in _read_file("/proc/self/status").splitlines():
        if line.startswith("VmRSS:"):
            out["rss_bytes"] = int(line.split()[1]) * 1024
            break
    t = os.times()
    cpu, mono = t.user + t.system, time.monotonic()
    prev_cpu, prev_mono = state.get("self_cpu"), state.get("self_mono")
    if prev_cpu is not None and mono > prev_mono:
        out["cpu_pct"] = round(100.0 * max(0.0, cpu - prev_cpu) / (mono - prev_mono), 2)
    state["self_cpu"], state["self_mono"] = cpu, mono
    return out


# ─── payload ─────────────────────────────────────────────────────────────────


def build_sample(mounts: list[str], state: dict,
                 docker_socket: str = DEFAULT_DOCKER_SOCKET, *,
                 docker_logs_enabled: bool = True,
                 docker_log_exclude=None) -> dict:
    """Collect one full sample. Partial samples allowed if a collector fails.

    `state` carries the previous network/disk counters + monotonic timestamp so
    throughput rates can be derived from deltas (and the per-container log
    since-cursors); it is mutated in place. `docker_socket` is probed
    best-effort — the `docker` key is omitted entirely when no containers are
    found, so non-Docker hosts add nothing to the payload. `docker_logs_enabled`
    / `docker_log_exclude` gate incremental container-log collection.
    """
    sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
    try:
        agg, cores = collect_cpu()
        sample["cpu_pct"] = agg
        sample["cpu_cores"] = cores
    except Exception:
        sample["cpu_pct"] = None
    try:
        sample["mem"] = collect_memory()
    except Exception:
        sample["mem"] = None
    try:
        sample["load"] = collect_load()
    except Exception:
        sample["load"] = None
    try:
        sample["uptime_secs"] = collect_uptime()
    except Exception:
        sample["uptime_secs"] = None
    try:
        sample["storage"] = collect_storage(mounts)
    except Exception:
        sample["storage"] = []
    try:
        sample["temps"] = collect_temps()
    except Exception:
        sample["temps"] = []
    try:
        sample["psi"] = collect_psi()
    except Exception:
        sample["psi"] = {}

    # Throughput rates: diff cumulative counters against the previous sample.
    now_mono = time.monotonic()
    try:
        net_raw = collect_net_raw()
    except Exception:
        net_raw = {}
    try:
        disk_raw = collect_diskio_raw()
    except Exception:
        disk_raw = {}
    dt = now_mono - state["mono"] if state.get("mono") else 0.0
    sample["net"] = [
        {"iface": n, "rx_bps": round(rx, 1), "tx_bps": round(tx, 1)}
        for n, (rx, tx) in _rates(net_raw, state.get("net", {}), dt).items()
    ]
    sample["diskio"] = [
        {"device": n, "read_bps": round(rd, 1), "write_bps": round(wr, 1)}
        for n, (rd, wr) in _rates(disk_raw, state.get("disk", {}), dt).items()
    ]
    try:
        nfs_raw = collect_nfs_raw()
    except Exception:
        nfs_raw = {}
    if nfs_raw:
        sample["nfs"] = nfs_delays(nfs_raw, state.get("nfs", {}), dt)
    state["net"], state["disk"], state["nfs"], state["mono"] = net_raw, disk_raw, nfs_raw, now_mono

    # Docker's status rides every sample, so Steward can tell "no containers"
    # (status ok, docker: []) from "couldn't ask" (unavailable / error).
    try:
        docker_status, docker_detail, docker = collect_docker(docker_socket)
    except Exception as exc:
        docker_status, docker_detail, docker = "error", (str(exc) or type(exc).__name__)[:200], []
    sample["docker_status"] = docker_status
    if docker_detail:
        sample["docker_detail"] = docker_detail
    if docker_status == "ok":
        sample["docker"] = docker

    # Swarm topology is only emitted by managers. A node whose /info says it
    # is not a manager says so, so Steward can expire what it once reported.
    try:
        is_manager, swarm = collect_swarm(docker_socket)
    except Exception:
        is_manager, swarm = None, None
    if swarm is not None:
        sample["swarm"] = swarm
    elif is_manager is False:
        sample["swarm_manager"] = False

    # Image/disk usage — host-level docker fact (one /system/df call), only
    # when Docker answered.
    if docker_status == "ok" and now_mono - state.get("df_mono", float("-inf")) >= DOCKER_DF_INTERVAL_SECONDS:
        state["df_mono"] = now_mono
        try:
            disk = collect_disk_usage(docker_socket)
        except Exception:
            disk = None
        if disk is not None:
            sample["docker_disk"] = disk

    # Container logs (m79): incremental per-container tail folded into the same
    # push as metrics. Skipped when there are no containers or logging is off;
    # the key is omitted when nothing new arrived this interval.
    if docker and docker_logs_enabled:
        try:
            logs = collect_docker_logs(docker_socket, docker, state,
                                       exclude=docker_log_exclude)
        except Exception:
            logs = []
        if logs:
            sample["docker_logs"] = logs

    # Pending updates: the last hourly answer rides every sample (#5765).
    try:
        refresh_updates(state)
        reboot = collect_reboot()
    except Exception:
        reboot = None
    if state.get("updates"):
        sample["updates"] = state["updates"]
    if reboot is not None:
        sample["reboot"] = reboot

    # Each drive's health, read hourly by the root helper (#5789).
    try:
        refresh_drives(state)
    except Exception:
        pass
    if state.get("drives"):
        sample["drives"] = state["drives"]

    # Last, so its CPU covers the whole of this sample's work.
    try:
        sample["agent"] = collect_self(state)
    except Exception:
        pass

    return sample


def build_payload(samples: list[dict], hostname: str, metadata: dict,
                  events: list | None = None) -> dict:
    # sent_at is this host's clock at the moment of sending: Steward compares
    # it with its own on arrival to say how far apart the two clocks are.
    payload = {
        "agent_version": AGENT_VERSION,
        "hostname": hostname,
        "metadata": metadata,
        "sent_at": datetime.now(timezone.utc).isoformat(),
        "samples": samples,
    }
    if events:
        payload["events"] = events
    return payload


# ─── what happened to the host (#5756) ───────────────────────────────────────
# Events ride the next post that gets through, so Steward records a reboot as a
# reboot and a lost connection as the span it lasted, not as the agent gone.


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def started_event() -> dict:
    """The agent starting, with when the host booted: Steward tells a host
    that rebooted from an agent that only restarted by it."""
    event = {"kind": "started", "at": _now_iso()}
    try:
        booted = time.time() - float(_read_file(UPTIME_PATH).split()[0])
        event["boot_at"] = datetime.fromtimestamp(booted, timezone.utc).isoformat()
    except Exception:
        pass
    return event


def host_shutting_down():
    """Whether the host itself is going down, not only this agent: systemd says
    "stopping" while it shuts down or reboots. None when it can't tell."""
    try:
        out = subprocess.run(["systemctl", "is-system-running"], stdout=subprocess.PIPE,
                             stderr=subprocess.DEVNULL, universal_newlines=True,
                             timeout=3).stdout.strip()
    except (OSError, subprocess.SubprocessError):
        return None
    return out == "stopping" if out else None


def stopping_event() -> dict:
    return {"kind": "stopping", "at": _now_iso(), "host_shutdown": host_shutting_down()}


def lost_contact_event(since: str) -> dict:
    """The span this agent could not reach Steward: from its first failed post
    to now, the post that is carrying it."""
    return {"kind": "lost_contact", "from": since, "to": _now_iso()}


def lost_contact(status) -> bool:
    """Whether a failed post means Steward couldn't be reached: no answer, or
    a proxy's 5xx in front of a Steward that is down. A 400 or 401 is Steward
    answering."""
    return status is None or status >= 500


# ─── POST + backoff ──────────────────────────────────────────────────────────


# Below Steward's 180 s staleness window (host_agent stale_after_seconds): once
# Steward is reachable again a host reports within a minute, instead of reading
# as stale for up to five (#5154; lesson: a check-in inherits its call's backoff).
BACKOFF_CAP = 60
# Samples in one POST: the server writes at most this many (routes.py
# MAX_SAMPLES_PER_INGEST), so a backlog drains a few per post, oldest first,
# rather than in one post too big to write before the timeout.
MAX_SAMPLES_PER_POST = 5


def _drop_logs(sample: dict) -> dict:
    """Strip container logs from a sample before it's buffered for retry.

    Logs are the one payload we never carry across a backoff — they'd bloat the
    ring buffer and go stale — while metrics are kept so an outage doesn't lose
    them. The since-cursor already advanced when the logs were collected, so the
    dropped lines are simply not re-sent (accepted loss during an outage).
    """
    sample.pop("docker_logs", None)
    return sample


def split_backlog(queued: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple:
    """(the samples to send in this post, the ones held for later): the oldest
    `limit` of the backlog with the new sample queued behind it.

    The new sample waits its turn. Sent ahead of the backlog it reached Steward
    before older samples did, and a rule counting checks in a row saw each
    post's run of stalled checks ended by the newest, healthy one: after the
    2026-10-17 NFS stall "Mount not answering" (10 in a row) never fired
    (#5995)."""
    return queued[:limit], queued[limit:]


def next_backoff(current: int) -> int:
    if current <= 0:
        return 30
    return min(current * 2, BACKOFF_CAP)


# Steward's answer to a post is small ({"ok", counts, maybe a maintenance
# step}); anything past this is not read.
ANSWER_MAX_BYTES = 65536


def post_payload(url: str, token: str, payload: dict,
                 answer: dict | None = None) -> tuple[bool, int | None]:
    """POST the payload. When `answer` is given, Steward's JSON reply is put
    in it (#5767: it carries any maintenance step for this host)."""
    body = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url.rstrip("/") + "/plugins/host_agent/ingest",
        data=body,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}",
            "User-Agent": f"steward-host-agent/{AGENT_VERSION}",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            if answer is not None:
                try:
                    got = json.loads(resp.read(ANSWER_MAX_BYTES).decode("utf-8"))
                    if isinstance(got, dict):
                        answer.update(got)
                except (ValueError, UnicodeDecodeError, OSError):
                    pass                # the post landed; only the reply is lost
            return (200 <= resp.status < 300, resp.status)
    except urllib.error.HTTPError as e:
        return (False, e.code)
    except (urllib.error.URLError, TimeoutError, socket.timeout, OSError):
        return (False, None)


# ─── main loop ───────────────────────────────────────────────────────────────


_reload_requested = False
_shutdown_requested = False


def _handle_hup(_signum, _frame):
    global _reload_requested
    _reload_requested = True


def _handle_term(_signum, _frame):
    global _shutdown_requested
    _shutdown_requested = True


def _log(level: str, msg: str) -> None:
    sys.stderr.write(f"[{level}] {msg}\n")
    sys.stderr.flush()


def main_loop(conf_path: str) -> int:
    global _reload_requested, _shutdown_requested
    signal.signal(signal.SIGHUP, _handle_hup)
    signal.signal(signal.SIGTERM, _handle_term)

    try:
        cfg = read_config(conf_path)
    except ConfigError as e:
        _log("ERROR", str(e))
        return 2

    metadata = collect_metadata()
    hostname = cfg.get("hostname") or socket.gethostname()
    mounts = cfg.get("mounts") or default_mounts()
    docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
    docker_logs_enabled = cfg.get("docker_logs_enabled", True)
    docker_log_exclude = cfg.get("docker_log_exclude") or []
    buffer = RingBuffer(maxlen=20)
    backoff = 0
    # Said once each, with the next post that gets through (#5756).
    events: list = [started_event()]
    lost_since = None
    # Carries previous net/disk counters + monotonic ts for rate computation.
    rate_state: dict = {}
    maint_seen: set = set()
    maint_requested: set = set()

    _log("INFO", f"steward-host-agent {AGENT_VERSION} starting "
                 f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)")

    while not _shutdown_requested:
        if _reload_requested:
            try:
                cfg = read_config(conf_path)
                mounts = cfg.get("mounts") or default_mounts()
                docker_socket = cfg.get("docker_socket") or DEFAULT_DOCKER_SOCKET
                docker_logs_enabled = cfg.get("docker_logs_enabled", True)
                docker_log_exclude = cfg.get("docker_log_exclude") or []
                metadata = collect_metadata()  # refresh host_ip/distro on reload
                _log("INFO", "config reloaded")
            except ConfigError as e:
                _log("ERROR", f"reload failed: {e}")
            _reload_requested = False

        sample = build_sample(mounts, rate_state, docker_socket,
                              docker_logs_enabled=docker_logs_enabled,
                              docker_log_exclude=docker_log_exclude)
        buffered = buffer.drain()
        send, held = split_backlog(buffered + [sample])
        maint_events, maint_keys = maintenance_results(maint_seen)
        payload = build_payload(
            samples=send,
            hostname=hostname,
            metadata=metadata,
            events=events + ([lost_contact_event(lost_since)] if lost_since else [])
            + maint_events,
        )
        answer: dict = {}
        ok, status = post_payload(cfg["url"], cfg["token"], payload, answer)

        if ok:
            take_maintenance(answer.get("maintenance"), maint_requested)
            if lost_since:
                _log("INFO", f"reached Steward again (unreachable since {lost_since})")
            events, lost_since = [], None
            maint_seen.update(maint_keys)
            # An update just finished: what's left waiting is known now, not in an hour.
            if any(e.get("action") == "update" and e.get("state") in ("done", "failed")
                   for e in maint_events):
                rate_state["updates_due"] = True
            for s in held:
                buffer.push(s)
            if buffered:
                _log("INFO", f"flushed {len(send)} buffered samples"
                             + (f", {len(held)} still held" if held else ""))
            backoff = 0
            sleep_for = cfg["interval_seconds"]
        else:
            if lost_since is None and lost_contact(status):
                lost_since = sample["ts"]
            if status == 400:
                _log("ERROR", "server rejected payload (400) — dropping sample")
            elif status == 401:
                _log("ERROR", "token rejected (401) — check config + UI")
                buffer.push(_drop_logs(sample))
            else:
                _log("WARN", f"POST failed (status={status}); buffering")
                for s in buffered:
                    buffer.push(_drop_logs(s))
                buffer.push(_drop_logs(sample))
            backoff = next_backoff(backoff)
            sleep_for = backoff

        slept = 0.0
        while slept < sleep_for and not _shutdown_requested and not _reload_requested:
            time.sleep(min(1.0, sleep_for - slept))
            slept += 1.0

    _log("INFO", "SIGTERM — flushing and exiting")
    # Say it is stopping, and whether the host is going down with it, so Steward
    # reads the silence that follows as a reboot rather than the agent lost.
    final_events = events + ([lost_contact_event(lost_since)] if lost_since else [])
    final_events.append(stopping_event())
    post_payload(cfg["url"], cfg["token"],
                 build_payload(buffer.drain(), hostname, metadata, events=final_events))
    return 0


if __name__ == "__main__":
    conf = os.environ.get("STEWARD_AGENT_CONFIG", "/etc/steward-agent.conf")
    sys.exit(main_loop(conf))