FabledSteward/plugins/host_agent/routes.py

# plugins/host_agent/routes.py
from __future__ import annotations

import hashlib
from datetime import datetime, timezone
from typing import Any

from pathlib import Path

import secrets
from quart import Blueprint, current_app, jsonify, request, Response, render_template, redirect, url_for, session
from steward.auth.middleware import require_role
from steward.models.users import UserRole
from sqlalchemy import select, func, or_, and_
from datetime import timedelta

from steward.core.settings import public_base_url
from steward.core.time_range import parse_range, RANGE_OPTIONS
from steward.models.hosts import Host
from steward.models.metrics import PluginMetric
from .models import HostAgentRegistration
# Query helpers live in a model-free module so integration tests can import them
# without the plugin-loader double-registration guard tripping (see metrics_query).
from .metrics_query import (
    SOURCE_MODULE,
    _history_for_host,
    _latest_metrics_for_host,
)

host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates")


def _hash_token(raw: str) -> str:
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def _parse_ts(ts: str) -> datetime:
    if ts.endswith("Z"):
        ts = ts[:-1] + "+00:00"
    return datetime.fromisoformat(ts)


def pick_host_address(current: str | None, reported: str | None) -> str | None:
    """Return the address to store on the Host, or None for no change.

    The agent-reported IP fills Host.address only when the current value is
    blank — an admin-typed address (DNS name, management IP) is never
    overwritten. The reported IP is always kept on the registration regardless,
    so admins can still see drift.
    """
    if reported and not (current or "").strip():
        return reported
    return None


def _expand_sample_to_metrics(
    sample: dict, host_name: str, recorded_at: datetime
) -> list[PluginMetric]:
    rows: list[PluginMetric] = []

    def row(metric: str, resource: str, value: float) -> None:
        rows.append(PluginMetric(
            source_module=SOURCE_MODULE,
            resource_name=resource,
            metric_name=metric,
            value=float(value),
            recorded_at=recorded_at,
        ))

    if sample.get("cpu_pct") is not None:
        row("cpu_pct", host_name, sample["cpu_pct"])

    mem = sample.get("mem") or {}
    if mem.get("total_bytes"):
        total = mem["total_bytes"]
        available = mem.get("available_bytes", 0)
        used_pct = 100.0 * (total - available) / total if total else 0.0
        row("mem_used_pct", host_name, used_pct)
        row("mem_available_bytes", host_name, available)
        row("swap_used_bytes", host_name, mem.get("swap_used_bytes", 0))

    load = sample.get("load") or {}
    for key in ("1m", "5m", "15m"):
        if key in load:
            row(f"load_{key}", host_name, load[key])

    if sample.get("uptime_secs") is not None:
        row("uptime_secs", host_name, sample["uptime_secs"])

    worst_pct = 0.0
    for disk in sample.get("storage") or []:
        total = disk.get("total_bytes", 0)
        used = disk.get("used_bytes", 0)
        pct = 100.0 * used / total if total else 0.0
        mount_res = f"{host_name}:{disk['mount']}"
        row("disk_used_pct", mount_res, pct)
        row("disk_used_bytes", mount_res, used)
        row("disk_total_bytes", mount_res, total)
        if pct > worst_pct:
            worst_pct = pct
    if sample.get("storage"):
        row("disk_used_pct_worst", host_name, worst_pct)

    # Per-core CPU. Sub-resources carry a ':' so the fleet widget filters them out.
    for i, core_pct in enumerate(sample.get("cpu_cores") or []):
        if core_pct is not None:
            row("cpu_pct", f"{host_name}:core{i}", core_pct)

    # Richer memory breakdown.
    if mem.get("cached_bytes") is not None:
        row("mem_cached_bytes", host_name, mem["cached_bytes"])
    if mem.get("buffers_bytes") is not None:
        row("mem_buffers_bytes", host_name, mem["buffers_bytes"])

    # Network throughput — per interface plus a host-level total for the fleet view.
    net_rx_total = net_tx_total = 0.0
    for iface in sample.get("net") or []:
        rx, tx = iface.get("rx_bps", 0.0), iface.get("tx_bps", 0.0)
        res = f"{host_name}:net:{iface['iface']}"
        row("net_rx_bps", res, rx)
        row("net_tx_bps", res, tx)
        net_rx_total += rx
        net_tx_total += tx
    if sample.get("net"):
        row("net_rx_bps", host_name, net_rx_total)
        row("net_tx_bps", host_name, net_tx_total)

    # Disk I/O — per device plus host-level total.
    dr_total = dw_total = 0.0
    for dev in sample.get("diskio") or []:
        rd, wr = dev.get("read_bps", 0.0), dev.get("write_bps", 0.0)
        res = f"{host_name}:diskio:{dev['device']}"
        row("disk_read_bps", res, rd)
        row("disk_write_bps", res, wr)
        dr_total += rd
        dw_total += wr
    if sample.get("diskio"):
        row("disk_read_bps", host_name, dr_total)
        row("disk_write_bps", host_name, dw_total)

    # Temperatures — per sensor plus the host max for at-a-glance.
    max_temp: float | None = None
    for t in sample.get("temps") or []:
        c = t.get("celsius")
        if c is None:
            continue
        row("temp_c", f"{host_name}:temp:{t['label']}", c)
        if max_temp is None or c > max_temp:
            max_temp = c
    if max_temp is not None:
        row("temp_c_max", host_name, max_temp)

    # Pressure stall information (PSI) — host-level gauges (mem/cpu/io some|full).
    for k, v in (sample.get("psi") or {}).items():
        if isinstance(v, (int, float)):
            row(f"psi_{k}", host_name, v)

    return rows


