Say WHICH critical event, and stop the feed hiding it

Reported from production: health showed '418 critical' while the ten most
recent events were all info. Both were true, which is the problem.

The health rollup counts critical events in a 48h window; the activity feed
shows the fifteen most recent events of ANY severity. A box that logged 418
service timeouts overnight and ordinary logons since produces exactly that
pair - a page contradicting itself with no way to tell why.

Two fixes.

The rollup now names the DOMINANT event instead of totalling: 'Service timeout
x418' rather than '418 critical'. Counting by event id costs nothing (one more
GROUP BY column) and turns a number into a lead. This is the same reasoning
that made health a phrase rather than a score; it stopped one step short.

The activity feed now fetches the newest criticals separately and merges them,
and keeps up to five of them when trimming to the cap. They hold their place in
time but survive the cut, so what health is reporting always appears in the
feed that exists to explain it.

Verified against the reported shape: 20 service timeouts two hours old under 12
newer logons. Before, the feed was fifteen logons; now it carries five critical
rows and health reads 'Service timeout x20'.

Worth noting the underlying signal is real - several hundred criticals on one
endpoint in 48h is a box crash-looping or a service failing on a retry cycle.
The display was hiding it, not inventing it.

Full suite 2124 passed, naming and parity gates green.

plugins/endpoints/api/routes.py

@@ -175,7 +175,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, 0))
+        row['health'] = _health_for(state, critical.get(state.assetid))
         data.append(row)
 
     return paginated_response(data, page, per_page, total)
@@ -258,30 +258,48 @@ def _machines_for_assets(assetids):
 
 
 def _critical_counts_for_assets(assetids):
-    """{assetid: count} of critical events in the stability window."""
+    """{assetid: (total, dominant_eventid)} of critical events in the window.
+
+    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.
+    """
     if not assetids:
         return {}
     cutoff = _naive_utcnow() - timedelta(hours=48)
-    rows = (db.session.query(EndpointEvent.assetid,
+    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)
+            .group_by(EndpointEvent.assetid, EndpointEvent.eventid)
             .all())
-    return dict(rows)
+
+    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()}
 
 
-def _health_for(state, criticalcount):
+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.
+    to do; a percentage does not - and neither does "418 critical", which is
+    why the dominant event is named rather than merely counted.
     """
-    if criticalcount:
-        return {'level': 'critical',
-                'label': f'{criticalcount} critical' if criticalcount > 1 else '1 critical'}
+    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:
@@ -517,6 +535,20 @@ def panel_activity(asset_id: int):
               .limit(15)
               .all())
 
+    # The newest 15 of ANY severity can be entirely routine logons while the
+    # health rollup is shouting about a burst of crashes underneath them - the
+    # page then contradicts itself for no visible reason. Pull the newest
+    # criticals separately and merge, so what health is reporting is always
+    # present in the feed that is supposed to explain it.
+    criticals = (EndpointEvent.query
+                 .filter(EndpointEvent.assetid == asset_id,
+                         EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS))
+                 .order_by(EndpointEvent.eventtime.desc())
+                 .limit(5)
+                 .all())
+    seen = {e.endpointeventid for e in events}
+    events.extend(e for e in criticals if e.endpointeventid not in seen)
+
     commands = (EndpointCommand.query
                 .filter_by(assetid=asset_id)
                 .order_by(EndpointCommand.createdat.desc())
@@ -560,6 +592,15 @@ def panel_activity(asset_id: int):
     # Undated rows sort last rather than crashing the comparison.
     rows.sort(key=lambda r: (r['eventtime'] is not None, r['eventtime']), reverse=True)
 
+    # Trim to the cap without dropping the criticals that were fetched on
+    # purpose: they keep their place in time, but they survive the cut.
+    if len(rows) > 15:
+        keep = [r for r in rows if r['severity'] == 'critical'][:5]
+        rest = [r for r in rows if r not in keep]
+        rows = sorted(keep + rest[:15 - len(keep)],
+                      key=lambda r: (r['eventtime'] is not None, r['eventtime']),
+                      reverse=True)
+
     return success_response([{
         'title': r['title'],
         'severity': r['severity'],