FabledSteward/plugins/docker/routes.py

# plugins/docker/routes.py
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone

from quart import (
    Blueprint, current_app, jsonify, redirect, render_template, request,
    session, url_for,
)
from sqlalchemy import func, select

from steward.auth.middleware import require_role
from steward.models.users import UserRole
from steward.models.hosts import Host
from steward.core.formatting import human_bytes, human_duration, seconds_since
from steward.core.time_range import parse_range, DEFAULT_RANGE
from . import service as docker_svc
from .dedup import dedup_by_container_id
from .swarm_view import (
    build_swarm_services, failing_services, parse_tasks, replaced_tasks, services_problems_first,
    task_summary, trouble_since,
)
from .models import (
    DockerContainer, DockerDiskUsage, DockerHostStatus, DockerImage, DockerMetric,
    DockerSwarmNode, DockerSwarmService,
)

docker_bp = Blueprint("docker", __name__, template_folder="templates")


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


def _uptime(started_at: datetime | None) -> str | None:
    """How long a container has run ('3d 4h'), or None if unknown."""
    secs = seconds_since(started_at)
    return None if secs is None or secs < 0 else human_duration(secs)


def _sparkline(values: list[float], width: int = 80, height: int = 20) -> str:
    if len(values) < 2:
        return f'<svg width="{width}" height="{height}"></svg>'
    mn, mx = min(values), max(values)
    if mx == mn:
        mx = mn + 1.0
    step = width / (len(values) - 1)
    pts = []
    for i, v in enumerate(values):
        x = i * step
        y = height - (v - mn) / (mx - mn) * (height - 2) - 1
        pts.append(f"{x:.1f},{y:.1f}")
    poly = " ".join(pts)
    return (
        f'<svg width="{width}" height="{height}" viewBox="0 0 {width} {height}" '
        f'style="vertical-align:middle;">'
        f'<polyline points="{poly}" fill="none" stroke="#6060c0" stroke-width="1.5"/>'
        f'</svg>'
    )


@docker_bp.get("/")
@require_role(UserRole.viewer)
async def index():
    poll_interval = current_app.config.get("MONITORS_POLL_INTERVAL", 60)
    # Show the Swarm link only when a manager has actually reported topology, so
    # non-swarm installs aren't offered an always-empty page.
    async with current_app.db_sessionmaker() as db:
        has_swarm = (await db.execute(
            select(DockerSwarmService.host_id).limit(1))).first() is not None
        has_disk = (await db.execute(
            select(DockerDiskUsage.host_id).limit(1))).first() is not None
    return await render_template(
        "docker/index.html",
        poll_interval=poll_interval,
        has_swarm=has_swarm,
        has_disk=has_disk,
    )


