# 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 import anomalies as anom
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 .swarm_nodes import node_cards
from .overview import (
    DOCKER_SORTS, DOCKER_STATES, NO_STACK, container_row, filter_rows, service_row, sort_rows,
    stack_of,
)
from .models import (
    DockerContainer, DockerDiskUsage, DockerHostStatus, DockerImage,
    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 _filters(args) -> dict:
    """The Docker page's search, filters and sort from the query string
    (#5299): q, stack, host, state, sort, dir; anything unknown is dropped."""
    sort = args.get("sort") if args.get("sort") in DOCKER_SORTS else "status"
    desc = {"desc": True, "asc": False}.get(args.get("dir"), DOCKER_SORTS[sort])
    return {"q": (args.get("q") or "").strip(), "stack": args.get("stack", ""),
            "host": args.get("host", ""),
            "state": args.get("state") if args.get("state") in DOCKER_STATES else "",
            "sort": sort, "dir": "desc" if desc else "asc"}


def _page_url(f: dict, **change) -> str:
    """The Docker page with some filters changed; defaults are left out."""
    from urllib.parse import urlencode
    want = {**f, **change}
    if want["dir"] == ("desc" if DOCKER_SORTS[want["sort"]] else "asc"):
        want["dir"] = ""
    if want["sort"] == "status":
        want["sort"] = ""
    query = urlencode({k: v for k, v in want.items() if v})
    return "/plugins/docker/" + ("?" + query if query else "")


def _sort_url(f: dict, key: str) -> str:
    """A column header's link: its own default direction, or the other way
    when it is already the sort."""
    if key == f["sort"]:
        return _page_url(f, sort=key, dir="asc" if f["dir"] == "desc" else "desc")
    return _page_url(f, sort=key, dir="desc" if DOCKER_SORTS[key] else "asc")


@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
        choices = await _filter_choices(db)
        # One line per host under the tables: what Docker keeps on its disk and
        # how much that grew in a week (#5306); the Disk page has the rest.
        disk = await docker_svc.disk_hosts(
            db, datetime.now(timezone.utc) - timedelta(days=7), points=14) if has_disk else []
    return await render_template(
        "docker/index.html",
        poll_interval=poll_interval,
        has_swarm=has_swarm,
        has_disk=has_disk, disk=disk,
        f=_filters(request.args), choices=choices, states=DOCKER_STATES, no_stack=NO_STACK,
    )


async def _filter_choices(db) -> dict:
    """The stacks and hosts the filter bar offers: every swarm service's stack
    and compose project, and every host or node that runs something."""
    names = set((await db.execute(select(DockerSwarmService.service_name).distinct())).scalars())
    names |= {n for n in (await db.execute(
        select(DockerContainer.service_name).distinct())).scalars() if n}
    stacks = {stack_of(n) for n in names} | {p for p in (await db.execute(
        select(DockerContainer.compose_project).distinct())).scalars() if p}
    host_ids = set((await db.execute(select(DockerContainer.host_id).distinct())).scalars())
    hosts = {h.name for h in (await docker_svc.host_map(db, host_ids)).values()}
    hosts |= {n for n in (await db.execute(select(DockerSwarmNode.hostname).distinct())).scalars() if n}
    return {"stacks": sorted((s for s in stacks if s), key=str.lower),
            "hosts": sorted(hosts, key=str.lower)}


async def _swarm_context(db, now: datetime) -> dict:
    """What the Docker page and a service's page both build from: containers,
    swarm services and nodes, host names, and the swarm view (each service with
    its current replicas, Swarm's task records and their failures)."""
    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})
    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, now=now,
        crash_loop_deaths=docker_svc.crash_loop_deaths(current_app.config))
    return {"containers": containers, "services": services, "nodes": nodes, "hosts": hosts,
            "node_hostname": node_hostname, "host_name": host_name, "swarm_view": swarm_view}


