Make the Fleet list answer the question it exists for

The fleet list showed hostname, user, uptime, CPU, memory and last-reported,
sorted by last-reported descending. Two of those columns have never held a
value, and that sort floats the healthiest endpoints to the top and buries the
ones somebody needs to act on.

What a shop-floor fleet view is actually for is which MACHINE stops if a PC is
dead, so that is now a column. Both edge directions are checked, deduplicated,
and constrained to assets typed 'machine' - without that last filter the column
filled up with office locations that happen to carry a controls edge.

Also here:

  - Sorted WORST FIRST: never reported, then quiet, then reporting. Expressed
    as a sortable SQL rank rather than a label applied after the fact, so the
    order survives pagination.
  - Health rollup as a PHRASE, not a score: 'eDNC stopped', '2 critical'. A
    percentage does not tell an operator what to do.
  - Filters for status, PC type, location and business unit; a sort parameter
    (the route accepted none); and search widened from hostname alone to asset
    number and the machine number a PC serves, which is how people actually
    look a bay up.
  - CPU and Memory columns dropped. The agent declares those fields and has
    never assigned them, so they rendered '-' on every row.
  - A summary strip wired to the dashboard routes, which existed and which
    nothing called.

Two bugs found while building it:

The health rollup first read a 'running' boolean, but service rows carry a
STATE STRING and the vocabulary varies by source, so every service would have
reported as stopped. The panel and the rollup now share one _service_running()
helper; two copies of that rule would drift.

The summary counted quiet as total - reporting, which folded never-reported
endpoints in with quiet ones. Those are different failures - an enrolment or
key problem versus a machine that went down - so 'never' is counted separately.

Per-page batching for machines and critical-event counts: three queries per
page rather than two per row.

Full suite 2124 passed, vitest 260 passed, naming and contract gates green.

plugins/endpoints/api/routes.py

@@ -31,6 +31,22 @@ def _naive_utcnow():
     return datetime.now(timezone.utc).replace(tzinfo=None)
 
 
+# A watched service or process reports a STATE STRING, not a boolean, and the
+# vocabulary varies by source. Both the services panel and the fleet health
+# rollup have to agree on what counts as running, so the rule lives here once -
+# reading it wrong makes every service look stopped.
+_RUNNING_STATES = ('running', 'started', 'up', 'ok', 'active')
+
+
+def _service_running(entry):
+    raw = (entry.get('state') or entry.get('status') or '').strip().lower()
+    return raw in _RUNNING_STATES
+
+
+def _service_name(entry):
+    return entry.get('name') or entry.get('service') or 'Unknown'
+
+
 def _format_uptime(seconds):
     """Whole-seconds uptime -> 'Nd Nh Nm', or None."""
     if seconds is None:
@@ -77,32 +93,215 @@ def list_endpoints():
     Joins through Computer so each row carries computerid and can deep-link
     into the existing PC detail page (/pcs/{computerid}). Endpoint state only
     exists for hostnames resolved to a Computer, so the join is not lossy.