@docker_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
    """HTMX fragment: what needs attention first, then swarm services
    (collapsed, cluster-complete), then non-swarm containers grouped by host."""
    now = datetime.now(timezone.utc)
    stale_after = docker_svc.stale_after_seconds(current_app.config)

    async with current_app.db_sessionmaker() as db:
        # Hosts whose Docker is not answering: their rows are what it last said.
        docker_down = {d.host_id: d for d in (await db.execute(
            select(DockerHostStatus).where(DockerHostStatus.status != "ok"))).scalars()}
        ranked = await docker_svc.ranked_containers(
            db, now, docker_svc.crash_loop_deaths(current_app.config))
        containers = await docker_svc.list_containers(db)
        services = list((await db.execute(select(DockerSwarmService))).scalars())
        nodes = list((await db.execute(select(DockerSwarmNode))).scalars())
        hosts = await docker_svc.host_map(
            db,
            {c.host_id for c in containers}
            | {n.host_id for n in nodes} | {s.host_id for s in services} | set(docker_down),
        )

        # Swarm tasks collapse into per-service rows (each replica labelled with
        # its host, plus "ghost" replicas for tasks on agent-less nodes, drawn
        # from the managers' placement). Non-swarm containers keep the host view.
        node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname}
        host_name = {h.id: h.name for h in hosts.values()}
        swarm_view = build_swarm_services(
            containers, services, node_hostname, host_name,
            crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config))
        capacity = await docker_svc.host_capacity(db, {c.host_id for c in containers})
        failures = docker_svc.failures_24h(
            await docker_svc.failure_events(db, now - timedelta(hours=24)),
            swarm_view["services"], host_name)

        standalone = [c for c in containers if not c.service_name]
        groups: dict[str, dict] = {}
        for c in standalone:
            g = groups.get(c.host_id)
            if g is None:
                host = hosts.get(c.host_id)
                g = groups[c.host_id] = {
                    "host": host,
                    "host_id": c.host_id,
                    "host_name": host.name if host else c.host_id,
                    "containers": [], "running": 0, "stopped": 0,
                }
            g["containers"].append({
                "container": c,
                "ports": json.loads(c.ports_json) if c.ports_json else [],
                "uptime": _uptime(c.started_at) if c.status == "running" else None,
                "stale": _is_stale(c, now, stale_after, docker_down),
                "usage": docker_svc.usage(c, *capacity.get(c.host_id, (None, None))),
                "exit_failed": docker_svc.exit_is_failure(c.exit_code, c.oom_killed),
            })
            if c.status == "running":
                g["running"] += 1
            else:
                g["stopped"] += 1

    # Sub-group each host's containers by compose project, preserving the
    # running-first ordering. Containers with no project fall into an unlabelled
    # bucket rendered flat, so plain hosts look unchanged.
    for g in groups.values():
        subs: dict[str, dict] = {}
        order: list[str] = []
        for item in g["containers"]:
            c = item["container"]
            label = c.compose_project or None
            key = label or ""
            if key not in subs:
                subs[key] = {"label": label, "containers": []}
                order.append(key)
            subs[key]["containers"].append(item)
        g["subgroups"] = [subs[k] for k in order]
        g["grouped"] = any(s["label"] for s in g["subgroups"])

    host_groups = sorted(groups.values(), key=lambda g: g["host_name"].lower())
    swarm_services = services_problems_first(swarm_view["services"])
    standalone_running = sum(g["running"] for g in host_groups)
    attention = [dict(r, stale=_is_stale(r["container"], now, stale_after, docker_down),
                      label=docker_svc.task_label(r["container"], node_hostname))
                 for r in docker_svc.needs_attention(ranked)]
    failing = failing_services(swarm_services)
    return await render_template(
        "docker/rows.html",
        host_groups=host_groups,
        swarm_services=swarm_services,
        failing_services=failing,
        failures=failures,
        services_healthy=sum(s["state"] == "healthy" for s in swarm_services),
        docker_hosts=len({c.host_id for c in containers}),
        docker_down=sorted(
            ({"host_id": h, "host_name": hosts[h].name if h in hosts else h, "status": d.status,
              "detail": d.detail, "since": d.since} for h, d in docker_down.items()),
            key=lambda d: d["host_name"].lower()),
        attention=attention, says=docker_svc.PROBLEM_SAYS, severity=docker_svc.PROBLEM_SEVERITY,
        unhealthy=sum("unhealthy" in r["problems"] for r in attention),
        restarting=sum(bool({"restarting", "crash_looping"} & set(r["problems"])) for r in attention),
        stale=sum(1 for g in host_groups for i in g["containers"] if i["stale"]),
        running=standalone_running,
        stopped=len(standalone) - standalone_running,
    )


def _is_stale(c, now: datetime, stale_after: int, docker_down=()) -> bool:
    """A container whose last report is older than the agent's staleness
    window, or whose host's Docker is not answering: what the page shows may
    no longer be true."""
    return (c.host_id in docker_down or c.scraped_at is None
            or (now - c.scraped_at).total_seconds() > stale_after)