@docker_bp.get("/rows")
@require_role(UserRole.viewer)
async def rows():
    """HTMX fragment: a verdict line and the problems that aren't zero, what
    needs attention, what failed in 24h (each with its logs), then the swarm
    services and the standalone containers as two tables that share the
    filter bar's search, filters and sort (#5298–#5301)."""
    now = datetime.now(timezone.utc)
    stale_after = docker_svc.stale_after_seconds(current_app.config)
    f = _filters(request.args)

    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))
        ctx = await _swarm_context(db, now)
        containers, hosts = ctx["containers"], ctx["hosts"]
        hosts.update(await docker_svc.host_map(db, set(docker_down) - set(hosts)))
        host_name = {h.id: h.name for h in hosts.values()}
        swarm_view = ctx["swarm_view"]
        capacity = await docker_svc.host_capacity(db, {c.host_id for c in containers})
        norms = await docker_svc.container_norms(
            db, {(c.host_id, c.name) for c in containers if not c.service_name}, now)
        failures = docker_svc.failures_24h(
            await docker_svc.failure_events(db, now - timedelta(hours=24)),
            swarm_view["services"], host_name)

    thresholds = current_app.config.get("THRESHOLDS", {})
    service_rows = [service_row(s, now) for s in swarm_view["services"]]
    standalone = [c for c in containers if not c.service_name]
    container_rows = [
        container_row(
            c, label=docker_svc.display_name(c), host=host_name.get(c.host_id, c.host_id),
            usage=docker_svc.usage(c, *capacity.get(c.host_id, (None, None)), thresholds=thresholds,
                                   norms=norms.get((c.host_id, c.name))),
            ports=json.loads(c.ports_json) if c.ports_json else [],
            stale=_is_stale(c, now, stale_after, docker_down),
            exit_failed=docker_svc.exit_is_failure(c.exit_code, c.oom_killed), now=now)
        for c in standalone]
    desc = f["dir"] == "desc"
    shown_services = sort_rows(filter_rows(service_rows, q=f["q"], stack=f["stack"], host=f["host"],
                                           state=f["state"]), f["sort"], desc)
    shown_containers = sort_rows(filter_rows(container_rows, q=f["q"], stack=f["stack"],
                                             host=f["host"], state=f["state"]), f["sort"], desc)

    failing = failing_services(swarm_view["services"])
    attention = [dict(r, stale=_is_stale(r["container"], now, stale_after, docker_down),
                      label=docker_svc.task_label(r["container"], ctx["node_hostname"]))
                 for r in docker_svc.needs_attention(ranked)]
    running = sum(r["running"] for r in container_rows)
    return await render_template(
        "docker/rows.html",
        services=shown_services, containers=shown_containers,
        services_total=len(service_rows), containers_total=len(container_rows),
        services_healthy=sum(s["state"] == "healthy" for s in service_rows),
        failing_services=failing, failures=failures,
        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(r["stale"] for r in container_rows),
        running=running, stopped=len(container_rows) - running,
        f=f, sort_url=lambda key: _sort_url(f, key), page_url=lambda **kw: _page_url(f, **kw),
        filtered=any(f[k] for k in ("q", "stack", "host", "state")),
    )


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 = await docker_svc.average_cpu(db, {(c.host_id, c.name) for c in running}, since)

        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})
        norms = await docker_svc.container_norms(db, {(c.host_id, c.name) for c in busiest})
    thresholds = current_app.config.get("THRESHOLDS", {})

    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),
                                  thresholds=thresholds, norms=norms.get((c.host_id, c.name))),
        "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})
        norms = await docker_svc.container_norms(db, {(c.host_id, c.name) for c in containers}, now)

    thresholds = current_app.config.get("THRESHOLDS", {})
    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)), thresholds=thresholds,
                                      norms=norms.get((c.host_id, c.name))),
            "uptime": _uptime(c.started_at) if c.status == "running" else None,
        })
    shown, folded = host_panel_rows(rows)
    running = sum(1 for r in rows if r["c"].status == "running")
    return await render_template(
        "docker/host_panel.html",
        rows=shown, folded=folded, 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), busiest=HOST_PANEL_BUSIEST,
    )


