Fix fleet alert server surface inherited from wj

The wj merge brought in a half-built fleet alert feature: server-side UI
behaviour still being stood up over a delivery proof of concept. It left
five red tests. This clears them and fixes a real hole underneath.

The cross-machine guard never ran. acknowledge_fleet_alert read .assetid
off _resolve_asset_for_hostname, which returns the id itself, so the
check raised AttributeError on every call and answered 500. That guard is
what stops one endpoint acknowledging another's alert, and the collector
key is shared across the fleet, so until now any bay could clear an alert
on another machine's behalf. Compare the id directly.

Gate both alert routes the way the rest of the agent channel is gated.
They are collector-key authenticated, not JWT+role, because a shop floor
bay has no user session - the same shape as claim_next_command and
upload_operator_report. Declared in the authz allowlists with the reason,
rather than inventing a permission no agent could hold.

Settle the delivery contract ahead of the durable agent work. The poll
response is authoritative for a host: every alert that should be live on
it right now. An alert absent from the feed is the agent's instruction to
retire the pending toast, which is how a server-side close or cancel
reaches a machine already displaying one. Acknowledgement is terminal per
product decision, so a close after an ack is a no-op and the agent can
retry a queued ack safely. A toast that merely timed out is not an
acknowledgement.

Stop reaching past the plugin contract surface for the mailer. One import
was a plain redirect; the other pulled a private helper to pre-check
recipients, duplicating what send_alert already does. It also returned
early when no email recipients existed, which silently suppressed the
independent webhook leg - misconfigured SMTP was dropping Teams alerts.

Register NotificationAcknowledgement in get_models() and give the
notifications plugin ownership of notificationacknowledgements, then move
the expected head to notifications0006fleetalerts.

Add the coverage whose absence let the .assetid bug ship. Thirteen tests
over gating, authoritative-feed withdrawal, terminal and idempotent
acknowledgement, per-asset scoping, and the cross-machine guard. Five of
them fail against the pre-fix handler.

Remaining suite failures are pre-existing and local: the symlink scan
needs elevation, and the docs check wants files absent from this checkout.

tests/test_core/test_authz.py

@@ -110,7 +110,18 @@ EXEMPT_ENDPOINTS = {'knowledgebase.track_click', 'users.update_user',
                     # collector key it already holds. The human-facing read and
                     # image-serve routes ARE permission-gated on endpoints.view
                     # and ARE swept.
-                    'endpoints.upload_operator_report'}
+                    'endpoints.upload_operator_report',
+                    # Agent-facing fleet alert acknowledgement. Authenticated by
+                    # the endpoints collector API key (same shape as the exempt
+                    # collector blueprint), not JWT+role: the operator dismissing
+                    # an alert on a shop-floor bay has no user session, and the
+                    # tray hands the acknowledgement to the service, which posts
+                    # it on the collector key it already holds. The handler binds
+                    # the acknowledgement to the asset the reported hostname
+                    # resolves to, so the key cannot ack another machine's alert.
+                    # AUTHORING a fleet alert is the actual authority and stays
+                    # permission-gated in the notifications plugin.
+                    'endpoints.acknowledge_fleet_alert'}
 
 
 @pytest.fixture(autouse=True)

tests/test_core/test_public_read_surface.py

@@ -120,6 +120,15 @@ PUBLIC_ENDPOINTS = {
     # It is a read of already-scoped policy output, not an authority: attaching a
     # monitor is gated on the policy editor, audited with the requesting user.
     'endpoints.resolved_monitors_for_agent',
+
+    # The endpoint agent fetches the active, unacknowledged fleet alerts for its
+    # own hostname so the tray can surface them to the operator. Authenticated
+    # inside the handler by the endpoints collector API key - the same shape as
+    # claim_next_command - because a bay has no user session. The response is
+    # scoped to the asset the reported hostname resolves to, so this reads one
+    # machine's alerts rather than the fleet's. Authoring a fleet alert is the
+    # actual authority and stays gated in the notifications plugin.
+    'endpoints.get_fleet_alerts',
 }
 
 # Reads that need a LOGIN but no named permission, because the handler decides

tests/test_plugin_migrations.py