@docker_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
    """HTMX dashboard widget: swarm services and containers, problems first,
    with how many things failed in the last 24h (linked to the list)."""
    show_stopped = request.args.get("show_stopped", "no") == "yes"
    widget_id = request.args.get("wid", "0")
    now = datetime.now(timezone.utc)

    async with current_app.db_sessionmaker() as db:
        all_containers = dedup_by_container_id(await docker_svc.list_containers(db))
        services = list((await db.execute(select(DockerSwarmService))).scalars())
        nodes = list((await db.execute(select(DockerSwarmNode))).scalars())
        events = await docker_svc.failure_events(db, now - timedelta(hours=24))
        hosts = await docker_svc.host_map(
            db,
            {c.host_id for c in all_containers}
            | {n.host_id for n in nodes} | {s.host_id for s in services},
        )

    # Swarm tasks collapse into per-service rows (each replica tagged with its
    # host); non-swarm containers stay grouped by host.
    node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname}
    host_name = {h.id: h.name for h in hosts.values()}
    swarm_services = services_problems_first(build_swarm_services(
        all_containers, services, node_hostname, host_name,
        crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config))["services"])
    failures = docker_svc.failures_24h(events, swarm_services, host_name)

    standalone = [c for c in all_containers if not c.service_name]
    running = [c for c in standalone if c.status == "running"]

    def troubled(c) -> bool:
        return ((c.status or "") == "restarting" or c.health == "unhealthy"
                or (c.status != "running" and docker_svc.exit_is_failure(c.exit_code, c.oom_killed)))

    # The widget-shape order (UI & Design ruling): what is wrong first, then
    # the healthy rest one line each, collapsing past HEALTHY_SHOWN.
    bad_containers = sorted((c for c in standalone if troubled(c)), key=lambda c: c.name)
    healthy = ([{"kind": "service", "s": s} for s in swarm_services if s["state"] == "healthy"]
               + [{"kind": "container", "c": c} for c in sorted(running, key=lambda c: c.name)
                  if not troubled(c)]
               + ([{"kind": "stopped", "c": c} for c in sorted(standalone, key=lambda c: c.name)
                   if c.status != "running" and not troubled(c)] if show_stopped else []))

    return await render_template(
        "docker/widget.html",
        bad_services=[s for s in swarm_services if s["state"] != "healthy"],
        bad_containers=bad_containers,
        healthy=healthy[:HEALTHY_SHOWN], healthy_more=max(0, len(healthy) - HEALTHY_SHOWN),
        services_total=len(swarm_services),
        services_healthy=sum(s["state"] == "healthy" for s in swarm_services),
        running_count=len(running),
        failures=failures,
        host_name=host_name,
        total_count=len(all_containers) + len(swarm_services),
        widget_id=widget_id,
    )


# Healthy services and containers listed on the dashboard widget before the
# rest collapse to "+N healthy"; the Docker page lists them all.
HEALTHY_SHOWN = 40
FAILURES_SHOWN = 5


CPU_AVERAGE_MINUTES = 5


@docker_bp.get("/widget/resources")
@require_role(UserRole.viewer)
async def widget_resources():
    """HTMX dashboard widget: the containers using the most CPU, averaged over
    the last few minutes so the order holds still between polls, with their
    CPU in cores and memory in bytes, each beside bars of the last few hours
    scaled to the row's own peak (never below a floor, #5214)."""
    limit = max(1, min(20, _arg_number("limit", 10, int)))
    hours = max(1, min(24, _arg_number("hours", 6, int)))
    cpu_floor = max(0.1, _arg_number("cpu_floor", 1.0, float))
    mem_floor = max(0.1, _arg_number("mem_floor", 1.0, float))
    widget_id = request.args.get("wid", "0")
    now = datetime.now(timezone.utc)
    since = now - timedelta(minutes=CPU_AVERAGE_MINUTES)
    slots, width = docker_svc.bar_slots(now, hours)

    async with current_app.db_sessionmaker() as db:
        # Dedup before the limit, else swarm tasks reported by both managers can
        # fill the list with duplicates of the same container.
        running = dedup_by_container_id([
            c for c in await docker_svc.list_containers(db) if c.status == "running"])
        average = {(h, n): v for h, n, v in (await db.execute(
            select(DockerMetric.host_id, DockerMetric.container_name, func.avg(DockerMetric.cpu_pct))
            .where(DockerMetric.scraped_at >= since)
            .group_by(DockerMetric.host_id, DockerMetric.container_name)
        )).all()}

        def cpu_of(c):
            v = average.get((c.host_id, c.name))
            return c.cpu_pct if v is None else float(v)

        busiest = sorted(running, key=lambda c: (-(cpu_of(c) or 0), c.name))[:limit]
        history = await docker_svc.recent_usage(
            db, {(c.host_id, c.name) for c in busiest}, slots[0], width)
        hosts = await docker_svc.host_map(db, {c.host_id for c in busiest})
        capacity = await docker_svc.host_capacity(db, {c.host_id for c in busiest})

    def bars(c, i, scale, floor):
        seen = history.get((c.host_id, c.name), {})
        values = [None if seen.get(t) is None or seen[t][i] is None else seen[t][i] * scale
                  for t in slots]
        return docker_svc.scaled_bars(values, floor)

    rows_data = [{
        "c": c,
        "label": docker_svc.display_name(c),
        "host_name": hosts[c.host_id].name if c.host_id in hosts else c.host_id,
        "usage": docker_svc.usage(c, *capacity.get(c.host_id, (None, None)), cpu_pct=cpu_of(c)),
        "cpu_bars": bars(c, 0, 0.01, cpu_floor),            # cpu_pct → cores
        "mem_bars": bars(c, 1, 1.0, mem_floor * 2**30),     # bytes; floor in GiB
    } for c in busiest]
    return await render_template(
        "docker/widget_resources.html",
        rows=rows_data,
        minutes=CPU_AVERAGE_MINUTES,
        hours=hours, cpu_floor=cpu_floor, mem_floor=mem_floor * 2**30,
        widget_id=widget_id,
    )


