feat(docker): Swarm task awareness — replaced tasks, service crash loops, last task error

Milestone 465 "Swarm task awareness". Fixes three defects from the live MCP
review #4973:
- #5008 "Swarm task history read as live state"
- #5009 "Crash-loop detection is blind to Swarm services"
- #5011 "Swarm task errors are not captured"

Agent and ingest (agent 1.8.0):
- A manager's agent now reports each service's newest tasks: slot, node,
  desired and current state, Swarm's Err/Message, exit code and time.
- They are stored in docker_swarm_services.tasks_json (migration
  docker_010_swarm_tasks).

Swarm view (pure):
- A task container is "replaced" when Swarm set its task's desired state off
  `running`. Without the task on record, a newer or running task in the same
  slot decides.
- Replaced containers are no longer replicas or crashes.
- Each service gets, from the task records:
  - failures_1h and failures_24h (failed or rejected tasks);
  - crash_looping (Settings → Thresholds count);
  - last_error, in Swarm's own words.

MCP:
- New findings: swarm_service_crash_looping (critical) and swarm_task_failed
  (warning, on a healthy service with a failure in 24h).
- Down and degraded findings name the last task error.
- list_swarm_services carries the failure fields.
- Container rows mark swarm_replaced and give a replaced task no problems.

Web:
- The containers page swarm cards and the Swarm topology page show crash
  loops, recent failures and the last task error.
- The dashboard widget flags a crash loop, and its "failed (24h)" count no
  longer counts a rolling update's killed task.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/docker/swarm_view.py

@@ -11,12 +11,28 @@ For every swarm service we list the real replicas we have detail for, then add
 "ghost" replicas for the remaining placement count on each node (agent-less
 nodes). Non-swarm containers pass through grouped by host, unchanged.
 
+A third source, DockerSwarmService.tasks_json (agent ≥ 1.8.0), is Swarm's own
+record of each service's newest tasks. It decides which task containers are
+*superseded* — replaced by Swarm, so neither a replica nor a crash — and gives
+each service its failures across task retries and the error of its last
+failed task. Swarm restarts a task by creating a new one (a new container), so
+failures can only be counted per service, never per container.
+
 Kept model-free (no ORM imports) so it's unit-testable and safe to import in
 tests without the plugin-loader "table already defined" gotcha.
 """
 from __future__ import annotations
 
 import json
+from datetime import datetime, timedelta, timezone
+
+# Task states that mean the task ended in error: `failed` (its container exited
+# non-zero or never started) and `rejected` (Swarm could not start it at all,
+# e.g. no such image, a bad mount).
+FAILED_TASK_STATES = ("failed", "rejected")
+# Desired states of a task Swarm still means to run. Anything else (shutdown,
+# remove, complete, orphaned) is a task Swarm has finished with.
+_CURRENT_DESIRED = ("running", "ready", "")
 
 
 def _service_state(running: int, desired: int) -> str:
@@ -36,30 +52,179 @@ def _placement(service) -> list[dict]:
     return [p for p in data if isinstance(p, dict)]
 
 
-def build_swarm_services(containers, services, node_hostname: dict, host_name: dict) -> dict:
-    """Return {"services": [...], "standalone": [...]}.
+def parse_tasks(service) -> list[dict]:
+    """A service row's task records, newest first; [] when its agent sent none."""
+    raw = getattr(service, "tasks_json", None) or "[]"
+    try:
+        data = json.loads(raw)
+    except (ValueError, TypeError):
+        return []
+    return [t for t in data if isinstance(t, dict)]
 
