feat(docker): ingest swarm topology + lifecycle events + health/restart alerts

Wires the agent's enriched + swarm payloads through the docker.persist_host_
samples capability:

  * Swarm topology — persist sample["swarm"] into docker_swarm_services /
    docker_swarm_nodes (upsert + prune stale, host-scoped so two managers don't
    clobber). Migration docker_005 adds services.placement_json for the
    task→node placement the agent now reports.
  * Lifecycle events — _derive_events (pure, unit-tested) diffs the newest
    snapshot against stored per-container state: start / stop / die (non-zero
    exit) / oom / health_change → docker_events rows. Skipped on a host's first
    snapshot so the baseline doesn't emit a start per existing container.
  * Alerts — record restart_count (always) and is_healthy (1.0/0.0, only when a
    HEALTHCHECK exists) alongside cpu/mem, under host-scoped resource names;
    METRIC_CATALOG[docker] gains restart_count + is_healthy so they're alertable.

host_agent ingest captures the newest sample's swarm object and threads it to
the capability (now persist_host_docker(session, host, snapshots, swarm=None));
invoked when containers OR swarm are present, under the same SAVEPOINT. Unit
tests cover the event-diff matrix; integration tests cover event derivation
across two snapshots and swarm topology round-trip (incl. placement).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC

plugins/docker/ingest.py

@@ -2,16 +2,23 @@
 """Persist host-scoped Docker samples pushed by the host agent.
 
 Published as the "docker.persist_host_samples" capability (see __init__.setup),
-so the host_agent plugin can hand off the `docker` array from a sample WITHOUT
-importing the docker models — the coupling is opportunistic and degrades to a
-no-op when the docker plugin is disabled. Runs inside the caller's open
-transaction (the ingest handler's session); never opens or commits its own.
+so the host_agent plugin can hand off a sample's `docker` array (and, on a swarm
+manager, its `swarm` object) WITHOUT importing the docker models — the coupling
+is opportunistic and degrades to a no-op when the docker plugin is disabled.
+Runs inside the caller's open transaction (the ingest handler's SAVEPOINT);
+never opens or commits its own.
 """
 from __future__ import annotations
 
 import json
 from datetime import datetime
 
+from sqlalchemy import delete, select
+
+
+# Stopped/terminal container states (anything not "running"/"paused"/"restarting").
+_STOPPED_STATES = {"exited", "dead", "stopped"}
+
 
 def _parse_started_at(value) -> datetime | None:
     """Parse the agent's ISO-8601 started_at string back to a datetime."""
@@ -23,18 +30,142 @@ def _parse_started_at(value) -> datetime | None:
         return None
 
 
-async def persist_host_docker(session, host, snapshots) -> None:
-    """Upsert containers + append time-series for one host's docker snapshots.
+def _derive_events(old_state: dict, new_containers: list) -> list:
+    """Diff a fresh snapshot against stored per-container state → lifecycle events.
+
+    Pure function (no DB) so it's unit-testable. `old_state` maps name → the
+    previously stored {status, health, oom_killed, exit_code}; `new_containers`
+    is the newest snapshot's list of container dicts. Returns a list of
+    (container_name, event, detail) tuples:
+
+      start         — a container transitions into running (or a genuinely new
+                      container appears already running)
+      stop          — running → not-running, or a running container is removed
+      die           — same as stop but with a non-zero exit code (abnormal)
+      oom           — OOMKilled flips False→True
+      health_change — HEALTHCHECK status string changes
+
+    The CALLER skips this entirely on a host's first-ever snapshot (empty
+    old_state) so the baseline doesn't emit a spurious "start" per existing
+    container — a later-appearing container still gets one because by then
+    old_state is populated and that container's prior entry is simply absent.
+    """
+    events: list = []
+    new_by_name = {c["name"]: c for c in new_containers if c.get("name")}
+
+    for name, c in new_by_name.items():
+        new_status = (c.get("status") or "").lower()
+        new_running = new_status == "running"
+        new_oom = bool(c.get("oom_killed"))
+        new_health = c.get("health")
+        new_exit = c.get("exit_code")
+        old = old_state.get(name)
+
+        if old is None:
+            # Newly observed container — only a start is meaningful (we have no
+            # prior state to diff a death against).
+            if new_running:
+                events.append((name, "start", c.get("image") or None))
+            continue
+
+        old_running = (old.get("status") or "").lower() == "running"
+        if new_running and not old_running:
+            events.append((name, "start", c.get("image") or None))
+        elif old_running and not new_running:
+            if new_exit not in (None, 0):
+                events.append((name, "die", f"exit {new_exit}"))
+            else:
+                events.append((name, "stop", None))
+
+        if new_oom and not bool(old.get("oom_killed")):
+            events.append((name, "oom", None))
+
+        if new_health != old.get("health") and (new_health or old.get("health")):
+            events.append((name, "health_change",
+                           f"{old.get('health') or '?'}→{new_health or '?'}"))
+
+    # A running container that vanished from the listing entirely (removed).
+    for name, old in old_state.items():
+        if name not in new_by_name and (old.get("status") or "").lower() == "running":
+            events.append((name, "stop", "removed"))
+
+    return events
+
+
+async def _persist_swarm(session, host, swarm: dict) -> None:
+    """Upsert this manager's swarm topology; drop rows no longer reported.
+
+    Current-state tables (not time-series): a manager re-reports its full
+    services/nodes set every sample, so we upsert what's present and delete what
+    isn't (scoped to this host so two managers don't clobber each other).
+    """
+    from datetime import timezone
+    from .models import DockerSwarmNode, DockerSwarmService
+
+    now = datetime.now(timezone.utc)
+
+    services = swarm.get("services") or []
+    seen_services: set[str] = set()
+    for s in services:
+        name = s.get("service_name")
+        if not name:
+            continue
+        seen_services.add(name)
+        row = await session.get(DockerSwarmService, (host.id, name))
+        if row is None:
+            row = DockerSwarmService(host_id=host.id, service_name=name)
+            session.add(row)
+        row.mode = s.get("mode") or "replicated"
+        row.desired = int(s.get("desired") or 0)
+        row.running = int(s.get("running") or 0)
+        row.image = s.get("image")
+        row.placement_json = json.dumps(s.get("placement") or [])
+        row.updated_at = now
+    stale_services = delete(DockerSwarmService).where(
+        DockerSwarmService.host_id == host.id)
+    if seen_services:
+        stale_services = stale_services.where(
+            DockerSwarmService.service_name.notin_(seen_services))
+    await session.execute(stale_services)
+
+    nodes = swarm.get("nodes") or []
+    seen_nodes: set[str] = set()
+    for n in nodes:
+        nid = n.get("node_id")
+        if not nid:
+            continue
+        seen_nodes.add(nid)
+        row = await session.get(DockerSwarmNode, (host.id, nid))
+        if row is None:
+            row = DockerSwarmNode(host_id=host.id, node_id=nid)
+            session.add(row)
+        row.hostname = n.get("hostname") or ""
+        row.role = n.get("role") or "worker"
+        row.availability = n.get("availability") or "active"
+        row.status = n.get("status") or "unknown"
+        row.leader = bool(n.get("leader", False))
+        row.updated_at = now
+    stale_nodes = delete(DockerSwarmNode).where(DockerSwarmNode.host_id == host.id)
+    if seen_nodes:
+        stale_nodes = stale_nodes.where(DockerSwarmNode.node_id.notin_(seen_nodes))
+    await session.execute(stale_nodes)
+
+
+async def persist_host_docker(session, host, snapshots, swarm=None) -> None:
+    """Upsert containers + time-series + lifecycle events + swarm for one host.
 
     `snapshots` is a list of (recorded_at: datetime, containers: list[dict]) —