def _arg_number(key, default, kind):
    """A numeric query argument, or `default` when missing or not a number."""
    try:
        return kind(request.args.get(key, default))
    except (TypeError, ValueError):
        return default


@docker_bp.get("/host/<host_id>")
@require_role(UserRole.viewer)
async def host_panel(host_id: str):
    """Per-host Docker fragment embedded on the core Hosts hub via HTMX:
    problems first (with exit code and since when), then running, then
    stopped. Swarm tasks Swarm has already replaced are history and left out.

    Renders nothing when the host reports no containers, so hosts without Docker
    don't carry an empty card.
    """
    now = datetime.now(timezone.utc)
    async with current_app.db_sessionmaker() as db:
        containers = await docker_svc.list_containers(db, host_id)
        if not containers:
            return ""
        services = list((await db.execute(select(DockerSwarmService))).scalars())
        nodes = list((await db.execute(select(DockerSwarmNode))).scalars())
        failures = await docker_svc.recent_failures_by_container(db, now - timedelta(hours=24))
        capacity = await docker_svc.host_capacity(db, {host_id})

    replaced = replaced_tasks(containers, services)
    node_hostname = {n.node_id: n.hostname for n in nodes if n.hostname}
    hour_ago = now - timedelta(hours=1)
    crash = docker_svc.crash_loop_deaths(current_app.config)
    rows = []
    for c in containers:
        if (c.host_id, c.name) in replaced:
            continue
        events = failures.get((c.host_id, c.name), [])
        recent = sum(1 for e in events if e.at >= hour_ago)
        problems = docker_svc.container_problems(c, recent, bool(events), crash)
        rows.append({
            "c": c, "label": docker_svc.task_label(c, node_hostname), "problems": problems,
            "since": docker_svc.problem_since(problems, events, hour_ago),
            "usage": docker_svc.usage(c, *capacity.get(host_id, (None, None))),
            "uptime": _uptime(c.started_at) if c.status == "running" else None,
        })
    rows.sort(key=lambda r: (
        docker_svc.PROBLEMS.index(r["problems"][0]) if r["problems"] else len(docker_svc.PROBLEMS),
        r["c"].status != "running", r["label"].lower()))
    running = sum(1 for r in rows if r["c"].status == "running")
    return await render_template(
        "docker/host_panel.html",
        rows=rows, says=docker_svc.PROBLEM_SAYS, severity=docker_svc.PROBLEM_SEVERITY,
        running=running, stopped=len(rows) - running,
        problems=sum(1 for r in rows if r["problems"]),
        replaced=len(replaced),
    )


# Glyph + colour per lifecycle event, so the timeline reads at a glance.
_EVENT_STYLE = {
    "start":         ("▲", "var(--green)"),
    "restart":       ("↻", "var(--yellow)"),
    "stop":          ("■", "var(--text-muted)"),
    "die":           ("✕", "var(--red)"),
    "oom":           ("☠", "var(--red)"),
    "health_change": ("◐", "var(--orange)"),
}