def _error(status: int, code: str, detail: str | None = None) -> tuple[Any, int]:
    body: dict = {"ok": False, "error": code}
    if detail:
        body["detail"] = detail
    return jsonify(body), status


@host_agent_bp.post("/ingest")
async def ingest():
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return _error(401, "invalid_token")
    raw = auth[len("Bearer "):].strip()
    if not raw:
        return _error(401, "invalid_token")

    try:
        payload = await request.get_json(force=True)
    except Exception:
        return _error(400, "malformed_payload", "invalid JSON")
    if not isinstance(payload, dict) or "samples" not in payload:
        return _error(400, "malformed_payload", "missing 'samples'")

    samples = payload.get("samples") or []
    if not isinstance(samples, list) or not samples:
        return _error(400, "malformed_payload", "'samples' must be a non-empty list")

    token_hash = _hash_token(raw)
    async with current_app.db_sessionmaker() as session:
        reg = (await session.execute(
            select(HostAgentRegistration).where(
                HostAgentRegistration.token_hash == token_hash)
        )).scalar_one_or_none()
        if reg is None:
            return _error(401, "invalid_token")
        host = (await session.execute(
            select(Host).where(Host.id == reg.host_id)
        )).scalar_one_or_none()
        if host is None:
            return _error(401, "invalid_token")

        accepted = 0
        latest_ts: datetime | None = None
        docker_snapshots: list[tuple[datetime, list]] = []
        latest_swarm: dict | None = None
        latest_swarm_ts: datetime | None = None
        latest_disk: dict | None = None
        latest_disk_ts: datetime | None = None
        for sample in samples:
            try:
                recorded_at = _parse_ts(sample["ts"])
            except (KeyError, ValueError):
                continue
            metrics = _expand_sample_to_metrics(sample, host.name, recorded_at)
            for m in metrics:
                session.add(m)
            docker = sample.get("docker")
            if isinstance(docker, list) and docker:
                docker_snapshots.append((recorded_at, docker))
            # Swarm is current-state, not time-series — keep only the newest
            # sample's topology (a manager re-reports it every interval).
            swarm = sample.get("swarm")
            if isinstance(swarm, dict) and (
                    latest_swarm_ts is None or recorded_at > latest_swarm_ts):
                latest_swarm, latest_swarm_ts = swarm, recorded_at
            # Disk usage is current-state too — keep only the newest sample's.
            disk = sample.get("docker_disk")
            if isinstance(disk, dict) and (
                    latest_disk_ts is None or recorded_at > latest_disk_ts):
                latest_disk, latest_disk_ts = disk, recorded_at
            accepted += 1
            if latest_ts is None or recorded_at > latest_ts:
                latest_ts = recorded_at

        if accepted == 0:
            await session.rollback()
            return _error(400, "malformed_payload", "no valid samples")

        # Hand host-scoped container data to the docker plugin if it's enabled
        # (opportunistic synergy via the capability registry — no hard import,
        # no-op when docker is disabled). A failure here must never sink the
        # whole ingest, so the metrics above still land.
        if docker_snapshots or latest_swarm is not None or latest_disk is not None:
            from steward.core.capabilities import has_capability, invoke_capability
            if has_capability("docker.persist_host_samples"):
                try:
                    # SAVEPOINT so a docker-side failure rolls back only the
                    # docker writes, leaving the host metrics intact to commit.
                    async with session.begin_nested():
                        await invoke_capability(
                            "docker.persist_host_samples", UserRole.admin,
                            session, host, docker_snapshots, latest_swarm,
                            latest_disk,
                        )
                except Exception:
                    current_app.logger.exception(
                        "host_agent ingest: docker persist failed for host=%s",
                        host.name)

        md = payload.get("metadata") or {}
        changed = False
        if latest_ts and (reg.last_seen_at is None or latest_ts > reg.last_seen_at):
            reg.last_seen_at = latest_ts
            changed = True
        version = payload.get("agent_version")
        if version and reg.agent_version != version:
            reg.agent_version = version
            changed = True
        for field in ("kernel", "distro", "arch"):
            if field in md and getattr(reg, field) != md[field]:
                setattr(reg, field, md[field])
                changed = True
        host_ip = md.get("host_ip")
        if host_ip and reg.host_ip != host_ip:
            reg.host_ip = host_ip
            changed = True
        # Mirror the reported IP into Host.address only when it's blank.
        new_address = pick_host_address(host.address, host_ip)
        if new_address is not None:
            host.address = new_address
        if changed:
            reg.updated_at = datetime.now(timezone.utc)

        if latest_ts:
            skew = abs((datetime.now(timezone.utc) - latest_ts).total_seconds())
            if skew > 300:
                current_app.logger.warning(
                    "host_agent ingest: clock skew %.0fs for host=%s", skew, host.name)

        await session.commit()

    return jsonify({"ok": True, "samples_accepted": accepted}), 200