-    one entry per ingested sample carrying docker data (usually just one; more
-    when the agent flushes a backlog). Every snapshot contributes time-series
-    DockerMetric rows; the newest snapshot drives current container state and
-    the alert pipeline (record_metric writes "now", so feeding it stale buffered
-    snapshots would be misleading).
+    one entry per ingested sample carrying docker data (usually one; more when
+    the agent flushes a backlog). Every snapshot contributes time-series
+    DockerMetric rows; the newest snapshot drives current container state, the
+    alert pipeline, and lifecycle-event derivation. `swarm` is the newest
+    sample's swarm object (or None off managers) — persisted when present.
     """
     from steward.core.alerts import record_metric
-    from .models import DockerContainer, DockerMetric
+    from .models import DockerContainer, DockerEvent, DockerMetric
+
+    if swarm is not None:
+        await _persist_swarm(session, host, swarm)
 
     if not snapshots:
         return
@@ -54,6 +185,24 @@ async def persist_host_docker(session, host, snapshots) -> None:
                     mem_usage_bytes=c.get("mem_usage_bytes") or 0,
                 ))
 
+    # Snapshot of stored per-container state BEFORE the upsert overwrites it —
+    # used to diff lifecycle events. Skip event derivation on the host's first
+    # snapshot (empty state) so we don't emit a start per pre-existing container.
+    old_rows = (await session.execute(
+        select(DockerContainer).where(DockerContainer.host_id == host.id)
+    )).scalars().all()
+    old_state = {
+        r.name: {"status": r.status, "health": r.health,
+                 "oom_killed": r.oom_killed, "exit_code": r.exit_code}
+        for r in old_rows
+    }
+    if old_state:
+        for name, event, detail in _derive_events(old_state, latest_containers):
+            session.add(DockerEvent(
+                host_id=host.id, container_name=name,
+                event=event, detail=detail, at=latest_at,
+            ))
+
     # Current state + alerts from the newest snapshot only.
     for c in latest_containers:
         name = c.get("name")
@@ -90,8 +239,8 @@ async def persist_host_docker(session, host, snapshots) -> None:
 
         # Alert pipeline — resource is host-scoped so containers of the same name
         # on different hosts don't collide in the metric/alert namespace.
+        resource = f"{host.name}/{name}"
         if c.get("status") == "running" and c.get("cpu_pct") is not None:
-            resource = f"{host.name}/{name}"
             await record_metric(
                 session=session, source_module="docker",
                 resource_name=resource, metric_name="cpu_pct", value=c["cpu_pct"],
@@ -101,3 +250,20 @@ async def persist_host_docker(session, host, snapshots) -> None:
                     session=session, source_module="docker",
                     resource_name=resource, metric_name="mem_pct", value=c["mem_pct"],
                 )
+        # Restart count is alertable regardless of state (crash-looping matters
+        # most while the container is down/restarting).
+        await record_metric(
+            session=session, source_module="docker",
+            resource_name=resource, metric_name="restart_count",
+            value=float(c.get("restart_count", 0) or 0),
+        )
+        # Health → 1.0/0.0 only for containers that actually define a HEALTHCHECK
+        # (health is None otherwise — recording 0 would false-alarm every plain
+        # container).
+        health = c.get("health")
+        if health in ("healthy", "unhealthy", "starting"):
+            await record_metric(
+                session=session, source_module="docker",
+                resource_name=resource, metric_name="is_healthy",
+                value=1.0 if health == "healthy" else 0.0,
+            )