-    services[i] = {name, mode, desired, running, state, replicas:[...]}
-      real replica  = {ghost:False, host, host_id, name, status, cpu_pct, mem_pct,
-                       health, restart_count}
-      ghost replica = {ghost:True, host, count}   # placement with no local row
-    standalone[i] = {host_id, host, containers:[...]}  # non-swarm, by host
+
+def task_time(task: dict) -> datetime | None:
+    """A task's status timestamp. Docker writes nanoseconds; Python ≥ 3.11 keeps
+    the first six digits."""
+    raw = task.get("at") or ""
+    try:
+        at = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return at if at.tzinfo else at.replace(tzinfo=timezone.utc)
+
+
+def task_failures(tasks: list[dict], since: datetime) -> int:
+    """How many tasks ended in error at or after `since`."""
+    n = 0
+    for t in tasks:
+        at = task_time(t)
+        if t.get("state") in FAILED_TASK_STATES and at is not None and at >= since:
+            n += 1
+    return n
+
+
+def last_task_error(tasks: list[dict]) -> dict | None:
+    """Why the service's newest troubled task is in trouble, from Swarm itself.
+
+    The newest task that failed or was rejected, or that Swarm attached an error
+    to while still trying (a pending task with "no suitable node", say). None
+    when no retained task has a problem.
     """
+    for t in tasks:  # newest first
+        if t.get("state") in FAILED_TASK_STATES or t.get("error"):
+            return {"state": t.get("state"), "error": t.get("error") or None,
+                    "message": t.get("message") or None, "exit_code": t.get("exit_code"),
+                    "node_id": t.get("node_id") or None, "at": t.get("at") or None}
+    return None
+
+
+def task_summary(tasks: list[dict], now: datetime, crash_loop_deaths: int,
+                 node_hostname: dict) -> dict:
+    """What a service's task records say: tasks_known, failures_1h,
+    failures_24h (None without records), crash_looping, and last_error with
+    the node's hostname as `node`."""
+    known = bool(tasks)
+    failures_1h = task_failures(tasks, now - timedelta(hours=1)) if known else None
+    last_error = last_task_error(tasks)
+    if last_error is not None:
+        nid = last_error["node_id"]
+        last_error["node"] = node_hostname.get(nid) or (nid[:12] if nid else None)
+        at = task_time(last_error)
+        last_error["age_s"] = max(0, int((now - at).total_seconds())) if at else None
+    return {
+        "tasks_known": known,
+        "failures_1h": failures_1h,
+        "failures_24h": task_failures(tasks, now - timedelta(hours=24)) if known else None,
+        "crash_looping": bool(known and failures_1h >= crash_loop_deaths),
+        "last_error": last_error,
+    }
+
+
+def _slot_key(c) -> str:
+    """A task container's slot within its service: "<service>.<slot>.<task>" for
+    replicated services, "<service>.<node id>.<task>" for global ones."""
+    svc, name = c.service_name, c.name or ""
+    rest = name[len(svc) + 1:] if name.startswith(svc + ".") else name
+    return rest.split(".", 1)[0]
+
+
+def superseded_containers(swarm_cs, tasks_by_id: dict[str, dict]) -> set[tuple]:
+    """(host_id, name) of task containers Swarm has replaced.
+
+    With the task on record, its desired state decides: Swarm sets a replaced
+    or failed task's desired state to shutdown. Without it (an older agent, or a
+    task older than the retained history), a container that is not running is
+    superseded when another task in the same slot is running or started later.
+    """
+    out: set[tuple] = set()
+    by_slot: dict[tuple, list] = {}
+    for c in swarm_cs:
+        by_slot.setdefault((c.service_name, _slot_key(c)), []).append(c)
+    for c in swarm_cs:
+        task = tasks_by_id.get(getattr(c, "task_id", None) or "")
+        if task is not None:
+            if task.get("desired_state") not in _CURRENT_DESIRED:
+                out.add((c.host_id, c.name))
+            continue
+        if (getattr(c, "status", "") or "").lower() == "running":
+            continue
+        mine = getattr(c, "started_at", None)
+        for other in by_slot[(c.service_name, _slot_key(c))]:
+            if other is c:
+                continue
+            theirs = getattr(other, "started_at", None)
+            if ((getattr(other, "status", "") or "").lower() == "running"
+                    or (theirs is not None and (mine is None or theirs > mine))):
+                out.add((c.host_id, c.name))
+                break
+    return out
+
+
+def _newest_per_name(services) -> dict[str, object]:
+    """Every manager reports the same cluster-global service set; keep the most
+    recently updated row per name, so a manager on an older agent (no tasks)
+    doesn't shadow one that reports them."""
+    by_name: dict[str, object] = {}
+    for s in services:
+        name = getattr(s, "service_name", None)
+        if not name:
+            continue
+        kept = by_name.get(name)
+        if kept is None:
+            by_name[name] = s
+            continue
+        a, b = getattr(s, "updated_at", None), getattr(kept, "updated_at", None)
+        if a is not None and (b is None or a > b):
+            by_name[name] = s
+    return by_name
+
+
+def replaced_tasks(containers, services) -> set[tuple]:
+    """(host_id, name) of the task containers Swarm has replaced, from the
+    container rows and the managers' service rows. What the container views use
+    to leave history out of their problem lists."""
+    swarm_cs = [c for c in containers if getattr(c, "service_name", None)]
+    tasks_by_id = {t["task_id"]: t for s in _newest_per_name(services).values()
+                   for t in parse_tasks(s) if t.get("task_id")}
+    return superseded_containers(swarm_cs, tasks_by_id)
+
+
+def build_swarm_services(containers, services, node_hostname: dict, host_name: dict,
+                         *, now: datetime | None = None, crash_loop_deaths: int = 3) -> dict:
+    """Return {"services": [...], "standalone": [...], "superseded": {(host_id, name)}}.
+
+    services[i] = {name, mode, desired, running, state, replicas:[...],
+                   previous_tasks, tasks_known, failures_1h, failures_24h,
+                   crash_looping, last_error}
+      real replica  = {ghost:False, host, host_id, name, status, cpu_pct, mem_pct,
+                       health, restart_count}   # current tasks only
+      ghost replica = {ghost:True, host, count}   # placement with no local row
+      previous_tasks = task containers Swarm replaced, left out of replicas
+      failures_*    = tasks that failed or were rejected in the window (None
+                      without task data); crash_looping = failures_1h ≥ the
+                      Settings → Thresholds crash-loop count
+      last_error    = last_task_error(), plus `node` (the node's hostname)
+    standalone[i] = {host_id, host, containers:[...]}  # non-swarm, by host
+    superseded    = the task containers Swarm replaced, so callers can skip them
+    """
+    now = now or datetime.now(timezone.utc)
     containers = list(containers)
     swarm_cs = [c for c in containers if getattr(c, "service_name", None)]
     standalone_cs = [c for c in containers if not getattr(c, "service_name", None)]
 