AGENT_SOURCE_PATH = Path(__file__).parent / "agent.py"


def _agent_version() -> str:
    try:
        for line in AGENT_SOURCE_PATH.read_text().splitlines():
            if line.startswith("AGENT_VERSION"):
                return line.split("=", 1)[1].strip().strip('"').strip("'")
    except OSError:
        pass
    return "unknown"


@host_agent_bp.get("/install.sh")
async def install_script():
    token = request.args.get("token", "")
    if not token:
        return _error(401, "invalid_token")
    async with current_app.db_sessionmaker() as session:
        reg = (await session.execute(
            select(HostAgentRegistration).where(
                HostAgentRegistration.token_hash == _hash_token(token))
        )).scalar_one_or_none()
        if reg is None:
            return _error(401, "invalid_token")
        host = (await session.execute(
            select(Host).where(Host.id == reg.host_id)
        )).scalar_one_or_none()
        if host is None:
            return _error(401, "invalid_token")

    url = public_base_url(request)

    rendered = await render_template(
        "install.sh.j2",
        url=url,
        token=token,
        agent_version=_agent_version(),
        host_name=host.name,
        host_address=host.address,
    )
    return Response(rendered, mimetype="text/plain")


@host_agent_bp.get("/agent.py")
async def agent_source():
    try:
        body = AGENT_SOURCE_PATH.read_text(encoding="utf-8")
    except OSError:
        return _error(500, "agent_missing")
    return Response(body, mimetype="text/x-python")


def _stale_after_seconds() -> int:
    return int(
        current_app.config.get("PLUGINS", {})
        .get(SOURCE_MODULE, {})
        .get("stale_after_seconds", 180)
    )


async def _fleet_rows(session) -> list[dict]:
    """One row per registered host with its latest host-level metrics + stale flag.

    Only bare host-level resources are used here (sub-resources carry a ':' and
    are skipped) so the fleet glance stays one line per host.
    """
    regs = (await session.execute(select(HostAgentRegistration))).scalars().all()
    host_ids = [r.host_id for r in regs]
    hosts = {
        h.id: h for h in (await session.execute(
            select(Host).where(Host.id.in_(host_ids))
        )).scalars().all()
    } if host_ids else {}

    subq = (
        select(
            PluginMetric.resource_name,
            PluginMetric.metric_name,
            func.max(PluginMetric.recorded_at).label("max_ts"),
        )
        .where(PluginMetric.source_module == SOURCE_MODULE)
        .group_by(PluginMetric.resource_name, PluginMetric.metric_name)
    ).subquery()
    latest_rows = (await session.execute(
        select(PluginMetric).join(
            subq,
            (PluginMetric.resource_name == subq.c.resource_name) &
            (PluginMetric.metric_name == subq.c.metric_name) &
            (PluginMetric.recorded_at == subq.c.max_ts),
        ).where(PluginMetric.source_module == SOURCE_MODULE)
    )).scalars().all()

    latest: dict[str, dict[str, float]] = {}
    root_disk: dict[str, float] = {}  # host_name → root (/) disk %
    for row in latest_rows:
        if row.resource_name.endswith(":/") and row.metric_name == "disk_used_pct":
            root_disk[row.resource_name[:-2]] = row.value
        if ":" in row.resource_name:
            continue
        latest.setdefault(row.resource_name, {})[row.metric_name] = row.value

    # Per-host recent series for the inline sparklines (cpu/mem/load host-level,
    # disk from the root mount), so each metric shows its trend beside the value —
    # the host-panel at-a-glance, on the fleet widget. 1h window keeps it cheap.
    host_names = [h.name for h in hosts.values()]
    spark_series: dict[str, dict[str, list]] = {
        n: {"cpu": [], "mem": [], "disk": [], "load": []} for n in host_names
    }
    if host_names:
        since = datetime.now(timezone.utc) - timedelta(hours=1)
        sp_rows = (await session.execute(
            select(PluginMetric).where(
                PluginMetric.source_module == SOURCE_MODULE,
                PluginMetric.recorded_at >= since,
                or_(
                    and_(PluginMetric.resource_name.in_(host_names),
                         PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "load_1m"))),
                    and_(PluginMetric.resource_name.in_([n + ":/" for n in host_names]),
                         PluginMetric.metric_name == "disk_used_pct"),
                ),
            ).order_by(PluginMetric.recorded_at)
        )).scalars().all()
        for r in sp_rows:
            if r.resource_name.endswith(":/"):
                name = r.resource_name[:-2]
                if name in spark_series:
                    spark_series[name]["disk"].append(r.value)
            else:
                key = {"cpu_pct": "cpu", "mem_used_pct": "mem", "load_1m": "load"}.get(r.metric_name)
                if key and r.resource_name in spark_series:
                    spark_series[r.resource_name][key].append(r.value)

    from steward.core.status import sparkline_svg

    stale_after = _stale_after_seconds()
    now = datetime.now(timezone.utc)
    rows = []
    for reg in regs:
        host = hosts.get(reg.host_id)
        if host is None:
            continue
        m = latest.get(host.name, {})
        ls = reg.last_seen_at
        stale = ls is None or (now - ls).total_seconds() > stale_after
        ser = spark_series.get(host.name, {})
        sparks = {
            k: (sparkline_svg(v[-60:], width=72, height=16) if len(v) >= 2 else "")
            for k, v in ser.items()
        }
        rows.append({
            "host": host,
            "reg": reg,
            "stale": stale,
            "cpu_pct": m.get("cpu_pct"),
            "mem_used_pct": m.get("mem_used_pct"),
            "disk_worst": m.get("disk_used_pct_worst"),  # health dot only
            "disk_root": root_disk.get(host.name),       # displayed at-a-glance
            "load_1m": m.get("load_1m"),
            "temp_max": m.get("temp_c_max"),
            "net_rx_bps": m.get("net_rx_bps"),
            "net_tx_bps": m.get("net_tx_bps"),
            "disk_read_bps": m.get("disk_read_bps"),
            "disk_write_bps": m.get("disk_write_bps"),
            "sparks": sparks,
        })
    rows.sort(key=lambda r: r["host"].name.lower())
    return rows