@@ -88,8 +88,9 @@ EXPECTED_HEAD_REVISION['printedparts'] = 'printedparts0004txnrev'
 # send needs one; every other tool in that plugin is browser-side.
 EXPECTED_HEAD_REVISION['tools'] = 'tools0001securesend'
 # notifications indexes businessunitid, then adds the per-type grace window and
-# the shared board category.
-EXPECTED_HEAD_REVISION['notifications'] = 'notifications0005boardorder'
+# the shared board category, then the fleet-alert columns plus the per-asset
+# acknowledgement table the endpoint agent writes to.
+EXPECTED_HEAD_REVISION['notifications'] = 'notifications0006fleetalerts'
 # warranty adds the proof-of-cover document columns on top of its anchor.
 EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof'
 # endpoints is a NEW post-cutover plugin: its 0001 genuinely creates the

tests/test_plugins/test_endpoints_fleet_alerts.py

@@ -0,0 +1,311 @@
+"""Fleet alert delivery to the endpoint agent, and terminal acknowledgement.
+
+A fleet alert is a notification flagged alertfleet. The endpoint agent polls
+GET /api/endpoints/alerts?hostname=... on the endpoints collector key - the same
+credential and kill switch as the rest of the agent channel, because a shop
+floor bay has no user session - and the tray surfaces what comes back. When the
+operator dismisses one, the agent posts it to the acknowledge route.
+
+Two contracts matter here and both are load-bearing for the durable agent-side
+surface:
+
+  1. The poll response is AUTHORITATIVE for that host. It is every alert that
+     should be live on the machine right now, so the agent can retire a pending
+     alert that has disappeared from it. That is the only way a server-side
+     close or cancel reaches an endpoint already displaying the alert.
+
+  2. Acknowledgement is TERMINAL for an asset. Once acked, the alert never comes
+     back for that machine, and a later server-side close is a no-op. This is
+     what lets the agent retry a queued acknowledgement safely.
+
+The cross-machine guard gets its own test because it regressed into an
+AttributeError: the handler read .assetid off _resolve_asset_for_hostname,
+which returns the id itself, so the check raised on every call and one bay could
+not be proven unable to acknowledge another bay's alert.
+"""
+
+from datetime import timedelta
+
+from shopdb.core.models import Asset, AssetType
+
+
+ALERTS_URL = '/api/endpoints/alerts'
+ACK_URL = '/api/endpoints/alerts/{}/acknowledge'
+COLLECTOR_KEY = 'testcollectorkey'
+
+
+def _collector_headers(app):
+    app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY
+    return {'X-API-Key': COLLECTOR_KEY}
+
+
+def _computer(db, hostname):
+    """Create a computer Asset + Computer; return the asset."""
+    from plugins.computers.models import Computer, ComputerType
+
+    atype = AssetType.query.filter_by(assettype='computer').first()
+    if not atype:
+        atype = AssetType(assettype='computer')
+        db.session.add(atype)
+        db.session.flush()
+
+    ctype = ComputerType.query.filter_by(computertype='CMM PC').first()
+    if not ctype:
+        ctype = ComputerType(computertype='CMM PC')
+        db.session.add(ctype)
+        db.session.flush()
+
+    asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
+                  isactive=True)
+    db.session.add(asset)
+    db.session.flush()
+    comp = Computer(assetid=asset.assetid, hostname=hostname,
+                    computertypeid=ctype.computertypeid)
+    db.session.add(comp)
+    db.session.commit()
+    return asset
+
+
+def _alert_type(db):
+    from plugins.notifications.models import NotificationType
+
+    ntype = NotificationType.query.filter_by(typename='Fleet Alert').first()
+    if not ntype:
+        ntype = NotificationType(typename='Fleet Alert', typecolor='#d93f3f',
+                                 isactive=True)
+        db.session.add(ntype)
+        db.session.commit()
+    return ntype
+
+
+def _fleet_alert(db, message='Line 3 is down', isactive=True, endtime=None,
+                 timeout=60):
+    from plugins.notifications.models import Notification
+
+    ntype = _alert_type(db)
+    note = Notification(notificationtypeid=ntype.notificationtypeid,
+                        notification=message, alertfleet=True,
+                        alerttimeout=timeout, isactive=isactive,
+                        endtime=endtime)
+    db.session.add(note)
+    db.session.commit()
+    return note
+
+
+def _naive_utcnow():
+    from datetime import datetime, timezone
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+def test_agent_polls_its_active_fleet_alerts(client, db, app):
+    """An active fleet alert reaches the endpoint that asks for it."""
+    _computer(db, 'BAY-01')
+    alert = _fleet_alert(db, message='Line 3 is down')
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01',
+                      headers=_collector_headers(app))
+
+    assert resp.status_code == 200, resp.get_data(as_text=True)
+    alerts = resp.get_json()['data']
+    assert [a['notificationId'] for a in alerts] == [alert.notificationid]
+    assert alerts[0]['message'] == 'Line 3 is down'
+    assert alerts[0]['timeout'] == 60
+    # The tray renders the type colour, so it has to come down with the alert
+    # rather than being guessed client side.
+    assert alerts[0]['typeColor'] == '#d93f3f'
+
+
+def test_poll_requires_the_collector_key(client, db, app):
+    """The alert feed is agent-facing, not anonymous."""
+    _computer(db, 'BAY-01')
+    _fleet_alert(db)
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01')
+
+    assert resp.status_code in (401, 403), resp.get_data(as_text=True)
+
+
+def test_unknown_hostname_is_not_found(client, db, app):
+    """A hostname that resolves to no asset is a 404, not an empty list."""
+    resp = client.get(f'{ALERTS_URL}?hostname=NOSUCHBAY',
+                      headers=_collector_headers(app))
+
+    assert resp.status_code == 404
+
+
+def test_deactivated_alert_drops_out_of_the_feed(client, db, app):
+    """Server-side close: the alert stops being offered to the endpoint.
+
+    This is how a cancel reaches a machine. The feed is authoritative, so an
+    alert missing from it is the agent's instruction to retire the pending
+    toast it is holding.
+    """
+    _computer(db, 'BAY-01')
+    alert = _fleet_alert(db)
+
+    alert.isactive = False
+    db.session.commit()
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01',
+                      headers=_collector_headers(app))
+
+    assert resp.status_code == 200
+    assert resp.get_json()['data'] == []
+
+
+def test_expired_alert_drops_out_of_the_feed(client, db, app):
+    """An alert past its endtime is withdrawn the same way a closed one is."""
+    _computer(db, 'BAY-01')
+    _fleet_alert(db, endtime=_naive_utcnow() - timedelta(minutes=5))
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01',
+                      headers=_collector_headers(app))
+
+    assert resp.status_code == 200
+    assert resp.get_json()['data'] == []
+
+
+def test_acknowledgement_removes_the_alert_for_that_asset(client, db, app):
+    """Acknowledgement is terminal: the alert never returns for this machine."""
+    asset = _computer(db, 'BAY-01')
+    alert = _fleet_alert(db)
+    headers = _collector_headers(app)
+
+    ack = client.post(ACK_URL.format(alert.notificationid),
+                      json={'hostname': 'BAY-01', 'assetid': asset.assetid},
+                      headers=headers)
+    assert ack.status_code == 201, ack.get_data(as_text=True)
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01', headers=headers)
+    assert resp.get_json()['data'] == []
+
+
+def test_acknowledgement_is_idempotent(client, db, app):
+    """The agent retries a queued acknowledgement, so a repeat must succeed.
+
+    Acknowledgement is terminal and the agent holds it on disk until the server
+    confirms, which means the same ack can legitimately arrive twice after a
+    network outage. The second one is a success, not a duplicate-key error.
+    """
+    from plugins.notifications.models import NotificationAcknowledgement
+
+    asset = _computer(db, 'BAY-01')
+    alert = _fleet_alert(db)
+    headers = _collector_headers(app)
+    payload = {'hostname': 'BAY-01', 'assetid': asset.assetid}
+
+    first = client.post(ACK_URL.format(alert.notificationid), json=payload,
+                        headers=headers)
+    second = client.post(ACK_URL.format(alert.notificationid), json=payload,
+                         headers=headers)
+
+    assert first.status_code == 201
+    assert second.status_code == 200, second.get_data(as_text=True)
+    assert second.get_json()['data']['acknowledged'] is True
+
+    rows = (NotificationAcknowledgement.query
+            .filter_by(notificationid=alert.notificationid,
+                       assetid=asset.assetid).all())
+    assert len(rows) == 1, 'a retried acknowledgement wrote a second row'
+
+
+def test_closing_an_acknowledged_alert_changes_nothing(client, db, app):
+    """A server-side close after acknowledgement is a no-op.
+
+    Acknowledgement already retired the alert for this asset, so there is
+    nothing left to withdraw and the alert must not reappear.
+    """
+    asset = _computer(db, 'BAY-01')
+    alert = _fleet_alert(db)
+    headers = _collector_headers(app)
+
+    client.post(ACK_URL.format(alert.notificationid),
+                json={'hostname': 'BAY-01', 'assetid': asset.assetid},
+                headers=headers)
+
+    alert.isactive = False
+    db.session.commit()
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01', headers=headers)
+    assert resp.status_code == 200
+    assert resp.get_json()['data'] == []
+
+
+def test_one_bay_cannot_acknowledge_for_another(client, db, app):
+    """The hostname must match the assetid it claims.
+
+    This guard regressed into an AttributeError and never actually ran, so it
+    is pinned: the collector key is shared across the fleet, and without this
+    check any endpoint could clear an alert on another machine's behalf.
+    """
+    _computer(db, 'BAY-01')
+    other = _computer(db, 'BAY-02')
+    alert = _fleet_alert(db)
+
+    resp = client.post(ACK_URL.format(alert.notificationid),
+                       json={'hostname': 'BAY-01', 'assetid': other.assetid},
+                       headers=_collector_headers(app))
+
+    assert resp.status_code == 400, resp.get_data(as_text=True)
+
+    # And BAY-02's alert is untouched by the attempt.
+    feed = client.get(f'{ALERTS_URL}?hostname=BAY-02',
+                      headers=_collector_headers(app))
+    assert [a['notificationId'] for a in feed.get_json()['data']] == [
+        alert.notificationid]
+
+
+def test_acknowledging_one_bay_leaves_the_others_alerted(client, db, app):
+    """A fleet alert is per-asset: one operator does not clear it for everyone."""
+    asset = _computer(db, 'BAY-01')
+    _computer(db, 'BAY-02')
+    alert = _fleet_alert(db)
+    headers = _collector_headers(app)
+
+    client.post(ACK_URL.format(alert.notificationid),
+                json={'hostname': 'BAY-01', 'assetid': asset.assetid},
+                headers=headers)
+
+    still = client.get(f'{ALERTS_URL}?hostname=BAY-02', headers=headers)
+    assert [a['notificationId'] for a in still.get_json()['data']] == [
+        alert.notificationid]
+
+
+def test_acknowledge_requires_the_collector_key(client, db, app):
+    """Acknowledgement is agent-facing and must not answer an anonymous caller."""
+    asset = _computer(db, 'BAY-01')
+    alert = _fleet_alert(db)
+
+    resp = client.post(ACK_URL.format(alert.notificationid),
+                       json={'hostname': 'BAY-01', 'assetid': asset.assetid})
+
+    assert resp.status_code in (401, 403), resp.get_data(as_text=True)
+
+
+def test_acknowledging_an_unknown_notification_is_not_found(client, db, app):
+    """A stale alert id from a rebuilt server is a 404, not a 500."""
+    asset = _computer(db, 'BAY-01')
+
+    resp = client.post(ACK_URL.format(999999),
+                       json={'hostname': 'BAY-01', 'assetid': asset.assetid},
+                       headers=_collector_headers(app))
+
+    assert resp.status_code == 404
+
+
+def test_non_fleet_notification_is_never_delivered(client, db, app):
+    """Ordinary board notifications are not agent alerts and must not leak."""
+    from plugins.notifications.models import Notification
+
+    _computer(db, 'BAY-01')
+    ntype = _alert_type(db)
+    db.session.add(Notification(notificationtypeid=ntype.notificationtypeid,
+                                notification='Board only', alertfleet=False,
+                                isactive=True))
+    db.session.commit()
+
+    resp = client.get(f'{ALERTS_URL}?hostname=BAY-01',
+                      headers=_collector_headers(app))
+
+    assert resp.status_code == 200
+    assert resp.get_json()['data'] == []