+
+    Sorted WORST FIRST by default: never-reported, then quiet, then reporting.
+    It used to sort by lastreportedat descending, which floated the healthiest
+    endpoints to the top and buried the ones somebody needs to look at.
+
+    Each row also carries what the fleet view is actually for: which MACHINE
+    depends on this PC, its PC type, and a health rollup. On a shop floor a PC
+    matters because a machine stops without it, so that is the column an
+    operator reads first.
     """
-    from plugins.computers.models import Computer
+    from plugins.computers.models import Computer, ComputerType
 
     page, per_page = get_pagination_params(request)
+    cutoff = _naive_utcnow() - timedelta(hours=_quiet_hours())
 
-    query = (db.session.query(EndpointState, Computer.computerid)
+    # Status as a sortable expression rather than a post-hoc label, so
+    # worst-first survives pagination. 0 never reported, 1 quiet, 2 reporting.
+    statusrank = db.case(
+        (EndpointState.lastreportedat.is_(None), 0),
+        (EndpointState.lastreportedat < cutoff, 1),
+        else_=2,
+    )
+
+    query = (db.session.query(EndpointState, Computer.computerid,
+                              Computer.computertypeid, ComputerType.computertype,
+                              Asset.assetnumber, statusrank.label('statusrank'))
              .join(Asset, Asset.assetid == EndpointState.assetid)
              .join(Computer, Computer.assetid == EndpointState.assetid)
+             .outerjoin(ComputerType,
+                        ComputerType.computertypeid == Computer.computertypeid)
              .filter(Asset.isactive.is_(True)))
 
-    if search := request.args.get('search'):
-        query = query.filter(EndpointState.hostname.ilike(f'%{search}%'))
+    # Search covers hostname, asset number and the machine number a PC serves.
+    # Hostname alone was too narrow: people look a PC up by the bay it runs.
+    if search := (request.args.get('search') or '').strip():
+        like = f'%{search}%'
+        machinehits = _assetids_serving_machine_like(like)
+        clauses = [EndpointState.hostname.ilike(like), Asset.assetnumber.ilike(like)]
+        if machinehits:
+            clauses.append(EndpointState.assetid.in_(machinehits))
+        query = query.filter(db.or_(*clauses))
 
-    query = query.order_by(EndpointState.lastreportedat.desc())
+    if status := request.args.get('status'):
+        wanted = {'never': 0, 'quiet': 1, 'reporting': 2}.get(status.lower())
+        if wanted is not None:
+            query = query.filter(statusrank == wanted)
+
+    if computertypeid := request.args.get('computertypeid'):
+        query = query.filter(Computer.computertypeid == computertypeid)
+    if locationid := request.args.get('locationid'):
+        query = query.filter(Asset.locationid == locationid)
+    if businessunitid := request.args.get('businessunitid'):
+        query = query.filter(Asset.businessunitid == businessunitid)
+
+    sort = (request.args.get('sort') or 'status').lower()
+    sorts = {
+        'status':       (statusrank.asc(), EndpointState.hostname.asc()),
+        'hostname':     (EndpointState.hostname.asc(),),
+        'lastreported': (EndpointState.lastreportedat.desc(),),
+        'uptime':       (EndpointState.uptimeseconds.desc(),),
+    }
+    query = query.order_by(*sorts.get(sort, sorts['status']))
 
     items, total = paginate_query(query, page, per_page)
+
+    # Batch the per-row extras for this PAGE only. Doing them inline would be
+    # 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)
+
     data = []
-    for state, computerid in items:
+    for state, computerid, computertypeid, computertype, assetnumber, rank in items:
         row = state.to_dict()
         row['computerid'] = computerid
+        row['assetnumber'] = assetnumber
+        row['computertypeid'] = computertypeid
+        row['computertype'] = computertype
         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))
         data.append(row)
 
     return paginated_response(data, page, per_page, total)
 
 
+def _assetids_serving_machine_like(like):
+    """Asset ids of PCs whose controlled machine matches a search term.
+
+    Checked in BOTH directions on purpose. The canonical edge is PC controls
+    machine, but production also holds reciprocal rows, so matching only
+    source->target silently drops half the fleet from a machine-number search.
+    """
+    from shopdb.api import AssetRelationship, RelationshipType, AssetType
+
+    controls = (RelationshipType.query
+                .filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
+                .first())
+    if controls is None:
+        return []
+
+    # Only MACHINE assets. A PC also 'controls' printers and other kit, and
+    # matching any controlled asset put office locations in the machine column.
+    other = db.aliased(Asset)
+    othertype = db.aliased(AssetType)
+    forward = (db.session.query(AssetRelationship.sourceassetid)
+               .join(other, other.assetid == AssetRelationship.targetassetid)
+               .join(othertype, othertype.assettypeid == other.assettypeid)
+               .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+                       db.func.lower(othertype.assettype) == 'machine',
+                       other.assetnumber.ilike(like)))
+    backward = (db.session.query(AssetRelationship.targetassetid)
+                .join(other, other.assetid == AssetRelationship.sourceassetid)
+                .join(othertype, othertype.assettypeid == other.assettypeid)
+                .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+                        db.func.lower(othertype.assettype) == 'machine',
+                        other.assetnumber.ilike(like)))
+    return [r[0] for r in forward.union(backward).all()]
+
+
+def _machines_for_assets(assetids):
+    """{assetid: [{assetid, assetnumber, name}]} for the machines each PC serves.
+
+    Both edge directions again, deduplicated by machine asset id. One query for
+    the whole page rather than one per row.
+    """
+    from shopdb.api import AssetRelationship, RelationshipType, AssetType
+
+    if not assetids:
+        return {}
+
+    controls = (RelationshipType.query
+                .filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
+                .first())
+    if controls is None:
+        return {}
+
+    other = db.aliased(Asset)
+    othertype = db.aliased(AssetType)
+    forward = (db.session.query(AssetRelationship.sourceassetid.label('pc'),
+                                other.assetid, other.assetnumber, other.name)
+               .join(other, other.assetid == AssetRelationship.targetassetid)
+               .join(othertype, othertype.assettypeid == other.assettypeid)
+               .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+                       db.func.lower(othertype.assettype) == 'machine',
+                       AssetRelationship.sourceassetid.in_(assetids)))
+    backward = (db.session.query(AssetRelationship.targetassetid.label('pc'),
+                                 other.assetid, other.assetnumber, other.name)
+                .join(other, other.assetid == AssetRelationship.sourceassetid)
+                .join(othertype, othertype.assettypeid == other.assettypeid)
+                .filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
+                        db.func.lower(othertype.assettype) == 'machine',
+                        AssetRelationship.targetassetid.in_(assetids)))
+
+    out = {}
+    for pc, machineid, assetnumber, name in forward.union(backward).all():
+        seen = out.setdefault(pc, [])
+        if not any(m['assetid'] == machineid for m in seen):
+            seen.append({'assetid': machineid, 'assetnumber': assetnumber, 'name': name})
+    return out
+
+
+def _critical_counts_for_assets(assetids):
+    """{assetid: count} of critical events in the stability window."""
+    if not assetids:
+        return {}
+    cutoff = _naive_utcnow() - timedelta(hours=48)
+    rows = (db.session.query(EndpointEvent.assetid,
+                             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)
+            .all())
+    return dict(rows)
+
+
+def _health_for(state, criticalcount):
+    """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.
+    """
+    if criticalcount:
+        return {'level': 'critical',
+                'label': f'{criticalcount} critical' if criticalcount > 1 else '1 critical'}
+
+    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'}
+
+
 @endpoints_bp.route('/<int:asset_id>', methods=['GET'])
 @require_permission('endpoints.view')
 def get_endpoint(asset_id: int):
@@ -223,7 +422,7 @@ def panel_services(asset_id: int):
             kind = entry.get('kind')
         else:
             name, raw, app, kind = str(entry), '', None, None
-        running = raw in ('running', 'started', 'up', 'ok', 'active')
+        running = _service_running(entry)
         row = {
             'name': name,
             'kind': (kind or '').lower() or None,
@@ -331,10 +530,22 @@ def dashboard_summary():
         EndpointState.lastreportedat >= cutoff,
     ).count()
 
+    # Counted separately from quiet. An endpoint that has NEVER reported is a
+    # different problem from one that reported and stopped: the first is an
+    # enrolment or key failure, the second is a machine that went down. Folding
+    # them together (quiet = total - reporting) hid the enrolment failures.
+    never = db.session.query(EndpointState).join(
+        Asset, Asset.assetid == EndpointState.assetid
+    ).filter(
+        Asset.isactive.is_(True),
+        EndpointState.lastreportedat.is_(None),
+    ).count()
+
     return success_response({
         'total': total,
         'reporting': reporting,
-        'quiet': total - reporting,
+        'quiet': total - reporting - never,
+        'never': never,
         'quiethours': hours,
     })