# Running containers shown before the rest fold away on a host's page (#5276).
HOST_PANEL_BUSIEST = 8


def host_panel_rows(rows: list) -> tuple[list, list]:
    """(shown, folded) for a host's Docker panel: every container with a
    problem, worst first; then the busiest running ones by CPU. The other
    running containers (busiest first) and the stopped ones (by name) fold
    away, so a host with dozens of containers keeps a short card."""
    def cpu(r):
        return getattr(r["c"], "cpu_pct", None) or 0.0
    problems = sorted((r for r in rows if r["problems"]), key=lambda r: (
        docker_svc.PROBLEMS.index(r["problems"][0]), r["label"].lower()))
    running = sorted((r for r in rows if not r["problems"] and r["c"].status == "running"),
                     key=lambda r: (-cpu(r), r["label"].lower()))
    stopped = sorted((r for r in rows if not r["problems"] and r["c"].status != "running"),
                     key=lambda r: r["label"].lower())
    return (problems + running[:HOST_PANEL_BUSIEST],
            running[HOST_PANEL_BUSIEST:] + stopped)


# Glyph + tone per lifecycle event, so the timeline reads at a glance.
_EVENT_STYLE = {
    "start":         ("▲", "ok"),
    "restart":       ("↻", "warn"),
    "stop":          ("■", "muted"),
    "die":           ("✕", "crit"),
    "oom":           ("☠", "crit"),
    "health_change": ("◐", "warn"),
}


