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 sys
import time
import urllib.error
import urllib.request
from collections import deque
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone

AGENT_VERSION = "1.6.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",)


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()]
                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)
    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"
HWMON_DIR = "/sys/class/hwmon"
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),
    }


def collect_storage(mounts: list[str]) -> list[dict]:
    out: list[dict] = []
    for m in mounts:
        try:
            u = shutil.disk_usage(m)
        except (FileNotFoundError, PermissionError):
            continue
        out.append({"mount": m, "total_bytes": u.total, "used_bytes": u.used})
    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(),
    }


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 collect_net_raw() -> dict:
    """Return {iface: (rx_bytes, tx_bytes)} cumulative counters from /proc/net/dev.

    Skips loopback and veth* (container) interfaces to cut noise. 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 out


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

    Whole disks only — loop/ram/sr/fd and partitions are skipped. 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 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


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_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) and ValueError on a non-JSON body — both are caught by the
    caller and treated as "no docker here", so collection degrades silently.
    """
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect(socket_path)
        req = (
            "GET " + path + " HTTP/1.1\r\n"
            "Host: docker\r\n"
            "Accept: application/json\r\n"
            "Connection: close\r\n"
            "\r\n"
        )
        sock.sendall(req.encode("ascii"))
        chunks = []
        while True:
            buf = sock.recv(65536)
            if not buf:
                break
            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 status != 200:
        raise OSError(f"docker API {path} returned {status}")
    if "transfer-encoding: chunked" in header_text.lower():
        body = _dechunk(body)
    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_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 = 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

    created = c.get("Created")
    started_at = (
        datetime.fromtimestamp(created, tz=timezone.utc).isoformat()
        if isinstance(created, (int, float)) else None
    )
    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,
        "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 collect_docker(socket_path: str) -> list:
    """Per-container state from the local Docker socket, or [] if unavailable.

    Absent socket / permission denied / non-Docker endpoint all yield [] — the
    agent silently reports no containers rather than erroring, so a Docker-less
    host needs no configuration.
    """
    try:
        containers = _docker_request(socket_path, "/containers/json?all=true")
    except (OSError, ValueError):
        return []
    if not isinstance(containers, list) or not containers:
        return []

    # 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).
    workers = min(8, len(containers))
    with ThreadPoolExecutor(max_workers=workers) as ex:
        return list(ex.map(lambda c: _collect_one_container(socket_path, c), containers))


# ─── 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"))


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).
    """
    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())],
        })
    return out


def _swarm_nodes(nodes: list) -> list:
    """Normalise node role / availability / status (+ manager leader flag)."""
    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)),
        })
    return out


def collect_swarm(socket_path: str):
    """Swarm topology from a manager node, or None on workers / non-swarm hosts.

    Self-detects manager via `/info` (one cheap call); only then queries the
    manager-only endpoints. Any transport failure degrades to None, same silent
    contract as collect_docker — a non-swarm host adds nothing to the payload.
    """
    try:
        info = _docker_request(socket_path, "/info")
    except (OSError, ValueError):
        return None
    if not _swarm_is_manager(info):
        return None
    try:
        services = _docker_request(socket_path, "/services")
        tasks = _docker_request(socket_path, "/tasks")
        nodes = _docker_request(socket_path, "/nodes")
    except (OSError, ValueError):
        return None
    return {
        "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)


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


def build_sample(mounts: list[str], state: dict,
                 docker_socket: str = DEFAULT_DOCKER_SOCKET) -> 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; 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.
    """
    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()
    ]
    state["net"], state["disk"], state["mono"] = net_raw, disk_raw, now_mono

    try:
        docker = collect_docker(docker_socket)
    except Exception:
        docker = []
    if docker:
        sample["docker"] = docker

    # Swarm topology is only emitted by managers (collect_swarm returns None
    # everywhere else), so the key is absent on workers / non-swarm hosts.
    try:
        swarm = collect_swarm(docker_socket)
    except Exception:
        swarm = None
    if swarm is not None:
        sample["swarm"] = swarm

    # Image/disk usage — host-level docker fact (one /system/df call), omitted on
    # Docker-less hosts. Only meaningful when there's docker here at all, so skip
    # the call entirely when collect_docker found nothing.
    if docker:
        try:
            disk = collect_disk_usage(docker_socket)
        except Exception:
            disk = None
        if disk is not None:
            sample["docker_disk"] = disk

    return sample


def build_payload(samples: list[dict], hostname: str, metadata: dict) -> dict:
    return {
        "agent_version": AGENT_VERSION,
        "hostname": hostname,
        "metadata": metadata,
        "samples": samples,
    }


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


BACKOFF_CAP = 300


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


def post_payload(url: str, token: str, payload: dict) -> tuple[bool, int | None]:
    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:
            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
    buffer = RingBuffer(maxlen=20)
    backoff = 0
    # Carries previous net/disk counters + monotonic ts for rate computation.
    rate_state: dict = {}

    _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
                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)
        buffered = buffer.drain()
        payload = build_payload(
            samples=buffered + [sample],
            hostname=hostname,
            metadata=metadata,
        )
        ok, status = post_payload(cfg["url"], cfg["token"], payload)

        if ok:
            if buffered:
                _log("INFO", f"flushed {len(buffered)} buffered samples")
            backoff = 0
            sleep_for = cfg["interval_seconds"]
        else:
            if status == 400:
                _log("ERROR", "server rejected payload (400) — dropping sample")
            elif status == 401:
                _log("ERROR", "token rejected (401) — check config + UI")
                buffer.push(sample)
            else:
                _log("WARN", f"POST failed (status={status}); buffering")
                for s in buffered:
                    buffer.push(s)
                buffer.push(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")
    final = buffer.drain()
    if final:
        post_payload(cfg["url"], cfg["token"],
                     build_payload(final, hostname, metadata))
    return 0


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