@docker_bp.get("/container/<host_id>/<name>")
@require_role(UserRole.viewer)
async def container_detail(host_id: str, name: str):
    """Full detail page for one container: facts, resource history, lifecycle."""
    async with current_app.db_sessionmaker() as db:
        c = await db.get(DockerContainer, (host_id, name))
        host = await db.get(Host, host_id)
        events = await docker_svc.container_events(db, host_id, name) if c is not None else []
        node_name = None
        if c is not None and c.node_id:
            node_name = (await db.execute(
                select(DockerSwarmNode.hostname)
                .where(DockerSwarmNode.node_id == c.node_id, DockerSwarmNode.hostname != "")
                .limit(1)
            )).scalar()
        cpus, mem_total = (await docker_svc.host_capacity(db, [host_id])).get(host_id, (None, None)) \
            if c is not None else (None, None)

    if c is None:
        return await render_template(
            "docker/container_detail.html",
            container=None, host_id=host_id, name=name,
        ), 404

    return await render_template(
        "docker/container_detail.html",
        container=c,
        host=host,
        host_id=host_id,
        name=name,
        label=docker_svc.task_label(c, {c.node_id: node_name} if node_name else None),
        node_name=node_name,
        usage=docker_svc.usage(c, cpus, mem_total),
        exit_failed=docker_svc.exit_is_failure(c.exit_code, c.oom_killed),
        ports=json.loads(c.ports_json) if c.ports_json else [],
        uptime=_uptime(c.started_at) if c.status == "running" else None,
        net_rx=human_bytes(c.net_rx_bytes), net_tx=human_bytes(c.net_tx_bytes),
        blk_read=human_bytes(c.blk_read_bytes), blk_write=human_bytes(c.blk_write_bytes),
        events=[
            {"event": e.event, "detail": e.detail, "at": e.at,
             "glyph": _EVENT_STYLE.get(e.event, ("•", "var(--text-muted)"))[0],
             "colour": _EVENT_STYLE.get(e.event, ("•", "var(--text-muted)"))[1]}
            for e in events
        ],
        current_range=request.args.get("range", DEFAULT_RANGE),
    )


@docker_bp.get("/swarm")
@require_role(UserRole.viewer)
async def swarm():
    """Swarm topology page: services with replica health, nodes, placement.

    Swarm services/nodes are cluster-global — every manager's API returns the
    same list — but each manager reports them independently (rows keyed by
    host_id). So we group the reporting managers into swarms (managers that share
    any node_id are the same cluster) and dedup within each: one row per service
    (by name) and per node (by node_id), keeping the freshest. Without this, two
    managers in one cluster list every service/node twice.
    """
    async with current_app.db_sessionmaker() as db:
        services = list((await db.execute(
            select(DockerSwarmService).order_by(DockerSwarmService.service_name)
        )).scalars())
        nodes = list((await db.execute(
            select(DockerSwarmNode).order_by(
                DockerSwarmNode.role, DockerSwarmNode.hostname)
        )).scalars())
        hosts = await docker_svc.host_map(db, {s.host_id for s in services} | {n.host_id for n in nodes})

    # ── Group reporting managers into swarms by shared node membership ──────────
    manager_ids = {s.host_id for s in services} | {n.host_id for n in nodes}
    nodes_by_mgr: dict[str, set[str]] = {}
    for n in nodes:
        nodes_by_mgr.setdefault(n.host_id, set()).add(n.node_id)

    parent = {m: m for m in manager_ids}

    def _find(x: str) -> str:
        while parent[x] != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    mgrs = list(manager_ids)
    for i in range(len(mgrs)):
        for j in range(i + 1, len(mgrs)):
            a, b = mgrs[i], mgrs[j]
            if nodes_by_mgr.get(a) and nodes_by_mgr.get(b) and (nodes_by_mgr[a] & nodes_by_mgr[b]):
                parent[_find(a)] = _find(b)

    swarm_members: dict[str, set[str]] = {}
    for m in manager_ids:
        swarm_members.setdefault(_find(m), set()).add(m)

    # ── Dedup services (by name) + nodes (by node_id) within each swarm ─────────
    swarms = []
    for members in swarm_members.values():
        svc_by_name: dict[str, DockerSwarmService] = {}
        for s in services:
            if s.host_id in members and (
                s.service_name not in svc_by_name
                or s.updated_at > svc_by_name[s.service_name].updated_at
            ):
                svc_by_name[s.service_name] = s
        node_by_id: dict[str, DockerSwarmNode] = {}
        for n in nodes:
            if n.host_id in members and (
                n.node_id not in node_by_id
                or n.updated_at > node_by_id[n.node_id].updated_at
            ):
                node_by_id[n.node_id] = n

        node_names = {nid: (nrow.hostname or nid) for nid, nrow in node_by_id.items()}
        now = datetime.now(timezone.utc)
        deaths = docker_svc.crash_loop_deaths(current_app.config)
        svc_dicts = []
        for s in svc_by_name.values():
            tasks = parse_tasks(s)
            summary = task_summary(tasks, now, deaths, node_names)
            state = ("healthy" if s.running >= s.desired and s.desired > 0
                     else "degraded" if s.running > 0 else "down")
            placement = []
            for p in (json.loads(s.placement_json) if s.placement_json else []):
                nid = p.get("node_id", "")
                placement.append({
                    "node": node_names.get(nid, nid[:12] or "?"),
                    "running": p.get("running", 0),
                })
            svc_dicts.append({
                "name": s.service_name, "mode": s.mode,
                "running": s.running, "desired": s.desired, "image": s.image,
                "healthy": s.running >= s.desired and s.desired > 0,
                "degraded": 0 < s.running < s.desired,
                "down": s.running == 0 and s.desired > 0,
                "state": state,
                "placement": placement,
                **summary,
                "trouble_since": trouble_since(tasks, state, summary["crash_looping"], now),
            })
        # Problems first: a node that is down or not taking work, then by role.
        node_rows = sorted(
            node_by_id.values(), key=lambda n: (
                n.status == "ready" and n.availability == "active",
                n.role, (n.hostname or n.node_id).lower())
        )
        manager_names = sorted(hosts[m].name if m in hosts else m for m in members)
        swarms.append({
            "managers": manager_names,
            "services": services_problems_first(svc_dicts),
            "nodes": node_rows,
        })

    swarms.sort(key=lambda g: g["managers"][0].lower() if g["managers"] else "")
    return await render_template("docker/swarm.html", swarms=swarms)