@docker_bp.get("/container/<host_id>/<name>")
@require_role(UserRole.viewer)
async def container_detail(host_id: str, name: str):
    """One container (#5310): a header that says whether it is well and what it
    is, why it stopped only while it has, what it moved since it started, its
    CPU and memory over time, and its 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)
        # Its CPU and memory against their normal, and its anomalies (#5742).
        norms = (await docker_svc.container_norms(db, {(host_id, name)})).get((host_id, name))
        resource = docker_svc.container_resource(host.name, name) if host is not None else None
        anomalies = await anom.views(db, await anom.list_anomalies(
            db, source="docker", resource=resource,
            since=datetime.now(timezone.utc) - timedelta(days=30), limit=20),
            links=False) if resource else []

    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, thresholds=current_app.config.get("THRESHOLDS", {}),
                               norms=norms),
        anomalies=anomalies, anomaly_back=f"/plugins/docker/container/{host_id}/{name}",
        anomaly_resource=resource,
        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, ("•", "muted"))[0],
             "tone": _EVENT_STYLE.get(e.event, ("•", "muted"))[1]}
            for e in events
        ],
        died_at=next((e.at for e in events if e.event in ("die", "oom")), None),
        current_range=request.args.get("range", DEFAULT_RANGE),
    )


@docker_bp.get("/swarm")
@require_role(UserRole.viewer)
async def swarm():
    """The Swarm page, laid out by node (#5303): each node's role, whether it
    takes work, what its containers use against what the machine has, and the
    services placed on it; services short of replicas first.

    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})
        containers = await docker_svc.list_containers(db)
        # The Steward host on each node: the one whose agent reports its
        # containers, else the host named as the node is.
        node_host = {c.node_id: c.host_id for c in containers if c.node_id}
        unmatched = {(n.hostname or "").lower() for n in nodes if n.node_id not in node_host} - {""}
        if unmatched:
            by_name = {h.name.lower(): h for h in (await db.execute(
                select(Host).where(func.lower(Host.name).in_(unmatched)))).scalars()}
            for n in nodes:
                h = by_name.get((n.hostname or "").lower())
                if n.node_id not in node_host and h is not None:
                    node_host[n.node_id] = h.id
        hosts.update(await docker_svc.host_map(db, set(node_host.values()) - set(hosts)))
        capacity = await docker_svc.host_capacity(db, set(node_host.values()))

    # ── 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),
            })
        manager_names = sorted(hosts[m].name if m in hosts else m for m in members)
        swarms.append({
            "managers": manager_names,
            "services_total": len(svc_dicts),
            "services_healthy": sum(d["state"] == "healthy" for d in svc_dicts),
            "troubled": failing_services(svc_dicts),
            "nodes": node_cards(
                node_by_id.values(), svc_by_name.values(), containers, node_host=node_host,
                host_names={h.id: h.name for h in hosts.values()}, capacity=capacity,
                service_state={d["name"]: d["state"] for d in svc_dicts}),
        })

    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():
    """What Docker keeps on each host's disk (#5304, #5305): the hosts side by
    side with how much each grew over the selected range (the /disk/hosts
    fragment), then each host's parts, cleanup and images by repository.

    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
    from .disk_view import group_images

    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": im.repo_tag, "image_id": im.image_id, "size": im.size,
            "shared_size": im.shared_size, "containers": im.containers})

    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": s,
            "total": s.images_size + s.containers_size + s.volumes_size + s.build_cache_size,
            "stopped": stopped_by_host.get(s.host_id, 0),
            "repos": group_images(images_by_host.get(s.host_id, [])),
            "images_listed": len(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["total"], g["host_name"].lower()))
    return await render_template(
        "docker/disk.html", host_groups=host_groups,
        ansible_available=has_capability("ansible.run_playbook"),
        current_range=request.args.get("range", "7d"),
    )


@docker_bp.get("/disk/hosts")
@require_role(UserRole.viewer)
async def disk_hosts_fragment():
    """HTMX fragment: every Docker host's disk side by side, with how much its
    total grew over the selected range and a line of it (#5304)."""
    from steward.core.status import sparkline_svg
    since, range_key = parse_range(request.args.get("range") or "7d")
    async with current_app.db_sessionmaker() as db:
        rows = await docker_svc.disk_hosts(db, since)
    for r in rows:
        r["spark"] = sparkline_svg([v for _, v in r["points"]], width=120, height=24,
                                   stroke="currentColor")
    return await render_template("docker/_disk_hosts.html", rows=rows, range_key=range_key)


@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: the container's CPU (cores) and memory over the
    selected range, with the latest and the peak (#5310)."""
    since, range_key = parse_range(request.args.get("range"))
    async with current_app.db_sessionmaker() as db:
        series = await docker_svc.service_history(db, [(host_id, name)], since)
    return await _usage_history(series, range_key)


# A history line is drawn against at least this much, so a container idling
# near zero draws flat (UI & Design ruling 8, #5310).
HISTORY_CPU_FLOOR = 1.0           # cores
HISTORY_MEM_FLOOR = 1 << 30       # bytes


async def _usage_history(series: list, range_key: str):
    """The CPU and memory history fragment both detail pages load: the latest
    and the peak over a line that spans its half, scaled with a floor."""
    from steward.core.status import sparkline_svg
    cpu = [c / 100 for _, c, _ in series]
    mem = [m for _, _, m in series]
    return await render_template(
        "docker/_usage_history.html", have_data=len(series) >= 2, range_key=range_key,
        cpu_now=cpu[-1] if cpu else None, cpu_peak=max(cpu) if cpu else None,
        mem_now=mem[-1] if mem else None, mem_peak=max(mem) if mem else None,
        cpu_floor=HISTORY_CPU_FLOOR, mem_floor=HISTORY_MEM_FLOOR,
        spark_cpu=sparkline_svg(cpu, height=48, stroke="currentColor", fill_width=True,
                                floor=HISTORY_CPU_FLOOR),
        spark_mem=sparkline_svg(mem, height=48, stroke="currentColor", fill_width=True,
                                floor=HISTORY_MEM_FLOOR),
    )


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


# ── A swarm service's own page (#5302) ───────────────────────────────────────

def _task_label(slot: str, task_id: str) -> str:
    """How a task reads beside its log lines: its slot and the start of its id."""
    return f"{slot} · {task_id[:10]}"


def _task_rows(tasks: list[dict], node_hostname: dict) -> list[dict]:
    """Swarm's records of a service's newest tasks, newest first, as the page
    lists them: where each ran, what state it reached, and why it ended."""
    from .swarm_view import FAILED_TASK_STATES, task_time
    out = []
    for t in tasks:
        nid = t.get("node_id") or ""
        out.append({
            "task_id": t.get("task_id") or "", "slot": str(t.get("slot") or ""), "node_id": nid,
            "node": node_hostname.get(nid) or (nid[:12] if nid else "—"),
            "state": t.get("state") or "unknown", "desired": t.get("desired_state") or "",
            "exit_code": t.get("exit_code"), "why": t.get("error") or t.get("message") or "",
            "at": task_time(t), "failed": t.get("state") in FAILED_TASK_STATES,
            "current": t.get("state") == "running" and t.get("desired_state") == "running",
        })
    return out


def _service_tasks(name: str, tasks: list[dict], replicas: list[dict],
                   node_host: dict, host_name: dict) -> list[dict]:
    """One table for a service's tasks (#5310): the running ones first, each
    with what its container uses, then Swarm's record of the rest, newest
    first. A node reads as its Steward host's name wherever one is known, so a
    node is spelled one way down the table. A running replica Swarm's record
    does not list (an agent before 1.8.0) still gets a row, and so does a node
    running tasks with no Steward agent on it."""
    live, unmatched = {}, []
    for r in (r for r in replicas if not r["ghost"]):
        slot, tid = docker_svc.task_of(name, r["name"]) or ("", "")
        tid = r.get("task_id") or tid          # the agent's own record of it, else its name's
        if tid:
            live[tid] = (slot, r)
        else:
            unmatched.append(r)

    def place(node_id, host_id, fallback):
        hid = host_id or node_host.get(node_id)
        return hid, (host_name.get(hid) if hid else None) or fallback

    rows = []
    for t in tasks:
        _, r = live.pop(t["task_id"], (None, None))
        hid, node = place(t["node_id"], r["host_id"] if r else None, t["node"])
        rows.append(dict(t, node=node, host_id=hid, container=r["name"] if r else None,
                         usage=r["usage"] if r else None, health=r["health"] if r else None,
                         at=(r["started_at"] or t["at"]) if r and t["current"] else t["at"]))
    for tid, (slot, r) in [*live.items(), *(("", ("", r)) for r in unmatched)]:
        rows.append({"task_id": tid, "slot": slot, "node_id": "", "node": r["host"],
                     "host_id": r["host_id"], "state": r["status"], "desired": "",
                     "exit_code": None, "why": "", "at": r["started_at"], "failed": False,
                     "current": (r["status"] or "").lower() == "running", "container": r["name"],
                     "usage": r["usage"], "health": r["health"]})
    listed = {t["node"] for t in tasks if t["current"]}
    for g in (r for r in replicas if r["ghost"] and r["host"] not in listed):
        rows.append({"task_id": "", "slot": "", "node_id": "", "node": g["host"], "host_id": None,
                     "state": "running", "desired": "", "exit_code": None,
                     "why": f"{g['count']} running on a node with no Steward agent: no figures",
                     "at": None, "failed": False, "current": True, "container": None,
                     "usage": None, "health": None})
    return sorted(rows, key=lambda r: not r["current"])     # stable: newest first within each


@docker_bp.get("/service/<name>")
@require_role(UserRole.viewer)
async def service_detail(name: str):
    """One swarm service: where its replicas run and what they use, its image,
    Swarm's record of its recent tasks, its CPU and memory over time across
    every replica, and the logs of every task it has run — newest first, each
    line marked with its task. `?task=` opens the logs on one task, which is
    where a failure's "Logs →" lands."""
    now = datetime.now(timezone.utc)
    async with current_app.db_sessionmaker() as db:
        ctx = await _swarm_context(db, now)
        task_containers = await docker_svc.service_containers(db, name)
        s = next((s for s in ctx["swarm_view"]["services"] if s["name"] == name), None)
        capacity = await docker_svc.host_capacity(
            db, {r["host_id"] for r in (s["replicas"] if s else []) if not r["ghost"]})
        norms = await docker_svc.container_norms(
            db, {(r["host_id"], r["name"]) for r in (s["replicas"] if s else []) if not r["ghost"]}, now)
    if s is None and not task_containers:
        return await render_template("docker/service.html", name=name, service=None), 404

    newest = max((v for v in ctx["services"] if v.service_name == name),
                 key=lambda v: v.updated_at, default=None)
    tasks = _task_rows(parse_tasks(newest) if newest is not None else [], ctx["node_hostname"])
    by_key = {(c.host_id, c.name): c for c in ctx["containers"]}
    replicas = []
    for r in (s["replicas"] if s else []):
        c = None if r["ghost"] else by_key.get((r["host_id"], r["name"]))
        replicas.append(dict(r, task_id=getattr(c, "task_id", None) if c is not None else None,
                             usage=docker_svc.usage(c, *capacity.get(r["host_id"], (None, None)),
                                                    thresholds=current_app.config.get("THRESHOLDS", {}),
                                                    norms=norms.get((c.host_id, c.name)))
                             if c is not None else None))
    node_host = {c.node_id: c.host_id for c in ctx["containers"] if c.node_id}
    table = _service_tasks(name, tasks, replicas, node_host, ctx["host_name"])
    # Every task Steward holds logs or events for, newest known first.
    known = {t["task_id"]: _task_label(t["slot"], t["task_id"]) for t in tasks if t["task_id"]}
    for _, cname in task_containers:
        slot, tid = docker_svc.task_of(name, cname)
        known.setdefault(tid, _task_label(slot, tid))
    focus = request.args.get("task", "")
    return await render_template(
        "docker/service.html", name=name, service=s,
        row=service_row(s, now) if s else None, svc=newest, tasks=table,
        task_choices=list(known.items()), focus_task=focus if focus in known else "",
        stack=stack_of(name), current_range=request.args.get("range", DEFAULT_RANGE),
    )


@docker_bp.get("/service/<name>/history")
@require_role(UserRole.viewer)
async def service_history_fragment(name: str):
    """HTMX fragment: the service's CPU (cores) and memory, summed across its
    replicas, over the selected range, with the latest and the peak."""
    since, range_key = parse_range(request.args.get("range"))
    async with current_app.db_sessionmaker() as db:
        series = await docker_svc.service_history(
            db, await docker_svc.service_containers(db, name), since)
    return await _usage_history(series, range_key)


@docker_bp.get("/service/<name>/logs/lines")
@require_role(UserRole.viewer)
async def service_logs_lines(name: str):
    """HTMX fragment: the newest retained lines across every task of the
    service, stream, text and task filtered, each marked with its task."""
    stream = request.args.get("stream", "all")
    query = (request.args.get("q") or "").strip()
    task = request.args.get("task") or None
    async with current_app.db_sessionmaker() as db:
        rows = await docker_svc.query_service_logs(
            db, await docker_svc.service_containers(db, name), stream, query, _LOG_TAIL_DEFAULT,
            task=task)
    lines = []
    for r in rows:
        slot, tid = docker_svc.task_of(name, r.container_name) or ("?", r.container_name)
        lines.append({"ts": r.ts, "stream": r.stream, "line": r.line, "task": _task_label(slot, tid)})
    return await render_template(
        "docker/_container_logs_lines.html", lines=lines, name=name, tail=_LOG_TAIL_DEFAULT,
        filtered=bool(query) or bool(task) or stream in ("stdout", "stderr"), service=True,
    )