feat(monitors): unify ping/dns/http into one Monitor entity + custom targets

Collapse the three former check types into a single core `Monitor` entity
with one management surface (/monitors), one result table (monitor_results),
and a single scheduled task. Every type can now watch a free-standing custom
destination (optional host_id) — not just a registered Host.

- models: Monitor + MonitorResult replace PingResult/DnsResult; Host loses its
  ping/dns facet columns (now Monitor rows linked by host_id).
- checks: monitors/{ping,dns,http}.py pure probes + runner.run_monitor
  dispatcher; one monitor_check scheduler with a per-monitor due-filter.
- status: single monitor_status_source replaces the three sources.
- UI: /monitors blueprint (type-aware add/edit/list/widget); host hub shows a
  host's linked monitors + "add monitor for this host"; nav + widget registry
  + alert metric catalog rewired. http plugin folded into core and removed.
- migration 0022 merges the http branch, data-migrates host facets +
  http_monitors + all three result histories, drops the old tables/columns.

Resolves the per-host ping/dns auto-attach issue (#275): monitors are now
explicit, never auto-added to every host.

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

steward/core/status.py

@@ -2,12 +2,11 @@
 """Unified monitor-status aggregation.
 
 A single readable "is everything up?" surface (the Status page + dashboard
-widget) is assembled from heterogeneous monitor types — core ping/DNS, plus
-any plugin that contributes (e.g. the http plugin). Rather than have the core
-status page import plugin tables (plugins are optional and loaded late), each
-monitor type registers a *status source*: an async callable(db) that returns a
-list of normalised StatusEntry objects. Core registers its ping/DNS sources in
-create_app; a plugin registers its own from setup().
+widget) is assembled from heterogeneous monitors. The core unified Monitor
+entity (ping/dns/http) contributes via `monitor_status_source`; plugins may
+register their own sources from setup(). Each source is an async callable(db)
+returning normalised StatusEntry objects so the Status page never imports
+plugin tables directly.
 """
 from __future__ import annotations
 
@@ -26,8 +25,8 @@ HEARTBEAT_COUNT = 30  # number of recent checks shown in the heartbeat bar
 @dataclass
 class StatusEntry:
     """One monitored thing, normalised across monitor types for display."""
-    kind: str                       # "ping" | "dns" | "http" | ...
-    key: str                        # unique within kind (host_id / monitor_id)
+    kind: str                       # "icmp" | "tcp" | "dns" | "http" | ...
+    key: str                        # unique within kind (monitor id)
     name: str
     target: str = ""                # address / URL shown as subtitle
     status: str = "pending"         # "up" | "down" | "pending"
@@ -98,49 +97,52 @@ def sparkline_svg(values: list[float], width: int = 80, height: int = 20,
     )
 
 
-# ── Shared query helpers (host-keyed result tables: ping, dns) ────────────────
+# ── Unified monitor status source ─────────────────────────────────────────────
 
-async def _last_n_by_host(db, model, ts_col, host_ids: list[str],
+async def _last_n_results(db, monitor_ids: list[str],
                           n: int = HEARTBEAT_COUNT) -> dict[str, list]:
-    """Return {host_id: [rows]} of the last n results per host, oldest-first."""
-    if not host_ids:
+    """Return {monitor_id: [MonitorResult]} of the last n results, oldest-first."""
+    from steward.models.monitors import MonitorResult
+    if not monitor_ids:
         return {}
     rn = func.row_number().over(
-        partition_by=model.host_id, order_by=ts_col.desc()
+        partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc()
     ).label("rn")
-    subq = select(model.id, rn).where(model.host_id.in_(host_ids)).subquery()
+    subq = select(MonitorResult.id, rn).where(
+        MonitorResult.monitor_id.in_(monitor_ids)
+    ).subquery()
     res = await db.execute(
-        select(model)
-        .join(subq, model.id == subq.c.id)
+        select(MonitorResult)
+        .join(subq, MonitorResult.id == subq.c.id)
         .where(subq.c.rn <= n)
-        .order_by(model.host_id, ts_col.asc())
+        .order_by(MonitorResult.monitor_id, MonitorResult.checked_at.asc())
     )
-    out: dict[str, list] = {hid: [] for hid in host_ids}
+    out: dict[str, list] = {mid: [] for mid in monitor_ids}
     for row in res.scalars():
-        out[row.host_id].append(row)
+        out[row.monitor_id].append(row)
     return out
 
 
-async def _uptime_by_host(db, model, ts_col, status_col, up_value,
-                          host_ids: list[str]) -> dict[str, dict]:
-    """Per-host uptime % over 24h/7d/30d in one grouped query."""
-    if not host_ids:
+async def _uptime_by_monitor(db, monitor_ids: list[str]) -> dict[str, dict]:
+    """Per-monitor uptime % over 24h/7d/30d in one grouped query."""
+    from steward.models.monitors import MonitorResult
+    if not monitor_ids:
         return {}
     now = datetime.now(timezone.utc)
     c30, c7, c24 = now - timedelta(days=30), now - timedelta(days=7), now - timedelta(hours=24)
     res = await db.execute(
         select(
-            model.host_id,
+            MonitorResult.monitor_id,
             func.count().label("total_30d"),
-            func.sum(case((status_col == up_value, 1), else_=0)).label("up_30d"),
-            func.sum(case((ts_col >= c7, 1), else_=0)).label("total_7d"),
-            func.sum(case((and_(ts_col >= c7, status_col == up_value), 1), else_=0)).label("up_7d"),
-            func.sum(case((ts_col >= c24, 1), else_=0)).label("total_24h"),
-            func.sum(case((and_(ts_col >= c24, status_col == up_value), 1), else_=0)).label("up_24h"),
+            func.sum(case((MonitorResult.is_up.is_(True), 1), else_=0)).label("up_30d"),
+            func.sum(case((MonitorResult.checked_at >= c7, 1), else_=0)).label("total_7d"),
+            func.sum(case((and_(MonitorResult.checked_at >= c7, MonitorResult.is_up.is_(True)), 1), else_=0)).label("up_7d"),
+            func.sum(case((MonitorResult.checked_at >= c24, 1), else_=0)).label("total_24h"),
+            func.sum(case((and_(MonitorResult.checked_at >= c24, MonitorResult.is_up.is_(True)), 1), else_=0)).label("up_24h"),
         )
-        .where(model.host_id.in_(host_ids))
-        .where(ts_col >= c30)
-        .group_by(model.host_id)
+        .where(MonitorResult.monitor_id.in_(monitor_ids))
+        .where(MonitorResult.checked_at >= c30)
+        .group_by(MonitorResult.monitor_id)
     )
 
     def _pct(up, total):
@@ -148,7 +150,7 @@ async def _uptime_by_host(db, model, ts_col, status_col, up_value,
 
     out: dict[str, dict] = {}
     for r in res:
-        out[r.host_id] = {
+        out[r.monitor_id] = {
             "24h": _pct(r.up_24h, r.total_24h),
             "7d": _pct(r.up_7d, r.total_7d),
             "30d": _pct(r.up_30d, r.total_30d),
@@ -156,78 +158,50 @@ async def _uptime_by_host(db, model, ts_col, status_col, up_value,
     return out
 
 
-# ── Core status sources: ping + DNS ───────────────────────────────────────────
+def _heartbeat_title(mtype: str, r) -> str:
+    """Type-aware tooltip for one heartbeat cell."""
+    ts = f"{r.checked_at:%H:%M:%S} UTC"
+    if not r.is_up:
+        return f"{(r.error_msg or 'Down')} — {ts}"
+    if mtype == "http":
+        code = r.status_code or "—"
+        ms = f"{r.response_ms:.0f} ms" if r.response_ms is not None else ""
+        return f"{code} · {ms} — {ts}".replace(" · — ", " — ")
+    if mtype == "dns":
+        return f"{r.resolved_ip or 'resolved'} — {ts}"
+    if r.response_ms is not None:
+        return f"{r.response_ms:.0f} ms — {ts}"
+    return f"Up — {ts}"
 
-async def ping_status_source(db) -> list[StatusEntry]:
-    from steward.models.hosts import Host
-    from steward.models.monitors import PingResult, PingStatus
 
-    hosts = (await db.execute(
-        select(Host).where(Host.ping_enabled.is_(True)).order_by(Host.name)
-    )).scalars().all()
-    ids = [h.id for h in hosts]
-    recent = await _last_n_by_host(db, PingResult, PingResult.probed_at, ids)
-    uptime = await _uptime_by_host(
-        db, PingResult, PingResult.probed_at, PingResult.status, PingStatus.up, ids
-    )
+async def monitor_status_source(db) -> list[StatusEntry]:
+    """Contribute every enabled Monitor (all types) to the Status page."""
+    from steward.models.monitors import Monitor
+
+    monitors = list((await db.execute(
+        select(Monitor).where(Monitor.enabled.is_(True)).order_by(Monitor.name)
+    )).scalars())
+    ids = [m.id for m in monitors]
+    recent = await _last_n_results(db, ids)
+    uptime = await _uptime_by_monitor(db, ids)
+    now = datetime.now(timezone.utc)
 
     entries: list[StatusEntry] = []
-    for h in hosts:
-        rows = recent.get(h.id, [])
+    for m in monitors:
+        rows = recent.get(m.id, [])
         latest = rows[-1] if rows else None
-        heartbeat = [{
-            "state": "down" if r.status == PingStatus.down else "up",
-            "title": (
-                f"Down — {r.probed_at:%H:%M:%S} UTC" if r.status == PingStatus.down
-                else f"{r.response_time_ms:.0f} ms — {r.probed_at:%H:%M:%S} UTC"
-                if r.response_time_ms is not None else f"Up — {r.probed_at:%H:%M:%S} UTC"
-            ),
-        } for r in rows]
-        status = "pending" if latest is None else (
-            "down" if latest.status == PingStatus.down else "up"
-        )
+        status = "pending" if latest is None else ("up" if latest.is_up else "down")
+        tls_days = None
+        if latest and latest.tls_expires_at:
+            tls_days = (latest.tls_expires_at - now).total_seconds() / 86400.0
         entries.append(StatusEntry(
-            kind="ping", key=h.id, name=h.name, target=h.address, status=status,
-            last_checked=latest.probed_at if latest else None,
-            uptime=uptime.get(h.id, {}), heartbeat=heartbeat,
-            latency_ms=latest.response_time_ms if latest else None,
-            spark=[r.response_time_ms for r in rows if r.response_time_ms is not None],
-            detail_url="/ping/",
-        ))
-    return entries
-
-
-async def dns_status_source(db) -> list[StatusEntry]:
-    from steward.models.hosts import Host
-    from steward.models.monitors import DnsResult, DnsStatus
-
-    hosts = (await db.execute(
-        select(Host).where(Host.dns_enabled.is_(True)).order_by(Host.name)
-    )).scalars().all()
-    ids = [h.id for h in hosts]
-    recent = await _last_n_by_host(db, DnsResult, DnsResult.resolved_at, ids)
-    uptime = await _uptime_by_host(
-        db, DnsResult, DnsResult.resolved_at, DnsResult.status, DnsStatus.resolved, ids
-    )
-
-    entries: list[StatusEntry] = []
-    for h in hosts:
-        rows = recent.get(h.id, [])
-        latest = rows[-1] if rows else None
-        heartbeat = [{
-            "state": "up" if r.status == DnsStatus.resolved else "down",
-            "title": (
-                f"{r.resolved_ip} — {r.resolved_at:%H:%M:%S} UTC" if r.status == DnsStatus.resolved
-                else f"Failed — {r.resolved_at:%H:%M:%S} UTC"
-            ),
-        } for r in rows]
-        status = "pending" if latest is None else (
-            "up" if latest.status == DnsStatus.resolved else "down"
-        )
-        entries.append(StatusEntry(
-            kind="dns", key=h.id, name=h.name, target=h.address, status=status,
-            last_checked=latest.resolved_at if latest else None,
-            uptime=uptime.get(h.id, {}), heartbeat=heartbeat,
-            detail_url="/dns/",
+            kind=m.type, key=m.id, name=m.name, target=m.target, status=status,
+            last_checked=latest.checked_at if latest else None,
+            uptime=uptime.get(m.id, {}),
+            heartbeat=[{"state": "up" if r.is_up else "down",
+                        "title": _heartbeat_title(m.type, r)} for r in rows],
+            latency_ms=latest.response_ms if latest else None,
+            spark=[r.response_ms for r in rows if r.response_ms is not None],
+            tls_days=tls_days, detail_url="/monitors/",
         ))
     return entries