perf(host_agent): aggregate metric history in SQL; DISTINCT ON for latest

Follow-on to the plugin_metrics indexes. plugin_metrics is already retention-
bounded (core cleanup prunes > data.retention_days, default 90d) and charts top
out at 30d, so the cost wasn't growth — it was the read path shipping raw rows
to Python.

- _history_for_host: bucket + average in SQL via date_bin (epoch-aligned, ~120
  buckets) instead of fetching every raw sample (a 30d range was hundreds of
  thousands of rows) and downsampling in Python. Uses the new
  (source_module, resource_name, recorded_at) index.
- _latest_metrics_for_host: DISTINCT ON (resource_name, metric_name) ORDER BY
  recorded_at DESC — newest row per group in one index-ordered pass, replacing
  the GROUP-BY-max subquery self-joined back to the whole history.
- Integration test validates both against Postgres.

Deliberately not a materialized raw→hourly rollup: these query-side changes
deliver the speed; a rollup would additionally cut storage and remains a future
option if scale demands it.

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

plugins/host_agent/routes.py

@@ -15,7 +15,7 @@ from sqlalchemy import select, func, or_, and_
 from datetime import timedelta
 
 from steward.core.settings import public_base_url
-from steward.core.time_range import parse_range, RANGE_OPTIONS
+from steward.core.time_range import parse_range, RANGE_OPTIONS, bucket_seconds
 from steward.models.hosts import Host
 from steward.models.metrics import PluginMetric
 from .models import HostAgentRegistration
@@ -550,27 +550,28 @@ HISTORY_METRICS = (
 
 
 async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[str, float]]:
-    """{resource_name: {metric: value}} of the latest sample for a host + sub-resources."""
+    """{resource_name: {metric: value}} — latest sample per (resource, metric) for
-    subq = (
+    a host + its sub-resources.
-        select(
+
+    DISTINCT ON picks the newest row per group in one index-ordered pass over
+    ix_plugin_metrics_module_resource_metric_recorded, instead of a GROUP-BY-max
+    subquery self-joined back to the table (two passes over the whole history).
+    """
+    rows = (await session.execute(
+        select(PluginMetric)
+        .where(
+            PluginMetric.source_module == SOURCE_MODULE,
+            or_(
+                PluginMetric.resource_name == host_name,
+                PluginMetric.resource_name.like(host_name + ":%"),
+            ),
+        )
+        .distinct(PluginMetric.resource_name, PluginMetric.metric_name)
+        .order_by(
             PluginMetric.resource_name,
             PluginMetric.metric_name,
-            func.max(PluginMetric.recorded_at).label("max_ts"),
+            PluginMetric.recorded_at.desc(),
         )
-        .where(PluginMetric.source_module == SOURCE_MODULE)
-        .where(or_(
-            PluginMetric.resource_name == host_name,
-            PluginMetric.resource_name.like(host_name + ":%"),
-        ))
-        .group_by(PluginMetric.resource_name, PluginMetric.metric_name)
-    ).subquery()
-    rows = (await session.execute(
-        select(PluginMetric).join(
-            subq,
-            (PluginMetric.resource_name == subq.c.resource_name) &
-            (PluginMetric.metric_name == subq.c.metric_name) &
-            (PluginMetric.recorded_at == subq.c.max_ts),
-        ).where(PluginMetric.source_module == SOURCE_MODULE)
     )).scalars().all()
     out: dict[str, dict[str, float]] = {}
     for r in rows:
@@ -579,24 +580,35 @@ async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[st
 
 
 async def _history_for_host(session, host_name: str, since) -> dict[str, list]:
-    """{metric: [[epoch_ms, value], …]} host-level series since `since`.
+    """{metric: [[epoch_ms, avg_value], …]} host-level series since `since`.
 
-    Epoch-ms x values let the charts use a linear axis (no Chart.js date adapter).
+    Buckets + averages in SQL (date_bin to ~120 buckets) so we return a readable
+    point count instead of shipping every raw sample to Python and downsampling
+    there — a 30d range was reading hundreds of thousands of rows per load. The
+    bucket width is epoch-aligned so the x axis is stable across refreshes.
+    Epoch-ms x values feed a linear chart axis (no Chart.js date adapter).
     """
+    width_s = bucket_seconds(since, 120)
+    bucket = func.date_bin(
+        func.make_interval(0, 0, 0, 0, 0, 0, width_s),  # width_s seconds
+        PluginMetric.recorded_at,
+        func.to_timestamp(0),                            # epoch origin
+    ).label("bucket")
     rows = (await session.execute(
-        select(PluginMetric).where(
+        select(PluginMetric.metric_name, bucket, func.avg(PluginMetric.value))
+        .where(
             PluginMetric.source_module == SOURCE_MODULE,
             PluginMetric.resource_name == host_name,
             PluginMetric.metric_name.in_(HISTORY_METRICS),
             PluginMetric.recorded_at >= since,
-        ).order_by(PluginMetric.recorded_at)
+        )
-    )).scalars().all()
+        .group_by(PluginMetric.metric_name, bucket)
+        .order_by(bucket)
+    )).all()
     series: dict[str, list] = {m: [] for m in HISTORY_METRICS}
-    for p in rows:
+    for metric_name, b, avg in rows:
-        series[p.metric_name].append([int(p.recorded_at.timestamp() * 1000), round(p.value, 2)])
+        series[metric_name].append([int(b.timestamp() * 1000), round(float(avg), 2)])
-    # Downsample to a readable point count (see _downsample) — raw agent cadence
+    return series
-    # is too dense to read over a multi-hour window.
-    return {m: _downsample(v) for m, v in series.items()}
 
 
 @host_agent_bp.get("/<host_id>/")

tests/integration/test_host_metrics.py

@@ -0,0 +1,80 @@
+"""Integration: host-metrics read paths (DISTINCT ON latest + SQL date_bin history).
+
+Validates the two host_agent query helpers that back the slow host views, against
+a live Postgres: the latest-per-(resource,metric) lookup and the bucket-averaged
+history (which now aggregates in SQL instead of shipping raw rows to Python).
+Requires STEWARD_DATABASE_URL.
+"""
+from __future__ import annotations
+
+import asyncio
+import os
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+pytestmark = pytest.mark.integration
+
+_NEEDS_DB = pytest.mark.skipif(
+    not os.environ.get("STEWARD_DATABASE_URL"),
+    reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)",
+)
+
+
+@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)
+
+
+@_NEEDS_DB
+def test_latest_distinct_on_and_sql_bucketed_history(app):
+    from sqlalchemy import text
+    from steward.models.metrics import PluginMetric
+    from plugins.host_agent.routes import (
+        SOURCE_MODULE, _history_for_host, _latest_metrics_for_host,
+    )
+
+    now = datetime.now(timezone.utc)
+    hostname = "metrics-host-" + uuid.uuid4().hex[:8]
+
+    async def _go():
+        async with app.db_sessionmaker() as s:
+            async with s.begin():
+                await s.execute(
+                    text("DELETE FROM plugin_metrics WHERE resource_name LIKE :p"),
+                    {"p": hostname + "%"},
+                )
+                rows = []
+                # Host-level CPU samples, oldest → newest (10, 20, 30).
+                for i, val in enumerate([10.0, 20.0, 30.0]):
+                    rows.append(PluginMetric(
+                        source_module=SOURCE_MODULE, resource_name=hostname,
+                        metric_name="cpu_pct", value=val,
+                        recorded_at=now - timedelta(minutes=30 - i * 10),
+                    ))
+                # A sub-resource (root mount) to exercise the host:% match.
+                rows.append(PluginMetric(
+                    source_module=SOURCE_MODULE, resource_name=hostname + ":/",
+                    metric_name="disk_used_pct", value=80.0,
+                    recorded_at=now - timedelta(minutes=5),
+                ))
+                s.add_all(rows)
+
+            latest = await _latest_metrics_for_host(s, hostname)
+            hist = await _history_for_host(s, hostname, now - timedelta(hours=1))
+            return latest, hist
+
+    latest, hist = asyncio.run(_go())
+
+    # DISTINCT ON returns the newest sample per (resource, metric).
+    assert latest[hostname]["cpu_pct"] == 30.0
+    assert latest[hostname + ":/"]["disk_used_pct"] == 80.0
+
+    # SQL date_bin aggregation returns cpu_pct buckets; averages stay in range.
+    cpu = hist["cpu_pct"]
+    assert cpu, "expected bucketed cpu_pct history"
+    assert all(10.0 <= v <= 30.0 for _, v in cpu)