# 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.7.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"
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

# 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


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))


# ─── 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). Reuses
    the connect / send / de-chunk scaffolding; raises OSError on any transport
    problem or non-2xx status so callers can silent-skip. A 200 with an empty
    body (no new lines since the cursor) returns b"".
    """
    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/octet-stream\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)
    status = int(parts[1]) if len(parts) >= 2 and parts[1].isdigit() else 0
    if not (200 <= status < 300):
        raise OSError(f"docker API {path} returned {status}")
    if "transfer-encoding: chunked" in header_text.lower():
        body = _dechunk(body)
    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 collect_docker_logs(socket_path: str, containers: list, state: dict,
                        exclude=None,
                        max_bytes: int = DOCKER_LOG_BATCH_MAX_BYTES) -> list:
    """New container-log lines since the last interval, for running containers.

    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. 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.
    Returns [{"container", "stream", "ts", "line"}, …].
    """
    exclude_set = set(exclude or ())
    cursors = state.setdefault("docker_log_cursors", {})
    running = {c.get("name") for c in (containers or [])
               if c.get("name") and c.get("status") == "running"}
    # Forget cursors for containers no longer running so state can't grow without
    # bound over the agent's lifetime.
    for gone in [k for k in cursors if k not in running]:
        del cursors[gone]

    out: list = []
    total = 0
    truncated = False
    for c in (containers or []):
        name = c.get("name")
        if not name or name not in running or name in exclude_set:
            continue
        prev = cursors.get(name)
        base = f"/containers/{name}/logs?stdout=1&stderr=1&timestamps=1"
        # 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}")
        try:
            raw = _docker_request_raw(socket_path, path)
        except (OSError, ValueError):
            continue
        newest = prev
        for dt, stream, line in _parse_container_logs(raw):
            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({"container": "_steward", "stream": "stderr",
                    "ts": datetime.now(timezone.utc).isoformat(),
                    "line": (f"[steward] 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"))


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, *,
                 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()
    ]
    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

    # 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

    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 _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 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
    docker_logs_enabled = cfg.get("docker_logs_enabled", True)
    docker_log_exclude = cfg.get("docker_log_exclude") or []
    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
                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()
        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(_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")
    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))