feat(docker): collect + persist /system/df image/disk usage (agent 1.6.0)

Backend for the image/disk panel (milestone 77 #942). Agent gains
collect_disk_usage() — one /system/df call (gated on containers existing, so
Docker-less hosts pay nothing), surfacing reclaimable bytes (image size held by
unreferenced images), layers/containers/volumes/build-cache sizes, and the top
50 images by size. Emitted as sample["docker_disk"]; host_agent ingest tracks
the newest sample's copy and hands it to the docker capability as a 5th arg.

New current-state tables docker_disk_usage (one row/host) + docker_images
(per-host image rows), docker_007 migration; ingest upserts the summary and
replaces the image set per host (stale images pruned). Unit tests for the df
parsing/reclaimable math + build_sample gating; integration test for
persistence + image-set replacement.

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

plugins/host_agent/routes.py

@@ -199,6 +199,8 @@ async def ingest():
         docker_snapshots: list[tuple[datetime, list]] = []
         latest_swarm: dict | None = None
         latest_swarm_ts: datetime | None = None
+        latest_disk: dict | None = None
+        latest_disk_ts: datetime | None = None
         for sample in samples:
             try:
                 recorded_at = _parse_ts(sample["ts"])
@@ -216,6 +218,11 @@ async def ingest():
             if isinstance(swarm, dict) and (
                     latest_swarm_ts is None or recorded_at > latest_swarm_ts):
                 latest_swarm, latest_swarm_ts = swarm, recorded_at
+            # Disk usage is current-state too — keep only the newest sample's.
+            disk = sample.get("docker_disk")
+            if isinstance(disk, dict) and (
+                    latest_disk_ts is None or recorded_at > latest_disk_ts):
+                latest_disk, latest_disk_ts = disk, recorded_at
             accepted += 1
             if latest_ts is None or recorded_at > latest_ts:
                 latest_ts = recorded_at
@@ -228,7 +235,7 @@ async def ingest():
         # (opportunistic synergy via the capability registry — no hard import,
         # no-op when docker is disabled). A failure here must never sink the
         # whole ingest, so the metrics above still land.
-        if docker_snapshots or latest_swarm is not None:
+        if docker_snapshots or latest_swarm is not None or latest_disk is not None:
             from steward.core.capabilities import has_capability, invoke_capability
             if has_capability("docker.persist_host_samples"):
                 try:
@@ -238,6 +245,7 @@ async def ingest():
                         await invoke_capability(
                             "docker.persist_host_samples", UserRole.admin,
                             session, host, docker_snapshots, latest_swarm,
+                            latest_disk,
                         )
                 except Exception:
                     current_app.logger.exception(