feat(mcp): swarm nodes link to their host and maintenance; the overview names a node that is down (#5835)

#5835 "MCP: swarm nodes", step 2 of milestone 520. list_swarm_services
already listed every node (role, leader, availability, status); the
audit missed it. What was missing:
- Each node row carries `host_id` (the Steward host of that name) and
  `maintenance` (the state of a maintenance job running on it), so an
  agent can go node -> get_host / get_maintenance_job.
- get_overview names a node whose status isn't ready, unless its host
  has a maintenance job running (rebooting on purpose).
- Drained alone is not a finding: VDNT-Util01 is a manager kept drained
  on purpose (#5777), and a drain a failed job left behind is already
  that job's maintenance_failed finding.
unique_nodes() keeps the newest report of each node, as the list did.

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

plugins/docker/mcp_tools.py

@@ -200,12 +200,63 @@ async def _swarm(db, now: datetime):
     return built["services"], nodes, host_names
 
 
+def unique_nodes(nodes: list) -> list:
+    """Each swarm node once: every manager reports them all, so keep the
+    newest report of each."""
+    seen, out = set(), []
+    for n in sorted(nodes, key=lambda n: n.updated_at, reverse=True):
+        if n.node_id not in seen:
+            seen.add(n.node_id)
+            out.append(n)
+    return out
+
+
+async def _maintaining(db) -> dict[str, str]:
+    """{host name, lower case: job state} for hosts with a maintenance job
+    running (#5835): a node drained or rebooting for one is expected. The
+    host agent is always loaded; None-safe all the same."""
+    from steward.core.plugin_manager import loaded_plugin_submodule
+    models = loaded_plugin_submodule("host_agent", "models")
+    maint = loaded_plugin_submodule("host_agent", "maintenance")
+    if models is None or maint is None:
+        return {}
+    job = models.HostMaintenanceJob
+    rows = await db.execute(select(Host.name, job.state).join(Host, Host.id == job.host_id)
+                            .where(job.state.in_(maint.ACTIVE)))
+    return {name.lower(): state for name, state in rows}
+
+
+NODE_AVAILABILITY_SAYS = {"drain": "drained", "pause": "paused"}
+
+
+def node_findings(nodes: list, maintaining: dict[str, str]) -> list[dict]:
+    """Swarm nodes down now (#5835): status not ready. A node whose host has
+    a maintenance job running is rebooting on purpose and isn't named.
+
+    Drained alone is not a finding: the operator keeps a manager drained on
+    purpose (#5777), and a drain a failed maintenance job left behind is that
+    job's finding."""
+    out = []
+    for n in unique_nodes(nodes):
+        name = n.hostname or n.node_id
+        if n.status == "ready" or name.lower() in maintaining:
+            continue
+        drained = NODE_AVAILABILITY_SAYS.get(n.availability)
+        out.append(finding(
+            "critical", "swarm_node_down", name,
+            f"swarm {n.role} {name} is {n.status}" + (f" and {drained}" if drained else ""),
+            ids={"node_id": n.node_id}, next="list_swarm_services()"))
+    return out
+
+
 async def mcp_overview(db) -> list[dict]:
-    """get_overview findings: failing containers and under-replicated swarm services."""
+    """get_overview findings: failing containers, under-replicated swarm
+    services, and swarm nodes that are down."""
     now = now_utc()
     rows = await _rows(db, now)
-    services, _, _ = await _swarm(db, now)
-    return container_findings(rows) + service_findings(services)
+    services, nodes, _ = await _swarm(db, now)
+    return (container_findings(rows) + service_findings(services)
+            + node_findings(nodes, await _maintaining(db)))
 
 
 # ── tools ────────────────────────────────────────────────────────────────────
@@ -438,21 +489,25 @@ async def list_swarm_services(limit: int = 100) -> dict:
       Swarm keeps few old tasks, so a long outage reads as a lower bound.
 
     Services with a problem come first. `nodes` lists every swarm node once,
-    with its role (manager/worker), availability and status.
+    with its role (manager/worker, `leader`), availability (active, drain,
+    pause) and status (ready, down, unknown); `host_id` is the Steward host
+    of that name (get_host, get_maintenance_job), and `maintenance` the state
+    of a maintenance job running on it, which drains and reboots it on
+    purpose (null when none).
     limit: default 100, max 500.
     """
     limit = clamp_limit(limit, 100, 500)
     now = now_utc()
     async with db_session() as db:
         services, nodes, host_names = await _swarm(db, now)
-    seen, node_rows = set(), []
-    for n in sorted(nodes, key=lambda n: n.updated_at, reverse=True):
-        if n.node_id in seen:
-            continue
-        seen.add(n.node_id)
-        node_rows.append({"node_id": n.node_id, "hostname": n.hostname, "role": n.role,
-                          "leader": n.leader, "availability": n.availability, "status": n.status,
-                          "reported_by": host_names.get(n.host_id)})
+        maintaining = await _maintaining(db)
+    host_ids = {name.lower(): hid for hid, name in host_names.items()}
+    node_rows = [{"node_id": n.node_id, "hostname": n.hostname, "role": n.role,
+                  "leader": n.leader, "availability": n.availability, "status": n.status,
+                  "host_id": host_ids.get((n.hostname or "").lower()),
+                  "maintenance": maintaining.get((n.hostname or "").lower()),
+                  "reported_by": host_names.get(n.host_id)}
+                 for n in unique_nodes(nodes)]
     node_rows.sort(key=lambda n: n["hostname"].lower())
     order = {"down": 0, "degraded": 1, "healthy": 2}
 

tests/plugins/docker/test_mcp_tools.py

@@ -196,3 +196,30 @@ def test_services_that_failed_together_are_one_finding_and_recovered_ones_none()
     assert "auth (1 of 2 replicas running), wiki (0 of 1 replicas running)" in together["summary"]
     assert together["since"] == "2026-10-06T11:30:00Z"
     assert alone["subject"] == "later" and alone["kind"] == "swarm_service_degraded"
+
+
+def _node(node_id, hostname, status="ready", availability="active", role="worker", minutes=0):
+    return NS(node_id=node_id, hostname=hostname, status=status, availability=availability,
+              role=role, leader=False, host_id="mgr",
+              updated_at=datetime(2026, 10, 15, 12, 0, tzinfo=timezone.utc) - timedelta(minutes=minutes))
+
+
+def test_unique_nodes_keep_the_newest_report_of_each():
+    old, new = _node("n1", "a", status="down", minutes=5), _node("n1", "a")
+    assert t.unique_nodes([old, new, _node("n2", "b")]) == [new, _node("n2", "b")]
+
+
+def test_node_findings_name_a_down_node_unless_maintenance_is_rebooting_it():
+    nodes = [_node("n1", "Web-1", status="down", role="manager"),
+             _node("n2", "web-2", availability="drain"),       # kept drained on purpose (#5777)
+             _node("n3", "web-3", status="unknown", availability="pause"),
+             _node("n4", "web-4", availability="drain", status="down"),   # being rebooted
+             _node("n5", "web-5")]
+    out = {f["subject"]: f for f in t.node_findings(nodes, {"web-4": "rebooting"})}
+    assert set(out) == {"Web-1", "web-3"}
+    assert out["Web-1"]["severity"] == "critical" and out["Web-1"]["kind"] == "swarm_node_down"
+    assert out["Web-1"]["summary"] == "swarm manager Web-1 is down"
+    assert out["web-3"]["summary"] == "swarm worker web-3 is unknown and paused"
+    # Once its job is over, the same node drained and down is named.
+    [f] = t.node_findings([_node("n4", "web-4", availability="drain", status="down")], {})
+    assert f["summary"] == "swarm worker web-4 is down and drained"