Stop repeating persistent failure alerts and digest routine ones

docs/consumer-deployment.md

@@ -92,6 +92,10 @@ Explicitly initialize delivery once with `python -m tools.alert_delivery --initi
 
 Each alert occurrence gets a durable sending intent and immutable payload before any network call. The Resend idempotency key is derived from its `deliveryId`; a returned provider message ID is persisted before polling. The message names the subject, occurrence, finding and repair action. Up to three least-recently-checked pending occurrences are processed per run, so bounced or ambiguous messages cannot starve later alerts. The provider transport uses fixed HTTPS endpoints, rejects redirects and bounds response bytes and total request time. Errors expose only fixed diagnostics, never API credentials or response bodies.
 
+Critical alerts (severity `error`, which includes every `monitor-failure`, and any `tls-regression`) are sent one occurrence per message as soon as they are queued. Other alerts wait until the oldest has been pending for an hour, then go out together as one digest of up to 20 alerts. A digest has its own durable record listing its member `deliveryId`s and its own idempotency key; its members stay pending until that one message is confirmed delivered, then they are acknowledged together. An unfinished digest keeps exactly its original members, so alerts that arrive meanwhile wait for the next one.
+
+A failure that persists across checks queues one occurrence. While it lasts, the same failure is queued again at most once a day as a reminder, instead of after every delivered occurrence. The next successful check forgets active failures, so a later recurrence alerts at once.
+
 API acceptance (`sent` or `queued`) is not delivery. The adapter retains the pending alert until the provider's matching message record reports `delivered`, `opened` or `clicked`; it durably records that result before acknowledging the exact occurrence. This is provider-reported delivery to the receiving mail system, not proof that a human read or acted on the alert. The provider ID and delivery event remain in the private journal for acceptance evidence. Bounces, suppression, failure and complaints remain pending and report `delivery-failed` for operator action. Unknown or inaccessible provider status cannot acknowledge an alert; the existing key must permit message retrieval as well as sending.
 
 Resend retains [idempotency keys for 24 hours](https://resend.com/changelog/idempotency-keys). An ambiguous submission can retry the same immutable request for at most 23 hours. After that it reports `reconciliation-required` and never blindly resends. Reconcile the occurrence tag/provider activity before any operator recovery; the tool deliberately cannot guess whether a message was sent when both acknowledgment and provider ID were lost. Delivered receipts can be retried locally after an outbox-write failure without sending another email. Retain receipt history across releases; reaching the bounded journal capacity requires reviewed archival, not silent deletion.

tests/test_alert_delivery.py

@@ -36,7 +36,7 @@ def setUp(self):
         self.config = self.root/'delivery.json'
         put(self.config, b'{"version":1,"recipient":"[email protected]","sender":"quantakrypto <[email protected]>"}')
         self.alert = {'id': 'test-alert', 'type': 'tls-regression', 'severity': 'error', 'action': 'Inspect the TLS deployment and restore approved hybrid key exchange.'}
-        self.pending = enqueue_failure(self.state/'state.json', self.alert)[0]
+        self.pending = enqueue_failure(self.state/'state.json', self.alert, now=NOW)[0]
         self.provider = Provider()
         from tools.alert_delivery import initialize_delivery
         initialize_delivery(self.state, NOW)
@@ -83,10 +83,11 @@ def test_receipt_persisted_before_ack_and_old_receipt_cannot_ack_recurrence(self
         self.deliver()
         self.assertEqual(self.pending_alerts(), [])
         self.assertEqual(len(self.provider.sent), 1)
-        new = enqueue_failure(self.state/'state.json', self.alert)[0]
+        # A persisting failure is reminded daily; the old receipt cannot acknowledge it.
+        new = enqueue_failure(self.state/'state.json', self.alert, now=NOW+dt.timedelta(hours=25))[0]
         self.assertNotEqual(new['deliveryId'], self.pending['deliveryId'])
         self.provider.event = 'sent'
-        self.deliver()
+        self.deliver(NOW+dt.timedelta(hours=25))
         self.assertEqual(self.pending_alerts()[0]['deliveryId'], new['deliveryId'])
         self.assertEqual(len(self.provider.sent), 2)
 
@@ -114,7 +115,7 @@ def test_no_network_without_api_key(self):
 
     def test_failed_first_batch_does_not_starve_later_occurrences(self):
         for number in range(3):
-            enqueue_failure(self.state/'state.json', dict(self.alert, id='other-'+str(number)))
+            enqueue_failure(self.state/'state.json', dict(self.alert, id='other-'+str(number)), now=NOW)
         self.provider.failure = True
         self.deliver()
         first = set(self.provider.sent)
@@ -130,5 +131,54 @@ def test_corrupted_delivery_clock_is_not_accepted(self):
         with self.assertRaises(ValueError): self.deliver()
         self.assertEqual(self.provider.sent, [])
 
+    def test_persistent_failure_is_one_occurrence_until_daily_reminder(self):
+        self.provider.event = 'delivered'
+        self.deliver()
+        self.assertEqual(self.pending_alerts(), [])
+        for minutes in (1, 2, 60, 23*60):
+            later = NOW+dt.timedelta(minutes=minutes)
+            self.assertEqual(enqueue_failure(self.state/'state.json', self.alert, now=later), [])
+            self.deliver(later)
+        self.assertEqual(len(self.provider.sent), 1)
+        other = enqueue_failure(self.state/'state.json', dict(self.alert, id='different-failure'), now=NOW+dt.timedelta(minutes=5))
+        self.assertEqual(other[0]['id'], 'different-failure')
+        reminder = enqueue_failure(self.state/'state.json', self.alert, now=NOW+dt.timedelta(hours=24))
+        self.assertIn(self.alert['id'], [a['id'] for a in reminder])
+
+    def routine(self, number, minutes=0):
+        alert = {'id': 'routine-'+str(number), 'type': 'new-finding', 'severity': 'warning', 'code': 'fallback-risk', 'ref': 'r'+str(number),
+                 'action': 'Review the referenced evidence and refresh or correct the manifest.'}
+        return enqueue_failure(self.state/'state.json', alert, now=NOW+dt.timedelta(minutes=minutes))[0]
+
+    def test_non_critical_alerts_wait_and_go_out_as_one_digest(self):
+        self.provider.event = 'delivered'
+        self.deliver()
+        self.assertEqual(len(self.provider.sent), 1)
+        first, second = self.routine(1), self.routine(2, 30)
+        self.deliver(NOW+dt.timedelta(minutes=59))
+        self.assertEqual(len(self.provider.sent), 1)
+        self.assertEqual(len(self.pending_alerts()), 2)
+        self.provider.event = 'sent'
+        results = self.deliver(NOW+dt.timedelta(minutes=61))
+        self.assertEqual(len(self.provider.sent), 2)
+        self.assertEqual(sorted(results[0]['members']), sorted([first['deliveryId'], second['deliveryId']]))
+        self.assertIn('digest (2 alerts)', self.provider.payload['subject'])
+        self.assertIn(first['deliveryId'], self.provider.payload['text'])
+        # A third alert arriving meanwhile does not change the unfinished digest.
+        self.routine(3, 62)
+        self.provider.event = 'delivered'
+        self.deliver(NOW+dt.timedelta(minutes=63))
+        self.assertEqual(len(self.provider.sent), 2)
+        self.assertEqual([a['id'] for a in self.pending_alerts()], ['routine-3'])
+
+    def test_critical_alert_is_not_held_for_the_digest(self):
+        self.provider.event = 'delivered'
+        self.deliver()
+        self.routine(1)
+        critical = enqueue_failure(self.state/'state.json', dict(self.alert, id='downgrade', severity='warning'), now=NOW)[0]
+        results = self.deliver(NOW+dt.timedelta(minutes=1))
+        self.assertEqual([r['deliveryId'] for r in results], [critical['deliveryId']])
+        self.assertEqual([a['id'] for a in self.pending_alerts()], ['routine-1'])
+
 
 if __name__ == '__main__': unittest.main()

tests/test_cross_runtime.py

@@ -266,8 +266,9 @@ def cycle(instant):
                 self.assertEqual(expired_manifest['assessment']['status'], 'invalid')
                 retained_baseline = json.loads(original)
                 after_failure = json.loads(remote_state.read_bytes())
-                self.assertEqual({k: v for k, v in after_failure.items() if k != 'pendingAlerts'},
-                                 {k: v for k, v in retained_baseline.items() if k != 'pendingAlerts'})
+                bookkeeping = ('pendingAlerts', 'activeFailures')
+                self.assertEqual({k: v for k, v in after_failure.items() if k not in bookkeeping},
+                                 {k: v for k, v in retained_baseline.items() if k not in bookkeeping})
                 self.assertTrue(any(a['type'] == 'monitor-failure' for a in after_failure['pendingAlerts']))
                 self.assertEqual(expired_manifest['changes']['alerts'][0]['type'], 'monitor-failure')
 

tools/alert_delivery.py

@@ -1,6 +1,7 @@
 """Durable Resend delivery of explicitly configured consumer alerts."""
 import argparse
 import datetime as dt
+import hashlib
 from email.utils import parseaddr
 import http.client
 import json
@@ -19,6 +20,12 @@
 DELIVERED = {'delivered', 'opened', 'clicked'}
 FAILED = {'bounced', 'canceled', 'complained', 'failed', 'suppressed'}
 IDEMPOTENCY_SECONDS = 23 * 3600  # Provider retains keys 24h; allow clock/network margin.
+# Critical alerts are sent one by one as soon as they are queued. The rest wait until
+# the oldest has been pending for DIGEST_SECONDS and then go out together in one digest.
+CRITICAL_TYPES = {'tls-regression'}
+DIGEST_SECONDS = 3600
+DIGEST_LIMIT = 20
+MESSAGE_LIMIT = 16384
 
 
 def _resend_worker(pipe, key, method, path, payload, idempotency_key):
@@ -101,19 +108,58 @@ def _configuration(path):
     return config
 
 
+def _critical(alert):
+    return alert.get('severity') == 'error' or alert.get('type') in CRITICAL_TYPES
+
+
+def _queued(alert):
+    # Alerts queued before digests existed carry no time and are treated as due.
+    try:
+        return dt.datetime.fromisoformat(alert['queuedAt']).timestamp()
+    except (KeyError, TypeError, ValueError):
+        return 0.0
+
+
 def _payload(alert, state, config):
     label = str(alert.get('type', 'monitor-failure'))
     if not re.fullmatch('[a-z0-9-]{1,80}', label):
         raise ValueError('invalid alert type')
     text = ('Crypto-agility monitoring alert\n\nSubject: '+str(state.get('subject') or 'unavailable')+
             '\nOccurrence: '+alert['deliveryId']+'\nAction: '+str(alert.get('action', 'Review the monitor findings.'))+
             '\n\nDetails:\n'+json.dumps(alert, sort_keys=True, indent=2)+'\n')
-    if len(text.encode()) > 16384:
+    if len(text.encode()) > MESSAGE_LIMIT:
         raise ValueError('alert message size limit')
     return {'from': config['sender'], 'to': [config['recipient']], 'subject': '[QuantaKrypto] Crypto-agility '+label,
             'text': text, 'tags': [{'name': 'monitor', 'value': 'crypto-agility'}, {'name': 'delivery', 'value': alert['deliveryId']}]}
 
 
+def _digest_section(alert):
+    label = str(alert.get('type', 'monitor-failure'))
+    if not re.fullmatch('[a-z0-9-]{1,80}', label):
+        raise ValueError('invalid alert type')
+    return ('\n['+label+'] Occurrence: '+alert['deliveryId']+'\nAction: '+str(alert.get('action', 'Review the monitor findings.'))+
+            '\nDetails:\n'+json.dumps(alert, sort_keys=True, indent=2)+'\n')
+
+
+def _digest(alerts, state, config):
+    """Group the oldest non-critical alerts that fit one message; returns (id, members, payload)."""
+    head = 'Crypto-agility monitoring digest\n\nSubject: '+str(state.get('subject') or 'unavailable')+'\n'
+    members, body = [], ''
+    for alert in sorted(alerts, key=lambda a: (_queued(a), a['deliveryId'])):
+        section = _digest_section(alert)
+        if len(members) >= DIGEST_LIMIT or len((head+body+section).encode()) > MESSAGE_LIMIT:
+            break
+        members.append(alert['deliveryId'])
+        body += section
+    if not members:
+        raise ValueError('alert message size limit')
+    identity = hashlib.sha256(('digest:'+','.join(sorted(members))).encode()).hexdigest()[:32]
+    payload = {'from': config['sender'], 'to': [config['recipient']],
+               'subject': '[QuantaKrypto] Crypto-agility digest ('+str(len(members))+' alerts)',
+               'text': head+body, 'tags': [{'name': 'monitor', 'value': 'crypto-agility'}, {'name': 'delivery', 'value': identity}]}
+    return identity, members, payload
+
+
 def deliver_pending(directory, config_path, client, *, now=None):
     directory = Path(directory)
     now = now or dt.datetime.now(dt.timezone.utc)
@@ -137,21 +183,38 @@ def deliver_pending(directory, config_path, client, *, now=None):
                     or any(type(record[name]) not in (int, float) or not math.isfinite(record[name]) or record[name] > now.timestamp()
                            for name in ('firstAttempt', 'lastChecked', 'deliveredAt') if name in record)):
                 raise ValueError('invalid delivery receipt timestamp')
-        pending = sorted(pending, key=lambda alert: journal['records'].get(alert.get('deliveryId'), {}).get('lastChecked', -1))[:3]
         for alert in pending:
-            identity = alert.get('deliveryId', '')
-            if not isinstance(identity, str) or not re.fullmatch('[a-f0-9]{32}', identity):
+            if not isinstance(alert.get('deliveryId'), str) or not re.fullmatch('[a-f0-9]{32}', alert['deliveryId']):
                 raise ValueError('invalid alert occurrence')
+        pending_ids = {alert['deliveryId'] for alert in pending}
+        urgent = sorted((a for a in pending if _critical(a)),
+                        key=lambda alert: journal['records'].get(alert['deliveryId'], {}).get('lastChecked', -1))[:3]
+        work = [(alert['deliveryId'], [alert['deliveryId']], lambda alert=alert: _payload(alert, state, config)) for alert in urgent]
+        routine = [a for a in pending if not _critical(a)]
+        # An unfinished digest keeps its exact members until its receipt resolves.
+        open_digest = next((name for name, record in journal['records'].items()
+                            if isinstance(record, dict) and record.get('members') and not record.get('deliveredAt')
+                            and record.get('lastEvent') not in FAILED and set(record['members']) <= pending_ids), None)
+        if open_digest:
+            work.append((open_digest, journal['records'][open_digest]['members'], None))
+        elif routine and now.timestamp() - min(_queued(a) for a in routine) >= DIGEST_SECONDS:
+            identity, members, payload = _digest(routine, state, config)
+            work.append((identity, members, lambda payload=payload: payload))
+        for identity, members, build in work:
             record = journal['records'].get(identity)
             if record is None:
                 if len(journal['records']) >= 10000:
                     raise ValueError('delivery receipt capacity requires archival')
-                record = {'firstAttempt': now.timestamp(), 'payload': _payload(alert, state, config)}
+                record = {'firstAttempt': now.timestamp(), 'payload': build()}
+                if len(members) > 1 or members[0] != identity:
+                    record['members'] = members
                 journal['records'][identity] = record
                 save(journal)  # Durable intent before any external request.
             record['lastChecked'] = now.timestamp()
             save(journal)
             result = {'deliveryId': identity, 'status': 'pending'}
+            if record.get('members'):
+                result['members'] = record['members']
             try:
                 payload = record['payload']
                 if payload['to'] != [config['recipient']] or payload['from'] != config['sender']:
@@ -180,7 +243,7 @@ def deliver_pending(directory, config_path, client, *, now=None):
                         record['deliveredAt'] = now.timestamp()
                     save(journal)
                 if record.get('deliveredAt'):
-                    acknowledge_alerts(directory/'state.json', [identity])
+                    acknowledge_alerts(directory/'state.json', record.get('members', [identity]))
                     result['status'] = 'delivered'
                 elif record.get('lastEvent') in FAILED:
                     result['status'] = 'delivery-failed'

tools/consumer.py

@@ -214,7 +214,7 @@ def check(self, now=None):
             try:
                 if not state_safe:
                     raise ValueError('consumer state directory is unsafe or unavailable')
-                changes['alerts'] = enqueue_failure(self.directory/'state.json', alert, require_existing=True)
+                changes['alerts'] = enqueue_failure(self.directory/'state.json', alert, require_existing=True, now=now)
             except (OSError, ValueError, TypeError, KeyError, RecursionError) as error:
                 changes['deliveryPersistenceFailure'] = str(error)
             return {'subject': None, 'checkedAt': now.isoformat(), 'assessment': {'status': 'invalid', 'findings': [finding]}, 'pairs': [], 'changes': changes}

tools/monitor.py

@@ -152,7 +152,7 @@ def read_input(name, path, limit=1048576):
         report["changes"] = {"alerts": [alert], "resolved": []}
         if state_path:
             try:
-                report["changes"]["alerts"] = enqueue_failure(state_path, alert, require_existing=require_existing_state)
+                report["changes"]["alerts"] = enqueue_failure(state_path, alert, require_existing=require_existing_state, now=now)
             except (OSError, ValueError, TypeError, KeyError, RecursionError) as exc:
                 # State failure cannot promise durability. Keep an explicit output
                 # signal for the supervisor rather than silently losing the failure.