-    # Dedup service rows by name (every manager reports the same cluster-global
-    # set); keep the first — placement is identical across managers.
-    svc_by_name: dict[str, object] = {}
-    for s in services:
-        name = getattr(s, "service_name", None)
-        if name and name not in svc_by_name:
-            svc_by_name[name] = s
+    svc_by_name = _newest_per_name(services)
+    tasks_by_svc = {name: parse_tasks(s) for name, s in svc_by_name.items()}
+    tasks_by_id = {t["task_id"]: t for ts in tasks_by_svc.values() for t in ts
+                   if t.get("task_id")}
+    superseded = superseded_containers(swarm_cs, tasks_by_id)
 
-    # Real containers grouped by service name.
+    # Current task containers grouped by service name; replaced ones are counted.
     cs_by_service: dict[str, list] = {}
+    previous: dict[str, int] = {}
     for c in swarm_cs:
+        if (c.host_id, c.name) in superseded:
+            previous[c.service_name] = previous.get(c.service_name, 0) + 1
+            continue
         cs_by_service.setdefault(c.service_name, []).append(c)
 
     def _host_of(c):
@@ -124,6 +289,8 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d
             "desired": desired,
             "state": _service_state(int(running or 0), int(desired or 0)),
             "replicas": replicas,
+            "previous_tasks": previous.get(name, 0),
+            **task_summary(tasks_by_svc.get(name, []), now, crash_loop_deaths, node_hostname),
         })
 
     # Non-swarm containers grouped by host, running-first order preserved.
@@ -137,4 +304,4 @@ def build_swarm_services(containers, services, node_hostname: dict, host_name: d
         g["containers"].append(c)
     standalone = sorted(groups.values(), key=lambda g: g["host"].lower())
 
-    return {"services": out_services, "standalone": standalone}
+    return {"services": out_services, "standalone": standalone, "superseded": superseded}