# How many images a host agent reports, largest first (host_agent agent.py
# _disk_images caps its list here); a list this long may be missing smaller ones.
IMAGES_REPORTED = 50


@docker_bp.get("/disk")
@require_role(UserRole.viewer)
async def disk():
    """Image/disk usage page: reclaimable space, per-image sizes, stopped count.

    Admins can reclaim space per host via the prune buttons, which fire the
    audited bundled prune playbook through Ansible (m78) — the collection agent
    itself stays read-only.
    """
    from steward.core.capabilities import has_capability
    from steward.models.ansible_inventory import AnsibleTarget

    async with current_app.db_sessionmaker() as db:
        summaries = list((await db.execute(select(DockerDiskUsage))).scalars())
        images = list((await db.execute(
            select(DockerImage).order_by(DockerImage.size.desc())
        )).scalars())
        # Stopped-container count per host, surfaced alongside reclaimable space.
        stopped_rows = (await db.execute(
            select(DockerContainer.host_id)
            .where(DockerContainer.status != "running")
        )).scalars()
        stopped_by_host: dict[str, int] = {}
        for hid in stopped_rows:
            stopped_by_host[hid] = stopped_by_host.get(hid, 0) + 1
        hosts = await docker_svc.host_map(db, {s.host_id for s in summaries})
        # Which hosts have a linked Ansible target — gates the prune buttons
        # (a target is required to route the playbook run to that host).
        host_ids = {s.host_id for s in summaries}
        target_by_host: dict[str, str] = {}
        if host_ids:
            for hid, tid in (await db.execute(
                select(AnsibleTarget.host_id, AnsibleTarget.id)
                .where(AnsibleTarget.host_id.in_(host_ids))
            )).all():
                target_by_host[hid] = tid

    images_by_host: dict[str, list] = {}
    for im in images:
        images_by_host.setdefault(im.host_id, []).append({
            "repo_tag": None if im.repo_tag in ("<none>", "", None) else im.repo_tag,
            "image_id": im.image_id, "size": human_bytes(im.size),
            "shared": human_bytes(im.shared_size), "containers": im.containers,
            "reclaimable": im.containers == 0,
        })

    host_groups = [
        {
            "host_id": s.host_id,
            "host_name": hosts[s.host_id].name if s.host_id in hosts else s.host_id,
            "as_of": s.updated_at,
            "summary": {
                "images_total": s.images_total, "images_active": s.images_active,
                "images_size": human_bytes(s.images_size),
                "images_reclaimable": human_bytes(s.images_reclaimable),
                "layers_size": human_bytes(s.layers_size),
                "containers_count": s.containers_count,
                "containers_size": human_bytes(s.containers_size),
                "volumes_count": s.volumes_count,
                "volumes_size": human_bytes(s.volumes_size),
                "build_cache_size": human_bytes(s.build_cache_size),
            },
            "stopped": stopped_by_host.get(s.host_id, 0),
            "images": images_by_host.get(s.host_id, []),
            # The agent reports the largest IMAGES_REPORTED images (agent._disk_images).
            "images_capped": len(images_by_host.get(s.host_id, [])) >= IMAGES_REPORTED,
            "has_target": s.host_id in target_by_host,
        }
        for s in summaries
    ]
    host_groups.sort(key=lambda g: g["host_name"].lower())
    return await render_template(
        "docker/disk.html", host_groups=host_groups,
        ansible_available=has_capability("ansible.run_playbook"),
    )


