# 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 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.14.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"
# 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),
    }


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
        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
        out.append(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(),
    }


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)


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


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_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). 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,
                        now: float | None = None) -> 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.
    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
    for name, c in listed.items():
        if name in exclude_set:
            continue
        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}")
        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}"
            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)
        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"))


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


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

    # 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

    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(buffered: list, limit: int = MAX_SAMPLES_PER_POST) -> tuple:
    """(the buffered samples to send with this one, the ones held for later):
    the oldest limit-1, leaving room for the new sample."""
    return buffered[:limit - 1], buffered[limit - 1:]


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
    # 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 = {}

    _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)
        payload = build_payload(
            samples=send + [sample],
            hostname=hostname,
            metadata=metadata,
            events=events + ([lost_contact_event(lost_since)] if lost_since else []),
        )
        ok, status = post_payload(cfg["url"], cfg["token"], payload)

        if ok:
            if lost_since:
                _log("INFO", f"reached Steward again (unreachable since {lost_since})")
            events, lost_since = [], None
            for s in held:
                buffer.push(s)
            if send:
                _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))