Cache endpoint health as a ranked rollup

Fleet ranked endpoints by raw critical-event counts, so a machine with a
noisy but harmless event log outranked one with a mapped drive down or a
blocking service stopped. Volume is not severity, and the aggregation ran
across the whole fleet on every page load.

Rank operational blockers above degraded conditions above event noise, and
cache the verdict per asset so Fleet reads a row instead of recomputing 48h
of history. Transitions are written only when a verdict changes, capped at
50 rows per asset.

Absence of a policy binding must never suppress a verdict: drives and events
are not policy-gated, and services fall back to a built-in default impact
when no mapping applies. Event 1001 drops to informational because the WER
bucket pairs with 1000 and double-counted every crash.

Parallelize the Fleet first-paint loaders that were awaiting serially.

plugins/endpoints/api/routes.py

@@ -27,7 +27,7 @@ from shopdb.api import (
 
 from ..models import (
     EndpointState, EndpointEvent, EndpointCommand, EndpointApplication,
-    OperatorReport,
+    EndpointHealthState, OperatorReport,
 )
 from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for
 from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
@@ -258,7 +258,10 @@ def list_endpoints():
     # one query per endpoint for machines and another for events.
     assetids = [state.assetid for state, *_ in items]
     machines = _machines_for_assets(assetids)
-    critical = _critical_counts_for_assets(assetids)
+    states = {state.assetid: state for state, *_ in items}
+    computertypes = {state.assetid: computertypeid
+                     for state, _computerid, computertypeid, *_ in items}
+    health = _health_for_assets(states, computertypes)
 
     data = []
     for state, computerid, computertypeid, computertype, assetnumber, rank in items:
@@ -270,7 +273,7 @@ def list_endpoints():
         row['uptimedisplay'] = _format_uptime(state.uptimeseconds)
         row['status'] = ('never', 'quiet', 'reporting')[rank]
         row['machines'] = machines.get(state.assetid, [])
-        row['health'] = _health_for(state, critical.get(state.assetid))
+        row['health'] = health.get(state.assetid)
         data.append(row)
 
     return paginated_response(data, page, per_page, total)
@@ -352,68 +355,41 @@ def _machines_for_assets(assetids):
     return out
 
 
-def _critical_counts_for_assets(assetids):
-    """{assetid: (total, dominant_eventid)} of critical events in the window.
+def _health_for_assets(states, computertypes):
+    """{assetid: {'level', 'label'}} for a page, from the cached rollup.
 
-    Broken down BY EVENT ID, not just totalled. A bare count says "418
-    critical", which tells an operator nothing about what to do; naming the
-    event that dominates - service timeout, application crash - is the
-    difference between a number and a lead.
+    ONE query for the page instead of the old per-page critical-event
+    aggregate, and no ranking work here - the verdict was decided by
+    services/health.py in the same commit as the telemetry it came from.
+
+    Endpoints with no cached row have not checked in since the rollup shipped.
+    Those are evaluated read-only for display rather than left blank, because a
+    blank Health column reads as "fine" and not doing that is this page's whole
+    job. Nothing is persisted on a GET: writing here would invent a firstseenat
+    no agent ever observed, and would make a page load look like an
+    observation.
     """
-    if not assetids:
+    if not states:
         return {}
-    cutoff = _naive_utcnow() - timedelta(hours=48)
-    rows = (db.session.query(EndpointEvent.assetid, EndpointEvent.eventid,
-                             db.func.count(EndpointEvent.endpointeventid))
-            .filter(EndpointEvent.assetid.in_(assetids),
-                    EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS),
-                    EndpointEvent.eventtime.isnot(None),
-                    EndpointEvent.eventtime >= cutoff)
-            .group_by(EndpointEvent.assetid, EndpointEvent.eventid)
-            .all())
 
-    perasset = {}
-    for assetid, eventid, count in rows:
-        total, dominant, best = perasset.get(assetid, (0, None, 0))
-        total += count
-        if count > best:
-            dominant, best = eventid, count
-        perasset[assetid] = (total, dominant, best)
-    return {assetid: (total, dominant)
-            for assetid, (total, dominant, _best) in perasset.items()}
+    from ..services.health import _verdict
 
+    assetids = list(states)
+    cached = {row.assetid: row for row in
+              EndpointHealthState.query
+              .filter(EndpointHealthState.assetid.in_(assetids))
+              .all()}
 
-def _health_for(state, critical):
-    """Worst-of rollup: a stopped watched service, or a recent critical event.
-
-    Deliberately a PHRASE, not a score. "eDNC stopped" tells an operator what
-    to do; a percentage does not - and neither does "418 critical", which is
-    why the dominant event is named rather than merely counted.
-    """
-    if critical:
-        total, dominant = critical
-        title = event_title(dominant) if dominant else 'critical'
-        label = title if total == 1 else '{} x{}'.format(title, total)
-        return {'level': 'critical', 'label': label}
-
-    stopped = []
-    try:
-        for entry in json.loads(state.servicesjson or '[]'):
-            if not _service_running(entry):
-                stopped.append(_service_name(entry))
-    except (TypeError, ValueError):
-        # Malformed payload is not a health verdict - say nothing rather than
-        # claiming the endpoint is unhealthy.
-        return {'level': 'unknown', 'label': None}
-
-    if stopped:
-        first = stopped[0]
-        extra = f' +{len(stopped) - 1}' if len(stopped) > 1 else ''
-        return {'level': 'warning', 'label': f'{first} stopped{extra}'}
-
-    if state.lastreportedat is None:
-        return {'level': 'unknown', 'label': 'no agent'}
-    return {'level': 'ok', 'label': 'ok'}
+    now = _naive_utcnow()
+    out = {}
+    for assetid, state in states.items():
+        row = cached.get(assetid)
+        if row is not None:
+            out[assetid] = {'level': row.level, 'label': row.label}
+            continue
+        verdict = _verdict(state, computertypes.get(assetid), assetid, now)
+        out[assetid] = {'level': verdict['level'], 'label': verdict['label']}
+    return out
 
 
 @endpoints_bp.route('/<int:asset_id>', methods=['GET'])