@docker_bp.post("/disk/<host_id>/prune")
@require_role(UserRole.admin)
async def disk_prune(host_id: str):
    """Reclaim Docker disk on a host by firing the bundled prune playbook via
    Ansible (audited, admin-gated) — the collection agent stays read-only (m78).
    `target` selects the scope: containers | images | system.
    """
    from steward.core.capabilities import has_capability, invoke_capability
    from steward.ansible.sources import BUILTIN_SOURCE_NAME
    from steward.models.ansible_inventory import AnsibleTarget

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

    form = await request.form
    prune_target = (form.get("target", "") or "").strip()
    if prune_target not in docker_svc.PRUNE_TARGETS:
        return _error(400, "bad_target", "Unknown prune target")

    async with current_app.db_sessionmaker() as db:
        target = (await db.execute(
            select(AnsibleTarget).where(AnsibleTarget.host_id == host_id)
        )).scalar_one_or_none()
    if target is None:
        return _error(400, "no_target",
                      "Link an Ansible target to this host before pruning")

    # extra_vars_map outranks the playbook's own `vars:` defaults.
    extra_vars = docker_svc.prune_extra_vars(prune_target)

    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="maintenance/docker_prune.yml",
        inventory_scope=f"steward:target:{target.id}",
        params={"extra_vars_map": extra_vars},
        triggered_by=session.get("user_id"),
    )
    if err:
        return _error(400, "prune_failed", err)
    return redirect(url_for("ansible.run_detail", run_id=run.id))


@docker_bp.get("/container/<host_id>/<name>/history")
@require_role(UserRole.viewer)
async def container_history(host_id: str, name: str):
    """HTMX fragment: CPU + memory sparklines for the selected time range."""
    since, range_key = parse_range(request.args.get("range"))
    async with current_app.db_sessionmaker() as db:
        cpu_hist, mem_hist = await docker_svc.container_history(db, host_id, name, since)
    return await render_template(
        "docker/_container_history.html",
        sparkline_cpu=_sparkline(cpu_hist, width=320, height=48),
        sparkline_mem=_sparkline(mem_hist, width=320, height=48),
        have_data=len(cpu_hist) >= 2,
        range_key=range_key,
    )


# Newest lines returned per fetch — the retained window is bounded by the ring,
# but a container can still hold thousands of lines; cap what one fetch renders.
_LOG_TAIL_DEFAULT = 500


@docker_bp.get("/container/<host_id>/<name>/logs")
@require_role(UserRole.viewer)
async def container_logs(host_id: str, name: str):
    """Full log-viewer page for one container (lines load via an HTMX fragment)."""
    async with current_app.db_sessionmaker() as db:
        host = await db.get(Host, host_id)
    return await render_template(
        "docker/container_logs.html", host=host, host_id=host_id, name=name,
    )


@docker_bp.get("/container/<host_id>/<name>/logs/lines")
@require_role(UserRole.viewer)
async def container_logs_lines(host_id: str, name: str):
    """HTMX fragment: the recent retained log lines, stream + text filtered.
    Polled every few seconds by the viewer for near-live follow."""
    stream = request.args.get("stream", "all")
    query = (request.args.get("q") or "").strip()
    async with current_app.db_sessionmaker() as db:
        rows = await docker_svc.query_logs(db, host_id, name, stream, query, _LOG_TAIL_DEFAULT)
    return await render_template(
        "docker/_container_logs_lines.html",
        lines=[{"ts": r.ts, "stream": r.stream, "line": r.line} for r in rows],
        name=name, tail=_LOG_TAIL_DEFAULT,
        filtered=bool(query) or stream in ("stdout", "stderr"),
    )