@host_agent_bp.get("/widget")
async def widget_table():
    """Fleet-glance dashboard widget: one row per monitored host, latest metrics."""
    async with current_app.db_sessionmaker() as session:
        rows = await _fleet_rows(session)
    return await render_template("widget_table.html", rows=rows)


@host_agent_bp.get("/widget/history")
async def widget_history():
    host_id = request.args.get("host_id", "")
    try:
        hours = max(1, min(168, int(request.args.get("hours", "6"))))
    except ValueError:
        hours = 6
    # wid (the dashboard widget row id) keeps the <canvas> id unique when several
    # history widgets share a page; fall back to the host id.
    wid = request.args.get("wid", "") or host_id
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)

    series: dict[str, list[list]] = {"cpu_pct": [], "mem_used_pct": [], "disk_root": []}
    host = None
    if host_id:
        async with current_app.db_sessionmaker() as session:
            host = (await session.execute(
                select(Host).where(Host.id == host_id))).scalar_one_or_none()
            if host is not None:
                points = (await session.execute(
                    select(PluginMetric).where(
                        PluginMetric.source_module == SOURCE_MODULE,
                        PluginMetric.recorded_at >= cutoff,
                        or_(
                            and_(PluginMetric.resource_name == host.name,
                                 PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct"))),
                            and_(PluginMetric.resource_name == host.name + ":/",
                                 PluginMetric.metric_name == "disk_used_pct"),
                        ),
                    ).order_by(PluginMetric.recorded_at)
                )).scalars().all()
                # Disk = root (/), consistent with the host panel — not "worst".
                # Epoch-ms x values let the chart use a plain linear axis (no
                # Chart.js date adapter needed), matching the host-detail charts.
                for p in points:
                    key = "disk_root" if p.resource_name != host.name else p.metric_name
                    series[key].append([int(p.recorded_at.timestamp() * 1000), round(p.value, 2)])

    # Downsample: the agent reports every few seconds, so a multi-hour window is
    # hundreds–thousands of points — a noisy, unreadable line. Bucket-average to a
    # readable point count, keeping the shape.
    series = {k: _downsample(v) for k, v in series.items()}

    # Stable y-axis ceiling. All three series are percentages, so snap the top up
    # to the next 10% band (floor 20, cap 100). Auto-scaling to the exact peak made
    # the chart "zoom" jump on every poll as the peak drifted a few %; snapping to
    # a band keeps the axis steady across refreshes while still filling the panel.
    peak = max((p[1] for s in series.values() for p in s), default=0.0)
    y_max = min(100, max(20, ((int(peak) // 10) + 1) * 10))
    return await render_template(
        "widget_history.html", host=host, series=series, hours=hours, wid=wid,
        y_max=y_max,
    )


def _downsample(series: list[list], target: int = 120) -> list[list]:
    """Bucket-average a time series [[epoch_ms, value], …] down to ~target points.

    Consecutive samples are grouped into equal buckets and averaged (x = bucket
    midpoint). Cheap, order-preserving, and smooths the high-frequency noise that
    makes dense agent series hard to read.
    """
    n = len(series)
    if n <= target:
        return series
    bucket = (n + target - 1) // target
    out: list[list] = []
    for i in range(0, n, bucket):
        chunk = series[i:i + bucket]
        mid = chunk[len(chunk) // 2][0]
        avg = sum(p[1] for p in chunk) / len(chunk)
        out.append([mid, round(avg, 2)])
    return out


# Full-metrics refresh cadences. The agent pushes every ~30s by default, so
# polling the current snapshot at 15s keeps it ≤ one cadence stale; the history
# charts move slowly over a multi-hour/day window, so they refresh less often.
_DETAIL_POLL_SECONDS = 15
_CHART_POLL_SECONDS = 60


def _split_host_metrics(host_name: str, latest: dict) -> tuple:
    """Split a host's latest snapshot into
    (hostlvl, cores, nets, disks_io, temps, mounts) for the current-state fragment."""
    hostlvl = latest.get(host_name, {})
    cores: list = []
    nets: dict = {}
    disks_io: dict = {}
    temps: dict = {}
    mounts: dict = {}
    plen = len(host_name) + 1
    for res, metrics in latest.items():
        if res == host_name:
            continue
        suffix = res[plen:]
        if suffix.startswith("core"):
            try:
                idx = int(suffix[4:])
            except ValueError:
                idx = 0
            cores.append((idx, metrics.get("cpu_pct")))
        elif suffix.startswith("net:"):
            nets[suffix[len("net:"):]] = metrics
        elif suffix.startswith("diskio:"):
            disks_io[suffix[len("diskio:"):]] = metrics
        elif suffix.startswith("temp:"):
            temps[suffix[len("temp:"):]] = metrics.get("temp_c")
        else:
            mounts[suffix] = metrics
    cores.sort(key=lambda c: c[0])
    return hostlvl, cores, nets, disks_io, temps, mounts


@host_agent_bp.get("/<host_id>/")
@require_role(UserRole.viewer)
async def host_detail(host_id: str):
    """Full-metrics shell: header + range toggle only. The current-state and
    history sections stream in via HTMX (host_detail_metrics / host_detail_charts)
    so the heavy history query never blocks first paint and the data live-updates."""
    _, range_key = parse_range(request.args.get("range"))
    async with current_app.db_sessionmaker() as session:
        host = (await session.execute(
            select(Host).where(Host.id == host_id))).scalar_one_or_none()
        if host is None:
            return _error(404, "not_found")
    return await render_template(
        "host_detail.html",
        host=host, current_range=range_key, range_options=RANGE_OPTIONS,
        poll_seconds=_DETAIL_POLL_SECONDS, chart_poll_seconds=_CHART_POLL_SECONDS,
    )


@host_agent_bp.get("/<host_id>/metrics")
@require_role(UserRole.viewer)
async def host_detail_metrics(host_id: str):
    """Current-state fragment (gauges/cores/filesystems/IO/temps) — polled live so
    the page shows the latest snapshot the server holds without a full reload."""
    async with current_app.db_sessionmaker() as session:
        host = (await session.execute(
            select(Host).where(Host.id == host_id))).scalar_one_or_none()
        if host is None:
            return "", 404
        reg = (await session.execute(select(HostAgentRegistration).where(
            HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
        latest = await _latest_metrics_for_host(session, host.name)
    hostlvl, cores, nets, disks_io, temps, mounts = _split_host_metrics(host.name, latest)
    ls = reg.last_seen_at if reg else None
    stale = ls is None or (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()
    return await render_template(
        "_host_metrics.html",
        host=host, reg=reg, stale=stale, hostlvl=hostlvl,
        cores=cores, nets=nets, disks_io=disks_io, temps=temps, mounts=mounts,
    )


@host_agent_bp.get("/<host_id>/charts")
@require_role(UserRole.viewer)
async def host_detail_charts(host_id: str):
    """History-charts fragment — lazy-loaded so the date_bin query never blocks
    the page shell; refreshed on a slow cadence and on range change."""
    since, range_key = parse_range(request.args.get("range"))
    async with current_app.db_sessionmaker() as session:
        host = (await session.execute(
            select(Host).where(Host.id == host_id))).scalar_one_or_none()
        if host is None:
            return "", 404
        series = await _history_for_host(session, host.name, since)
    return await render_template("_host_charts.html", series=series, range_key=range_key)


def _new_token_pair() -> tuple[str, str]:
    raw = secrets.token_urlsafe(32)
    return raw, _hash_token(raw)


@host_agent_bp.get("/vitals/<host_id>")
@require_role(UserRole.viewer)
async def host_vitals(host_id: str):
    """Compact live vitals strip (CPU/MEM/DISK/LOAD + pressure) for the Hosts hub.

    Self-contained fragment, polled by hosts/detail.html. Renders nothing until
    the agent reports — the Agent panel below then carries provisioning.
    """
    async with current_app.db_sessionmaker() as db:
        host = (await db.execute(
            select(Host).where(Host.id == host_id))).scalar_one_or_none()
        if host is None:
            return "", 404
        reg = (await db.execute(select(HostAgentRegistration).where(
            HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
        latest = await _latest_metrics_for_host(db, host.name) if reg else {}

        # Recent series for the at-a-glance sparklines (cpu/mem/load host-level,
        # disk from the root mount sub-resource).
        series_raw: dict[str, list[float]] = {"cpu": [], "mem": [], "disk": [], "load": []}
        if reg:
            since = datetime.now(timezone.utc) - timedelta(hours=6)
            spark_rows = (await db.execute(
                select(PluginMetric).where(
                    PluginMetric.source_module == SOURCE_MODULE,
                    PluginMetric.recorded_at >= since,
                    or_(
                        and_(PluginMetric.resource_name == host.name,
                             PluginMetric.metric_name.in_(("cpu_pct", "mem_used_pct", "load_1m"))),
                        and_(PluginMetric.resource_name == host.name + ":/",
                             PluginMetric.metric_name == "disk_used_pct"),
                    ),
                ).order_by(PluginMetric.recorded_at)
            )).scalars().all()
            for r in spark_rows:
                if r.resource_name == host.name:
                    key = {"cpu_pct": "cpu", "mem_used_pct": "mem", "load_1m": "load"}.get(r.metric_name)
                    if key:
                        series_raw[key].append(r.value)
                else:
                    series_raw["disk"].append(r.value)

    from steward.core.status import sparkline_svg
    hostlvl = latest.get(host.name, {})
    # At-a-glance disk = the root filesystem (what people actually care about),
    # not the "worst" mount which is opaque. Worst stays on the full-metrics page.
    disk_root = (latest.get(host.name + ":/", {}) or {}).get("disk_used_pct")
    sparks = {k: sparkline_svg(v[-120:]) for k, v in series_raw.items()}
    # Normalize load by core count so it's comparable across hardware (load/core
    # as %). Cores derived from the per-core CPU sub-resources already collected.
    cores = sum(1 for k in latest if k.startswith(host.name + ":core"))
    load1 = hostlvl.get("load_1m")
    load_per_core = round(load1 / cores * 100) if (cores and load1 is not None) else None
    psi = {
        "cpu": hostlvl.get("psi_cpu_some_avg10"),
        "mem": hostlvl.get("psi_mem_some_avg10"),
        "io": hostlvl.get("psi_io_some_avg10"),
    }
    ls = reg.last_seen_at if reg else None
    reporting = bool(reg) and ls is not None
    stale = reporting and (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()
    return await render_template(
        "_host_vitals.html",
        reg=reg, cpu=hostlvl.get("cpu_pct"), mem=hostlvl.get("mem_used_pct"),
        disk_root=disk_root, load_per_core=load_per_core, sparks=sparks, psi=psi,
        reporting=reporting, stale=stale,
    )


@host_agent_bp.get("/panel/<host_id>")
@require_role(UserRole.viewer)
async def host_panel(host_id: str):
    """Per-host agent management panel embedded into the Hosts hub via HTMX.

    Lifecycle actions (update / rotate / remove / re-provision) when the host
    reports, otherwise provisioning — tied to the host's linked Ansible target.
    The live vitals are a separate strip (host_vitals); this panel is management
    only. Self-contained fragment (no base.html) so it can be hx-swapped in.
    """
    from steward.core.capabilities import has_capability
    from steward.models.ansible_inventory import AnsibleTarget

    async with current_app.db_sessionmaker() as db:
        host = (await db.execute(
            select(Host).where(Host.id == host_id))).scalar_one_or_none()
        if host is None:
            return "", 404
        reg = (await db.execute(select(HostAgentRegistration).where(
            HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
        target = (await db.execute(select(AnsibleTarget).where(
            AnsibleTarget.host_id == host_id))).scalar_one_or_none()

    ls = reg.last_seen_at if reg else None
    # "Deployed" is gated on the agent actually checking in (last_seen_at), NOT on
    # the registration row — that row is minted before the playbook runs, so a
    # failed provision would otherwise look deployed.
    reporting = bool(reg) and ls is not None
    stale = reporting and (datetime.now(timezone.utc) - ls).total_seconds() > _stale_after_seconds()

    ansible_cfg = current_app.config.get("ANSIBLE", {})
    return await render_template(
        "panel.html",
        host=host, reg=reg, reporting=reporting, stale=stale, target=target,
        ansible_available=has_capability("ansible.run_playbook"),
        managed_key_set=bool((ansible_cfg.get("ssh_public_key") or "").strip()),
    )


@host_agent_bp.get("/fleet/")
@require_role(UserRole.admin)
async def fleet():
    from steward.core.capabilities import has_capability
    from steward.models.ansible_inventory import AnsibleTarget, AnsibleGroup

    ansible_available = has_capability("ansible.run_playbook")
    async with current_app.db_sessionmaker() as db:
        regs = (await db.execute(select(HostAgentRegistration))).scalars().all()
        hosts_by_id = {
            h.id: h for h in (await db.execute(
                select(Host).where(Host.id.in_([r.host_id for r in regs])))).scalars().all()
        } if regs else {}
        targets, groups = [], []
        if ansible_available:
            targets = (await db.execute(
                select(AnsibleTarget).order_by(AnsibleTarget.name))).scalars().all()
            groups = (await db.execute(
                select(AnsibleGroup).order_by(AnsibleGroup.name))).scalars().all()

    new_token = request.args.get("new_token")
    new_host_id = request.args.get("host_id")
    install_url = None
    if new_token and new_host_id:
        install_url = f"{public_base_url(request)}/plugins/host_agent/install.sh?token={new_token}"

    managed_key_set = bool(
        (current_app.config.get("ANSIBLE", {}).get("ssh_public_key") or "").strip())

    return await render_template(
        "settings_list.html",
        registrations=[
            {"reg": r, "host": hosts_by_id.get(r.host_id)} for r in regs
        ],
        new_token=new_token,
        install_url=install_url,
        ansible_available=ansible_available,
        deploy_targets=targets,
        deploy_groups=groups,
        managed_key_set=managed_key_set,
    )


@host_agent_bp.post("/fleet/add-host")
@require_role(UserRole.admin)
async def add_host():
    form = await request.form
    name = (form.get("name") or "").strip()
    address = (form.get("address") or "").strip()
    if not name:
        return _error(400, "missing_name")

    async with current_app.db_sessionmaker() as session:
        host = (await session.execute(
            select(Host).where(Host.name == name))).scalar_one_or_none()
        if host is None:
            host = Host(name=name, address=address)
            session.add(host)
            await session.flush()
        existing = (await session.execute(
            select(HostAgentRegistration).where(
                HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
        if existing is not None:
            await session.rollback()
            return _error(400, "already_registered")

        raw, hashed = _new_token_pair()
        reg = HostAgentRegistration(host_id=host.id, token_hash=hashed)
        session.add(reg)
        await session.commit()
        host_id = host.id

    return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}")


@host_agent_bp.post("/fleet/<host_id>/rotate-token")
@require_role(UserRole.admin)
async def rotate_token(host_id: str):
    async with current_app.db_sessionmaker() as session:
        reg = (await session.execute(
            select(HostAgentRegistration).where(
                HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
        if reg is None:
            return _error(404, "not_found")
        raw, hashed = _new_token_pair()
        reg.token_hash = hashed
        reg.token_created_at = datetime.now(timezone.utc)
        await session.commit()
    return redirect(url_for("host_agent.fleet") + f"?new_token={raw}&host_id={host_id}")


@host_agent_bp.post("/fleet/<host_id>/delete")
@require_role(UserRole.admin)
async def delete_registration(host_id: str):
    async with current_app.db_sessionmaker() as session:
        reg = (await session.execute(
            select(HostAgentRegistration).where(
                HostAgentRegistration.host_id == host_id))).scalar_one_or_none()
        if reg is not None:
            await session.delete(reg)
            await session.commit()
    return redirect(url_for("host_agent.fleet"))


# ── Deploy via Ansible (plugin↔core synergy via the capability registry) ──────

async def _ensure_host_for_target(db, target) -> "Host":
    """Find or create the Host that an AnsibleTarget should report under."""
    if getattr(target, "host_id", None):
        h = await db.get(Host, target.host_id)
        if h:
            return h
    h = (await db.execute(
        select(Host).where(Host.name == target.name))).scalar_one_or_none()
    if h is None:
        h = Host(name=target.name, address=getattr(target, "address", "") or "")
        db.add(h)
        await db.flush()
    return h


async def _mint_registration_token(db, host) -> str:
    """Create or rotate the host's agent registration; return the raw token.

    Tokens are stored hashed (unrecoverable), so deploying always mints a fresh
    token — the run installs the agent with it. Existing agents get rotated.
    """
    reg = (await db.execute(select(HostAgentRegistration).where(
        HostAgentRegistration.host_id == host.id))).scalar_one_or_none()
    raw, hashed = _new_token_pair()
    if reg is None:
        db.add(HostAgentRegistration(host_id=host.id, token_hash=hashed))
    else:
        reg.token_hash = hashed
        reg.token_created_at = datetime.now(timezone.utc)
    return raw


@host_agent_bp.post("/deploy")
@require_role(UserRole.admin)
async def deploy_via_ansible():
    """Install/update the agent on Ansible inventory targets via the bundled
    install playbook — the 'curl | sh becomes a button' synergy.

    Uses the core "ansible.run_playbook" capability (no hard import of the
    runner) and injects a freshly-minted per-host token as an inventory hostvar.
    """
    from steward.core.capabilities import has_capability, invoke_capability
    from steward.ansible.inventory_gen import (
        fetch_scope_targets, generate_inventory, inventory_to_yaml)
    from steward.ansible.sources import BUILTIN_SOURCE_NAME

    if not has_capability("ansible.run_playbook"):
        return _error(400, "ansible_unavailable", "Ansible is not available")

    form = await request.form
    scope = (form.get("inventory_scope", "") or "").strip()
    try:
        interval = max(5, int(form.get("agent_interval", "30")))
    except (TypeError, ValueError):
        interval = 30
    if not (scope.startswith("steward:target:")
            or scope.startswith("steward:group:")
            or scope == "steward:all"):
        return _error(400, "bad_scope", "Choose a target or group")

    url = public_base_url(request)
    async with current_app.db_sessionmaker() as db:
        targets = await fetch_scope_targets(db, scope)
        if not targets:
            return _error(400, "no_targets", "No Ansible targets in that scope")
        # The read above autobegins the session transaction (SQLAlchemy 2.0), so
        # mint into it directly and commit — a nested db.begin() would double-begin.
        tokens: dict[str, str] = {}
        for t in targets:
            host = await _ensure_host_for_target(db, t)
            tokens[t.name] = await _mint_registration_token(db, host)
        inv = generate_inventory(targets)  # built while ORM objects are still live
        # Per-host token as a host var; steward_url goes in as an extra-var below.
        for name, tok in tokens.items():
            inv["_meta"]["hostvars"].setdefault(name, {})["steward_token"] = tok
        inventory_content = inventory_to_yaml(inv)
        await db.commit()

    actor_role = UserRole(session.get("user_role", "viewer"))
    run, _source, err = await invoke_capability(
        "ansible.run_playbook", actor_role,
        current_app._get_current_object(),  # type: ignore[attr-defined]
        source_name=BUILTIN_SOURCE_NAME,
        playbook_path="host_agent/install.yml",
        inventory_content=inventory_content,
        inventory_scope=scope,
        # extra-vars outrank the playbook's `vars:` defaults; token stays per-host.
        params={"extra_vars_map": {"steward_url": url, "agent_interval": interval}},
        triggered_by=session.get("user_id"),
    )
    if err:
        return _error(400, "deploy_failed", err)
    return redirect(url_for("ansible.run_detail", run_id=run.id))


@host_agent_bp.post("/provision")
@require_role(UserRole.admin)
async def provision_via_ansible():
    """First-contact provisioning: create the steward account + install the
    managed key + NOPASSWD sudo, then install the agent — over bootstrap
    (password) auth. After this, hosts are reachable with the managed key.

    The bootstrap password is passed to the runner as a connection override and
    is NOT persisted on the run row.
    """
    from steward.core.capabilities import has_capability, invoke_capability
    from steward.ansible.inventory_gen import (
        fetch_scope_targets, generate_inventory, inventory_to_yaml)
    from steward.ansible.sources import BUILTIN_SOURCE_NAME

    if not has_capability("ansible.run_playbook"):
        return _error(400, "ansible_unavailable", "Ansible is not available")

    ansible_cfg = current_app.config.get("ANSIBLE", {})
    pubkey = (ansible_cfg.get("ssh_public_key") or "").strip()
    if not pubkey:
        return _error(400, "no_managed_key",
                      "Generate a managed SSH key in Settings → Ansible first")
    steward_user = (ansible_cfg.get("ssh_user") or "steward").strip() or "steward"

    form = await request.form
    scope = (form.get("inventory_scope", "") or "").strip()
    boot_user = (form.get("bootstrap_user", "") or "").strip()
    boot_password = form.get("bootstrap_password", "") or ""
    try:
        interval = max(5, int(form.get("agent_interval", "30")))
    except (TypeError, ValueError):
        interval = 30
    if not (scope.startswith("steward:target:")
            or scope.startswith("steward:group:")
            or scope == "steward:all"):
        return _error(400, "bad_scope", "Choose a target or group")
    if not boot_user or not boot_password:
        return _error(400, "missing_bootstrap",
                      "Bootstrap user and password are required for first contact")

    url = public_base_url(request)
    async with current_app.db_sessionmaker() as db:
        targets = await fetch_scope_targets(db, scope)
        if not targets:
            return _error(400, "no_targets", "No Ansible targets in that scope")
        # The read above autobegins the session transaction (SQLAlchemy 2.0), so
        # mint into it directly and commit — a nested db.begin() would double-begin.
        tokens: dict[str, str] = {}
        for t in targets:
            host = await _ensure_host_for_target(db, t)
            tokens[t.name] = await _mint_registration_token(db, host)
        inv = generate_inventory(targets)  # built while ORM objects are still live
        # Only the per-host token is a host var; the globals go in as extra-vars
        # below (extra-vars outrank play vars, host vars do not).
        for name, tok in tokens.items():
            inv["_meta"]["hostvars"].setdefault(name, {})["steward_token"] = tok
        inventory_content = inventory_to_yaml(inv)
        await db.commit()

    actor_role = UserRole(session.get("user_role", "viewer"))
    run, _source, err = await invoke_capability(
        "ansible.run_playbook", actor_role,
        current_app._get_current_object(),  # type: ignore[attr-defined]
        source_name=BUILTIN_SOURCE_NAME,
        playbook_path="host_agent/provision.yml",
        inventory_content=inventory_content,
        inventory_scope=scope,
        # extra_vars_map → a JSON -e @file: highest precedence (beats play vars)
        # and space-safe (steward_pubkey carries a comment with a space).
        params={"extra_vars_map": {
            "steward_url": url,
            "steward_pubkey": pubkey,
            "steward_user": steward_user,
            "agent_interval": interval,
        }},
        triggered_by=session.get("user_id"),
        connection={"user": boot_user, "password": boot_password},
    )
    if err:
        return _error(400, "provision_failed", err)
    return redirect(url_for("ansible.run_detail", run_id=run.id))


@host_agent_bp.post("/update")
@require_role(UserRole.admin)
async def update_via_ansible():
    """Refresh agent.py on already-installed hosts via the bundled update
    playbook. Connects as the managed steward account — no token rotation, no
    config rewrite (the host keeps its existing identity)."""
    from steward.core.capabilities import has_capability, invoke_capability
    from steward.ansible.sources import BUILTIN_SOURCE_NAME

    if not has_capability("ansible.run_playbook"):
        return _error(400, "ansible_unavailable", "Ansible is not available")

    form = await request.form
    scope = (form.get("inventory_scope", "") or "").strip()
    if not (scope.startswith("steward:target:")
            or scope.startswith("steward:group:")
            or scope == "steward:all"):
        return _error(400, "bad_scope", "Choose a target or group")

    url = public_base_url(request)
    actor_role = UserRole(session.get("user_role", "viewer"))
    run, _source, err = await invoke_capability(
        "ansible.run_playbook", actor_role,
        current_app._get_current_object(),  # type: ignore[attr-defined]
        source_name=BUILTIN_SOURCE_NAME,
        playbook_path="host_agent/update.yml",
        inventory_scope=scope,
        params={"extra_vars": [f"steward_url={url}"]},
        triggered_by=session.get("user_id"),
    )
    if err:
        return _error(400, "update_failed", err)
    return redirect(url_for("ansible.run_detail", run_id=run.id))