feat(docker): Docker's own status reaches Steward — an empty answer clears a host, a lost one greys it, a demoted manager's swarm rows go (#5152)

The agent (1.9.0) said nothing when Docker could not be asked and nothing
when Docker had no containers, so Steward could tell neither apart from the
other, and kept a host's last containers on screen as running either way.

- Agent: collect_docker returns (status, detail, containers). "ok" sends
  docker: [] when there are none; "unavailable" (no socket, nothing
  listening) and "error" (permission, timeout, bad answer) say why.
  collect_swarm says whether /info answered, and a node that is not a
  manager sends swarm_manager: false. A failed /info says nothing.
- Ingest: an empty docker list is a snapshot; the newest sample's status
  goes to the Docker plugin with the rest.
- docker_host_status (docker_011): per host, status, detail, since and
  last report. A host that never had Docker gets no row.
- An "ok" empty snapshot reaps the host's containers; an error keeps them.
  swarm_manager: false drops the host's swarm services and nodes.
- Docker page: a "Docker not answering" card names each such host, why,
  and since when; its containers read as stale. Duplicate header comment
  in rows.html removed.

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

tests/integration/test_host_agent_docker_status.py

@@ -0,0 +1,79 @@
+"""Integration: an agent's Docker status reaches Steward through ingest (#5152).
+
+The newest sample's status wins over a backlog's older ones; "error" keeps
+the containers Docker last listed; an "ok" with an empty list clears them.
+
+Rows are read with plain SQL: importing the plugin's models again after the
+app loaded the plugin raises "Table already defined" (see test_mcp_plugin_tools)."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def app():
+    if not os.environ.get("STEWARD_DATABASE_URL"):
+        pytest.skip("needs Postgres")
+    from steward.app import create_app
+    return create_app(testing=False)
+
+
+def test_ingest_records_docker_status_and_clears_on_an_empty_answer(app):
+    from sqlalchemy import text
+    from steward.models.hosts import Host
+
+    web = {"name": "web", "status": "running", "cpu_pct": 1.0, "mem_pct": 1.0,
+           "restart_count": 0, "exit_code": None, "oom_killed": False, "health": None}
+
+    async def _go():
+        tag = uuid.uuid4().hex[:8]
+        token = f"tok-{tag}-{uuid.uuid4().hex}"
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                host = Host(name=f"dock-{tag}", address="10.9.9.3")
+                s.add(host)
+                await s.flush()
+                await s.execute(text(
+                    "INSERT INTO host_agent_registrations "
+                    "(id, host_id, token_hash, token_created_at, created_at, updated_at) "
+                    "VALUES (:id, :host, :hash, now(), now(), now())"),
+                    {"id": str(uuid.uuid4()), "host": host.id,
+                     "hash": hashlib.sha256(token.encode()).hexdigest()})
+            host_id = host.id
+
+        client = app.test_client()
+        now = datetime.now(timezone.utc)
+
+        async def post(*samples):
+            resp = await client.post("/plugins/host_agent/ingest", json={"samples": list(samples)},
+                                     headers={"Authorization": f"Bearer {token}"})
+            assert resp.status_code == 200, await resp.get_data(as_text=True)
+            async with app.db_sessionmaker() as s:
+                status = (await s.execute(text(
+                    "SELECT status, detail FROM docker_host_status WHERE host_id = :h"),
+                    {"h": host_id})).first()
+                names = [r[0] for r in (await s.execute(text(
+                    "SELECT name FROM docker_containers WHERE host_id = :h"), {"h": host_id})).all()]
+            return (tuple(status) if status else None), names
+
+        def at(seconds_ago):
+            return (now - timedelta(seconds=seconds_ago)).isoformat()
+
+        # A backlog, newest first: Docker listed web, then stopped answering.
+        backlog = await post(
+            {"ts": at(60), "docker_status": "error", "docker_detail": "permission denied on /s"},
+            {"ts": at(90), "docker_status": "ok", "docker": [web]})
+        emptied = await post({"ts": at(30), "docker_status": "ok", "docker": []})
+        return backlog, emptied
+
+    backlog, emptied = asyncio.run(_go())
+    assert backlog == (("error", "permission denied on /s"), ["web"])
+    assert emptied == (("ok", None), [])