fix(notifications): scope outcome improvements to backups

.github/scripts/tests/notification_recovery_fixture.py

@@ -1,86 +0,0 @@
-"""Pinned, inert recovery review. No operational module imports or threads.
-Run with python3 -S under bwrap --unshare-net; DBs and export use TMPDIR.
-"""
-import ast, contextlib, datetime, hashlib, io, json, os, pathlib, sqlite3, subprocess, sys, tarfile, tempfile, threading, types, typing
-from unittest.mock import patch
-REPO=pathlib.Path('/home/martino/projects/proxmox/notification-maintainer-followup')
-BASE=datetime.datetime(2026,9,30,20,0,0).timestamp()
-class Clock(datetime.datetime):
-    epoch=BASE
-    tick=0.0
-    @classmethod
-    def now(cls,tz=None):
-        value=cls.fromtimestamp(cls.epoch,tz)
-        cls.epoch+=cls.tick
-        return value
-TIME=types.SimpleNamespace(time=lambda:Clock.epoch)
-
-def extract(path,name,owner,ns):
-    tree=ast.parse(path.read_text())
-    nodes=tree.body if owner is None else next(n.body for n in tree.body if isinstance(n,ast.ClassDef) and n.name==owner)
-    node=next(n for n in nodes if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef)) and n.name==name)
-    node.decorator_list=[]
-    exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),ns)
-    return ns[name]
-
-from notification_fixture import SCRIPTS as scripts
-pns=dict(vars(typing),datetime=Clock,timedelta=datetime.timedelta,json=json,sqlite3=sqlite3,contextmanager=contextlib.contextmanager,re=__import__('re'),_re_disk_base=__import__('re'))
-pns['disk_base_name']=extract(scripts/'health_persistence.py','disk_base_name',None,pns)
-methods={n:extract(scripts/'health_persistence.py',n,'HealthPersistence',pns) for n in ('_get_conn','_db_connection','_init_database','record_error','_record_error_impl','resolve_error','_resolve_error_impl','get_recovery_evidence','_record_event','_entity_from_details','clear_error','get_active_errors','is_error_active','is_error_acknowledged','_get_setting_impl','get_setting','set_setting','acknowledge_error','_acknowledge_error_impl','get_excluded_interface_names')}
-def make_store(directory):
-    s=types.SimpleNamespace(db_path=pathlib.Path(directory)/'health.sqlite',_db_lock=threading.RLock(),DEFAULT_SUPPRESSION_HOURS=24,CATEGORY_SETTING_MAP={})
-    for n,f in methods.items():
-        if n=='_entity_from_details':setattr(s,n,f)
-        elif n=='_db_connection':setattr(s,n,types.MethodType(contextlib.contextmanager(f),s))
-        else:setattr(s,n,types.MethodType(f,s))
-    s._init_database()
-    return s
-def sql(s,query,args=()):
-    with s._db_connection() as c:
-        data=c.execute(query,args).fetchall();c.commit();return data
-def record(s,key='cpu_usage',category='cpu',reason='CPU high',details=None):
-    Clock.epoch=BASE-600
-    with patch.dict(sys.modules,{'os':types.SimpleNamespace(path=types.SimpleNamespace(exists=lambda p:True))}):s.record_error(key,category,'WARNING',reason,details)
-    Clock.epoch=BASE-10
-    with patch.dict(sys.modules,{'os':types.SimpleNamespace(path=types.SimpleNamespace(exists=lambda p:True))}):s.record_error(key,category,'WARNING',reason,details)
-    Clock.epoch=BASE
-    assert s.get_active_errors()
-    return sql(s,'SELECT first_seen FROM errors WHERE error_key=?',(key,))[0][0]
-def cpu(s,current=20,history=None,warning=85,critical=95):
-    ns=dict(vars(typing),time=TIME,os=types.SimpleNamespace(cpu_count=lambda:4),health_persistence=s,psutil=types.SimpleNamespace(cpu_percent=lambda **kw:current,cpu_count=lambda:4))
-    fn=extract(scripts/'health_monitor.py','_check_cpu_with_hysteresis','HealthMonitor',ns)
-    if history is None:history=[{'value':20,'time':BASE-i*10} for i in range(1,11)]
-    target=types.SimpleNamespace(state_history={'cpu_usage':list(history)},CPU_WARNING=85,CPU_CRITICAL=95,CPU_RECOVERY=75,CPU_WARNING_DURATION=300,CPU_CRITICAL_DURATION=300,CPU_RECOVERY_DURATION=120,_check_cpu_temperature=lambda:None)
-    refresh=extract(scripts/'health_monitor.py','_refresh_thresholds','HealthMonitor',{})
-    with patch.dict(sys.modules,{'health_thresholds':types.SimpleNamespace(get=lambda section,key:({'warning':warning,'critical':critical}.get(key) if section=='cpu' else None))}):refresh(target)
-    return fn(target)
-def poll(s,first,key='cpu_usage',category='cpu',reason='CPU high',details=None,first_done=True,foreign=False,restored=False):
-    events=[]
-    meta={'category':category,'reason':reason,'severity':'WARNING','first_seen':first,'details':details}
-    c=types.SimpleNamespace(_hostname='node-a',_ENTITY_MAP={'cpu':('node',''),'pve_services':('node',''),'network':('node','')},_first_poll_done=first_done,_known_errors={key:meta},_notified_severity={key:'WARNING'},_last_notified={key:BASE-1},SAME_ERROR_COOLDOWN=86400,_get_cooldown_from_db=lambda *a:BASE-1,_queue=types.SimpleNamespace(put=events.append),_guest_storage_error_is_now_foreign=lambda *a:foreign,_save_known_errors_meta=lambda:None)
-    ns=dict(vars(typing),time=TIME,json=json,re=__import__('re'),NotificationEvent=lambda *a,**kw:types.SimpleNamespace(event_type=a[0],severity=a[1],data=a[2],**kw),startup_grace=types.SimpleNamespace(should_suppress_category=lambda *a:False))
-    c._guest_storage_error_is_now_foreign=extract(scripts/'notification_events.py','_guest_storage_error_is_now_foreign','PollingCollector',ns)
-    fn=extract(scripts/'notification_events.py','_check_persistent_health','PollingCollector',ns)
-    with patch.dict(sys.modules,{'health_persistence':types.SimpleNamespace(health_persistence=s),'flask_server':types.SimpleNamespace(get_proxmox_node_name=lambda:'node-a',get_cached_pvesh_cluster_resources_vm=lambda:[{'vmid':100,'type':'lxc','node':'node-b' if foreign else 'node-a'}]),'datetime':types.SimpleNamespace(**{**vars(datetime),'datetime':Clock})}):
-        if restored:
-            c._KNOWN_ERRORS_SETTING_KEY='pollingcollector_known_errors_v1'
-            s.set_setting(c._KNOWN_ERRORS_SETTING_KEY,json.dumps(c._known_errors));c._known_errors={}
-            extract(scripts/'notification_events.py','_load_known_errors_meta','PollingCollector',ns)(c)
-            c._first_poll_done=bool(c._known_errors)
-        fn(c)
-    return [e.data for e in events],c
-def service(store, rc=0, stdout='active\n', raised=False, services=('pvedaemon',), clustered=False):
-    calls=[]
-    def run(argv, **kw):
-        calls.append((argv, kw))
-        assert argv[:2] == ['systemctl', 'is-active']
-        if raised: raise TimeoutError('inert timeout')
-        return types.SimpleNamespace(returncode=rc, stdout=stdout)
-    ns=dict(vars(typing),time=TIME,os=types.SimpleNamespace(path=types.SimpleNamespace(exists=lambda p:clustered)),subprocess=types.SimpleNamespace(run=run),health_persistence=store)
-    result=extract(scripts/'health_monitor.py','_check_pve_services','HealthMonitor',ns)(types.SimpleNamespace(PVE_SERVICES=list(services)))
-    return result,calls
-
-@contextlib.contextmanager
-def case():
-    Clock.epoch=BASE
-    with tempfile.TemporaryDirectory(prefix='recovery-review-db-',dir=os.environ.get('TMPDIR')) as d:yield make_store(d)

.github/scripts/tests/test_command_descriptions.py

@@ -138,14 +138,10 @@ class CommandDescriptionsTests(unittest.TestCase):
                         source = catalog('en')['runtime']['notifications']
                         for key, value in source['backup'].items():
                             local.setdefault('backup', {}).setdefault(key, value)
-                        local['channels']['email']['severity'].setdefault(
-                            'observation', source['channels']['email']['severity']['observation'])
                         local['channels']['email']['status'].setdefault(
                             'unconfirmed', source['channels']['email']['status']['unconfirmed'])
                         local['channels']['email']['status'].setdefault(
                             'completed_with_warnings', source['channels']['email']['status']['completed_with_warnings'])
-                        for key, value in source['healthRecovery'].items():
-                            local.setdefault('healthRecovery', {}).setdefault(key, value)
                     path.write_text(json.dumps(temporary, ensure_ascii=False))
                 # Model steady state after the bot fills these intentional new
                 # messages; keep repository locales and all other leaves intact.

.github/scripts/tests/test_notification_backup_split.py

@@ -0,0 +1,116 @@
+"""Backup-only maintainer contract at actual render/dispatch/email seams.
+
+Scope-approved seams: receiver, template lookup, rich enrichment, inert queued
+and manual delivery with an email capture sink. No host operations or sends.
+"""
+import unittest
+from notification_fixture import templates, LANGUAGES
+from notification_final_fixture import deliver
+
+
+class BackupSplitTests(unittest.TestCase):
+    def test_recovery_default_is_upstream_resolved_without_proof(self):
+        for language in LANGUAGES:
+            data = {'hostname': 'alias', 'category': 'temperature',
+                    'reason': 'Temperature high (recovered)', 'duration': '2m',
+                    'original_severity': 'WARNING'}
+            rendered = templates.render_template('error_resolved', data, language)
+            expected = templates.runtime_message('templates.error_resolved.title', language,
+                hostname='alias', category='temperature', entity_suffix='')
+            self.assertEqual(rendered['title'], expected)
+            rich, _ = templates.enrich_with_emojis('error_resolved', rendered['title'], rendered['body'], data)
+            self.assertTrue(rich.startswith('✅ '))
+            result = deliver('error_resolved', data, 'OK', language)
+            self.assertIn('background:#f0fdf4;', result['html'])
+
+    def test_pre_guest_native_subject_keeps_only_cause_once_with_display_alias(self):
+        from notification_fixture import receive
+        raw_host = 'pve-production.internal.example'
+        message = ('Details\n=======\nVMID    Name    Status    Time     Size     Filename\n'
+                   '\nTotal running time: 0s\nTotal size: 0 B')
+        for raw in (message, 'ERROR: unable to activate storage PBS\n' + message):
+            event = receive(raw, 'error', 'vzdump backup status (' + raw_host +
+                            '): backup failed: unable to activate storage PBS')
+            event.data['hostname'] = 'display-alias {rack.location}'
+            for language in LANGUAGES:
+                for manual in (False, True):
+                    result = deliver(event.event_type, event.data, event.severity, language, manual=manual)
+                    self.assertNotIn(raw_host, result['text'])
+                    self.assertNotIn('vzdump backup status', result['text'])
+                    self.assertEqual(result['text'].count('unable to activate storage PBS'), 1)
+                    self.assertIn('display-alias {rack.location}', result['title'])
+                    self.assertEqual(result['data']['pve_title'], event.data['pve_title'])
+                    self.assertEqual(result['data']['pve_message'], raw)
+        # No global hostname or guest-name substitution: actual diagnostics are
+        # authoritative, even when they happen to contain the original hostname.
+        event = receive(message, 'error', 'vzdump backup status (' + raw_host +
+                        '): backup failed: cannot connect to ' + raw_host)
+        result = deliver(event.event_type, {**event.data, 'hostname': 'alias'}, event.severity)
+        self.assertEqual(result['body'].count('cannot connect to ' + raw_host), 1)
+        self.assertNotIn('vzdump backup status', result['body'])
+
+    def test_backup_legacy_keys_keep_meaning_and_unknown_uses_new_keys(self):
+        # Frozen upstream contract; no Git history required in shipped tests.
+        legacy = {'en': {'title': '{hostname} → {storage}: Backup complete — {vmname} ({vmid})', 'body': 'Backup of {vmname} (ID: {vmid}) completed successfully on {storage}.\nSize: {size}', 'label': 'Backup complete'}, 'de': {'title': '{hostname} → {storage}: Sicherung abgeschlossen – {vmname} ({vmid})', 'body': 'Die Sicherung von {vmname} (ID: {vmid}) wurde am {storage} erfolgreich abgeschlossen.\nGröße: {size}', 'label': 'Sicherung abgeschlossen'}, 'es': {'title': '{hostname} → {storage}: Backup completado — {vmname} ({vmid})', 'body': 'El backup de {vmname} (ID: {vmid}) se ha completado correctamente en {storage}.\nTamaño: {size}', 'label': 'Backup completado'}, 'fr': {'title': '{hostname} → {storage}\xa0: Sauvegarde terminée — {vmname} ({vmid})', 'body': "La sauvegarde de {vmname} (ID\xa0: {vmid}) s'est terminée avec succès le {storage}.\nTaille\xa0: {size}", 'label': 'Sauvegarde terminée'}, 'it': {'title': '{hostname} → {storage}: Backup completato — {vmname} ({vmid})', 'body': 'Backup di {vmname} (ID: {vmid}) completato con successo su {storage}.\nTaglia: {size}', 'label': 'Backup completato'}, 'pt': {'title': '{hostname} → {storage}: Backup concluído — {vmname} ({vmid})', 'body': 'Backup de {vmname} (ID: {vmid}) concluído com sucesso em {storage}.\nTamanho: {size}', 'label': 'Backup concluído'}, 'sk': {'title': '{hostname} → {storage}: Záloha dokončená — {vmname} ({vmid})', 'body': 'Záloha {vmname} (ID: {vmid}) na úložisku {storage} bola úspešne dokončená.\nVeľkosť: {size}', 'label': 'Záloha bola dokončená'}, 'sv': {'title': '{hostname} → {storage}: Säkerhetskopiering klar — {vmname} ({vmid})', 'body': 'Säkerhetskopiering av {vmname} (ID: {vmid}) slutfördes framgångsrikt på {storage}.\nStorlek: {size}', 'label': 'Säkerhetskopieringen är klar'}}
+        for language, expected in legacy.items():
+            self.assertEqual(templates._load_runtime_catalog(language)['templates']['backup_complete'], expected)
+            result = templates.render_template('backup_complete', {'hostname': 'alias {rack.location}'}, language)
+            self.assertIn('alias {rack.location}', result['title'])
+            self.assertNotIn('()', result['title'])
+            new_title = templates.runtime_message('backup.unconfirmedTitle', language, hostname='alias {rack.location}')
+            self.assertTrue(new_title)
+            self.assertEqual(result['title'], new_title)
+            self.assertIn(templates.runtime_message('backup.unconfirmedBody', language), result['body'])
+        # Absent/blank/non-string translation uses English per-key fallback.
+        from unittest.mock import patch
+        english = templates._load_runtime_catalog('en')
+        for value in (None, '', {}, []):
+            missing = {'backup': {'unconfirmedTitle': value, 'unconfirmedBody': value}}
+            with patch.object(templates, '_load_runtime_catalog', side_effect=lambda lang: english if lang == 'en' else missing):
+                result = templates.render_template('backup_complete', {'hostname': 'alias'}, 'it')
+                self.assertEqual(result['title'], 'alias: Backup outcome unconfirmed')
+                self.assertEqual(result['body'], 'The backup outcome is not confirmed.')
+        # Both catalogs missing: new outcome text still has explicit EN defaults.
+        with patch.object(templates, '_load_runtime_catalog', return_value={}):
+            result = templates.render_template('backup_complete', {'hostname': 'alias'}, 'it')
+            self.assertEqual(result['title'], 'alias: Backup outcome unconfirmed')
+            self.assertEqual(result['body'], 'The backup outcome is not confirmed.')
+
+    def test_restore_keeps_original_ready_line_and_success_icon_with_warnings(self):
+        from notification_final_fixture import restore_event
+        ready = {'en': 'The node is now fully ready to use.', 'de': 'Der Knoten ist nun vollständig einsatzbereit.', 'es': 'El nodo está listo para usarse.', 'fr': 'Le nœud est maintenant entièrement prêt à être utilisé.', 'it': "Il nodo è ora completamente pronto per l'uso.", 'pt': 'O nó agora está totalmente pronto para uso.', 'sk': 'Uzol je teraz úplne pripravený na použitie.', 'sv': 'Noden är nu helt redo att användas.'}
+        for warning in ('', 'missing module zfs'):
+            event = restore_event(warning)
+            for language in LANGUAGES:
+                result = deliver(event['event_type'], event['data'], event['severity'], language)
+                self.assertTrue(result['title'].startswith('✅ '))
+                self.assertIn(ready[language], result['body'])
+                self.assertIn(ready[language], result['text'])
+                self.assertIn('2m', result['text'])
+                if warning:
+                    self.assertIn(warning, result['text'])
+                quiet = deliver(event['event_type'], event['data'], event['severity'], language, quiet=True)
+                self.assertIn('✅', quiet['body'])
+                self.assertIn('    ' + ready[language], quiet['body'])
+                self.assertIn('white-space:pre-wrap;', quiet['html'])
+
+    def test_spanish_outcome_and_restore_titles_are_capitalized_and_failure_is_exact(self):
+        expected = {'confirmed': 'Backup completado', 'completed_with_warnings': 'Backup completado con advertencias',
+                    'unconfirmed': 'Resultado del backup sin confirmar', 'failed': 'Backup fallido'}
+        for outcome, title in expected.items():
+            result = templates.render_template('backup_complete', {'hostname': 'alias', 'backup_outcome': outcome}, 'es')
+            self.assertEqual(result['title'], 'alias: ' + title)
+        failure = templates.render_template('backup_fail', {'hostname': 'alias'}, 'es')
+        self.assertEqual(failure['title'], 'alias: Backup fallido')
+        restore = templates.render_template('system_restore_completed', {'hostname': 'alias'}, 'es')
+        self.assertEqual(restore['title'], 'alias: Restauración del host finalizada')
+
+    def test_recovery_only_quiet_release_keeps_upstream_summary_contract(self):
+        result = deliver('error_resolved', {'hostname': 'alias', 'category': 'temperature',
+                         'reason': 'old (recovered)', 'duration': '2m'}, 'OK', quiet=True)
+        self.assertIn(templates.runtime_message('digest.lead', 'en', count=1).strip(), result['body'])
+        self.assertIn(templates.runtime_message('digest.footer', 'en'), result['body'])
+        self.assertIn('✅', result['body'])
+
+
+if __name__ == '__main__': unittest.main()

.github/scripts/tests/test_notification_corrections.py

@@ -54,8 +54,6 @@ class CorrectionTests(unittest.TestCase):
         probe.catalogs = {lang: copy.deepcopy(templates._load_runtime_catalog(lang)) for lang in LANGUAGES}
         parity(probe)  # shipped missing keys remain allowed
         probe.catalogs['sk']['backup'] = copy.deepcopy(probe.catalogs['en']['backup'])
-        for key in ('observation',):
-            probe.catalogs['sk']['channels']['email']['severity'][key] = probe.catalogs['en']['channels']['email']['severity'][key]
         probe.catalogs['sk']['channels']['email']['status']['unconfirmed'] = probe.catalogs['en']['channels']['email']['status']['unconfirmed']
         parity(probe)  # generation of exactly the pending keys is legal
         probe.catalogs['sk']['backup']['confirmedTitle'] = 'Missing hostname token'
@@ -63,59 +61,14 @@ class CorrectionTests(unittest.TestCase):
 
 
     def test_actual_neutral_style_is_not_success_green(self):
-        for event, severity, data in (('error_resolved', 'OK', {}),
-                                      ('backup_complete', 'INFO', {'backup_outcome': 'unconfirmed'})):
+        for event, severity, data in (('backup_complete', 'INFO', {'backup_outcome': 'unconfirmed'}),):
             result, markup = email(event, data, severity)
             self.assertIn('background:#f9fafb;', markup)
             self.assertNotIn('background:#f0fdf4;', markup)
 
 
-    def test_actual_manual_caller_carries_event_presentation_context(self):
-        from notification_fixture import extract, SCRIPTS, EmailChannel
-        from typing import Optional, Dict, Any
-        from threading import Lock
-        captured = []
-        channel = object.__new__(EmailChannel)
-        channel.subject_prefix = '[ProxMenux]'
-        class Sink:
-            def send(self, title, body, severity, data):
-                captured.append((data, channel._format_html(title, body, severity, data)))
-                return {'success': True}
-        ns = {'Optional': Optional, 'Dict': Dict, 'Any': Any, 'TEMPLATES': templates.TEMPLATES,
-              'resolve_notification_hostname': lambda host, config: host or 'node-a',
-              'render_template': templates.render_template, '_should_bypass_ai': lambda event: True}
-        send = extract(SCRIPTS / 'notification_manager.py', 'send_notification', 'NotificationManager', ns)
-        class Manager:
-            _channels = {'email': Sink()}
-            _config = {}
-            _lock = Lock()
-            def _notification_language(self): return 'en'
-            def is_event_enabled(self, event): return True
-            def _build_ai_config(self): return {}
-            def _record_history(self, *args): pass
-        data = {'category': 'temperature', 'reason': 'old', 'duration': '3d', 'original_severity': 'WARNING',
-                '_event_type': 'node_reconnect', '_group': 'cluster'}
-        result = send(Manager(), 'error_resolved', 'OK', '', '', data)
-        self.assertTrue(result['success'])
-        context, markup = captured[0]
-        self.assertEqual(context['_event_type'], 'error_resolved')
-        self.assertEqual(context['_group'], 'health')
-        self.assertIn('NO LONGER REPORTED', markup)
-        self.assertNotIn('>RESOLVED</span>', markup)
-        self.assertNotIn('color:#16a34a', markup)
-        self.assertEqual(data['_event_type'], 'node_reconnect')  # caller not mutated
 
 
-    def test_disappearance_body_keeps_observation_age_not_green_severity(self):
-        data = {'hostname': 'node-a', 'category': 'temperature', 'reason': 'old observation',
-                'duration': '3d 2h', 'original_severity': 'WARNING', 'severity': 'OK'}
-        for language in LANGUAGES:
-            result, markup = email('error_resolved', data, 'OK', language)
-            for line in result['body'].splitlines():
-                if line.strip(): self.assertIn(line.strip(), html.unescape(markup))
-            self.assertNotIn('>OK</span>', markup)
-            self.assertNotIn('color:#16a34a', markup)
-            self.assertNotIn('>RESOLVED</span>', markup)
 
 
     def test_actual_restore_endpoint_warnings_and_counts_reach_email(self):

.github/scripts/tests/test_notification_final_corrections.py

@@ -77,23 +77,13 @@ class FinalCorrectionsTests(unittest.TestCase):
             self.assertNotIn('script', result['tags'])
 
 
-    def test_long_disappearance_reason_is_present_once_in_actual_dispatch(self):
-        reason = 'Temperature exceeded configured limit; the source stopped reporting this observation after expiry.'
-        for language in LANGUAGES:
-            for manual in (False, True):
-                result = deliver('error_resolved', {'hostname':'node-a','category':'temperature',
-                    'reason':reason,'duration':'3d 2h','original_severity':'WARNING'}, 'OK', language, manual=manual)
-                self.assertEqual(result['text'].count(reason), 1)
-                self.assertNotIn('>OK</span>', result['html'])
-                self.assertNotIn('>RESOLVED</span>', result['html'])
 
 
-    def test_raw_restore_and_observation_cells_use_event_scoped_mail_wrapping(self):
+    def test_raw_restore_cells_use_event_scoped_mail_wrapping(self):
         token = 'b' * 64
         event = restore_event('Boot check: recorded token ' + token + '; verification pending')
         for language in LANGUAGES:
             results = [deliver(event['event_type'],event['data'],event['severity'],language,quiet=quiet) for quiet in (False,True)]
-            results.append(deliver('error_resolved',{'hostname':'node-a','reason':token,'category':'temperature','duration':'3d 2h'},'OK',language))
             for result in results:
                 self.assertIn('table-layout:fixed;', result['html'])
                 self.assertIn('word-wrap:break-word;', result['html'])

.github/scripts/tests/test_notification_maintainer_followup.py

@@ -61,7 +61,7 @@ class MaintainerFollowupTests(unittest.TestCase):
                         report+'\nINFO: Starting Backup of VM 100 (lxc)'):
             self.assertEqual(templates._parse_vzdump_message(message)['vms'][0]['type'],'')
 
-    def test_original_subject_only_retained_when_no_guest_context(self):
+    def test_native_subject_keeps_cause_without_host_envelope(self):
         subject = 'vzdump backup status (raw-host): backup failed: multiple problems'
         event = receive('ERROR: archive write failed\n'+NATIVE_REPORT,'error',subject)
         event.data['hostname']='configured-alias'
@@ -72,7 +72,8 @@ class MaintainerFollowupTests(unittest.TestCase):
             self.assertIn('configured-alias',result['title'])
         setup=receive('Details\n=======\nVMID    Name    Status    Time     Size     Filename\n\nTotal running time: 0s\nTotal size: 0 B','error',subject.replace('multiple problems','unable to open storage'))
         result=deliver(setup.event_type,setup.data,setup.severity)
-        self.assertEqual(result['text'].count(setup.data['pve_title']),1)
+        self.assertEqual(result['text'].count('unable to open storage'),1)
+        self.assertNotIn(setup.data['pve_title'],result['text'])
         unique=receive(NATIVE_REPORT,'error',subject.replace('multiple problems','job-end hook denied'))
         result=deliver(unique.event_type,unique.data,unique.severity)
         self.assertIn('job-end hook denied',result['body'])

.github/scripts/tests/test_notification_outcome_wording.py

@@ -15,17 +15,7 @@ from notification_fixture import templates as actual_templates
 ROOT = Path(__file__).resolve().parents[3]
 SCRIPTS = ROOT / 'AppImage/scripts'
 CATALOG = ROOT / 'AppImage/messages/en/common.json'
-EXPECTED = {
-    'error_resolved': {
-        'title': '{hostname}: No longer reported - {category}{entity_suffix}',
-        'body': 'The {category} issue is no longer in active health records.\n{reason}\n🚦 Previous severity: {original_severity}\n⏱️ Time since first observation: {duration}',
-        'label': 'Recovery notification',
-    },
-
-    'system_restore_completed': {
-        'body': 'Post-restore tasks completed in background.\n\nGuests applied: {guests}\nBind-mount stubs: {stubs}\nStale node dirs removed: {stale_nodes}\nComponents reinstalled: {components}\nDuration: {duration}\n{warnings_block}',
-    },
-}
+EXPECTED = {'system_restore_completed': {'body': 'Post-restore tasks completed in background.\n\nGuests applied: {guests}\nBind-mount stubs: {stubs}\nStale node dirs removed: {stale_nodes}\nComponents reinstalled: {components}\nDuration: {duration}\n{warnings_block}\nThe node is now fully ready to use.'}}
 
 
 def extract(path, name, owner=None, namespace=None):
@@ -82,7 +72,7 @@ class OutcomeWording(unittest.TestCase):
         result = module.render_template('system_restore_completed', {
             'hostname':'node-a','guests':3,'stubs':0,'stale_nodes':0,
             'components':1,'duration':'2m','warnings_block':''}, 'es')
-        self.assertIn('Configuraciones de guests aplicadas: 3', result['body'])
+        self.assertIn('Guests aplicados: 3', result['body'])
         self.assertNotIn('invitados', result['body'].lower())
 
     def test_settings_labels_stay_at_upstream_values_in_all_locales(self):
@@ -298,11 +288,6 @@ class OutcomeWording(unittest.TestCase):
                     _SEV_DEFAULT = EmailChannel._SEV_DEFAULT
                     subject_prefix = 'ProxMenux'
                     _build_detail_rows = staticmethod(build)
-                badge = catalog['channels']['email']['severity'].get('observation') or english['channels']['email']['severity']['observation']
-                recovery = fmt(Email(), 'No longer reported', 'Body', 'OK', {'_event_type': 'error_resolved',
-                    '_notification_language': lang, '_group': 'health'})
-                self.assertIn('>' + badge.upper() + '</span>', recovery)
-                self.assertIn('color:#6b7280;', recovery)
                 unrelated = fmt(Email(), 'Reconnected', 'Body', 'OK', {'_event_type': 'node_reconnect',
                     '_notification_language': lang, '_group': 'cluster'})
                 self.assertIn('>' + catalog['channels']['email']['severity']['ok'].upper() + '</span>', unrelated)
@@ -346,7 +331,7 @@ class OutcomeWording(unittest.TestCase):
                                           self.catalog['runtime']['notifications']['backup'][key]).format(hostname=data['hostname'])
                         self.assertTrue(result['title'].startswith(expected_title), result['title'])
                     else:
-                        self.assertEqual(result['title'], catalog['templates']['backup_complete']['title'].format_map(module._SafeFormatDict(data)))
+                        self.assertEqual(result['title'], module.runtime_message('backup.unconfirmedTitle', lang, hostname=data['hostname']))
                     self.assertNotIn('{hostname}', result['title'])
                     if state == 'unconfirmed':
                         source = (catalog if catalog.get('backup', {}).get('unconfirmedBody')
@@ -357,54 +342,12 @@ class OutcomeWording(unittest.TestCase):
                                       self.catalog['runtime']['notifications']['backup']['errorBody'], result['body'])
                     enriched, _ = module.enrich_with_emojis('backup_complete', result['title'], result['body'], data)
                     self.assertTrue(enriched.startswith({'confirmed':'💾✅','unconfirmed':'💾❔','failed':'💾❌'}[state]))
-            recovery = module.render_template('error_resolved', {'hostname':'node','category':'temperature',
-                'reason':'Old observation','duration':'3d','original_severity':'WARNING'}, lang)
-            recovery_source = catalog
-            self.assertEqual(recovery['title'], recovery_source['templates']['error_resolved']['title'].format(hostname='node',category='temperature',entity_suffix=''))
-            self.assertNotIn('resolved', recovery['title'].lower()) if lang == 'en' else None
             restore = module.render_template('system_restore_completed', {'hostname':'node', 'guests':4,
                 'stubs':1,'stale_nodes':2,'components':1,'duration':'2m','warnings_block':'Missing module'},lang)
             self.assertIn('Missing module',restore['body'])
-            self.assertNotIn('fully ready',restore['body'].lower())
+            if lang == 'en': self.assertIn('fully ready',restore['body'].lower())
 
-    def test_stale_record_disappearance_is_not_claimed_recovery(self):
-        data = {'hostname': 'node-a', 'category': 'temperature', 'reason': 'Temperature observation (no longer reported)',
-                'original_severity': 'WARNING', 'duration': '2d 0h', 'severity': 'OK'}
-        output = self.render('error_resolved', data, 'en')
-        self.assertIn('no longer in active health records', output['body'])
-        self.assertNotIn('resolved', (output['title'] + output['body']).lower())
-        self.assertIn('Time since first observation', output['body'])
 
-    def test_actual_poller_stale_disappearance_keeps_reason_factual(self):
-        class Store:
-            def get_active_errors(self): return []
-            def is_error_acknowledged(self, key): return False
-        class Event:
-            def __init__(self, *args, **kwargs): self.kind, self.severity, self.data = args[:3]
-        class Queue:
-            def __init__(self): self.items = []
-            def put(self, event): self.items.append(event)
-        ns = {'time': time, 'json': json, 'NotificationEvent': Event, 'Dict': dict}
-        poll = extract(SCRIPTS / 'notification_events.py', '_check_persistent_health', 'PollingCollector', ns)
-        class Collector:
-            _hostname = 'node-a'
-            _ENTITY_MAP = {'temperature': ('node', '')}
-            _first_poll_done = True
-            _known_errors = {'temp': {'category': 'temperature', 'reason': 'Temperature high',
-                                      'severity': 'WARNING', 'first_seen': '2026-09-25T00:00:00'}}
-            _notified_severity = {'temp': 'WARNING'}
-            _last_notified = {'temp': 1}
-            _queue = Queue()
-            def _guest_storage_error_is_now_foreign(self, *a): return False
-            def _save_known_errors_meta(self): pass
-        with patch.dict(sys.modules, {'health_persistence': types.SimpleNamespace(health_persistence=Store())}):
-            poll(Collector())
-        events = Collector._queue.items
-        self.assertEqual(len(events), 1)
-        self.assertEqual((events[0].kind, events[0].severity), ('error_resolved', 'OK'))
-        self.assertEqual(events[0].data['reason'], 'Temperature high (no longer reported)')
-        rendered = self.render(events[0].kind, events[0].data, 'en')
-        self.assertNotIn('recovered', rendered['body'].lower())
 
     def test_warning_and_clean_restore_keep_only_reported_outcome(self):
         for warnings in ('', '⚠️  Boot sanity: missing modules\n'):
@@ -423,24 +366,9 @@ class OutcomeWording(unittest.TestCase):
                 self.assertEqual(event['severity'], 'WARNING' if warnings else 'INFO')
                 result = self.render(event['event_type'], event['data'], 'en')
                 self.assertIn('Post-restore tasks completed', result['body'])
-                self.assertNotIn('fully ready', result['body'])
+                self.assertIn('fully ready', result['body'])
                 if warnings: self.assertIn('missing modules', result['body'])
 
-    def test_missing_key_fallback_and_synthetic_translation(self):
-        catalog = copy.deepcopy(self.catalog)
-        translated = copy.deepcopy(self.catalog)
-        for event, fields in EXPECTED.items():
-            for field in fields: translated['runtime']['notifications']['templates'][event].pop(field)
-        _, render = renderer(catalog, translated)
-        for event, fields in EXPECTED.items():
-            for field in fields:
-                if field not in ('title', 'body'):
-                    continue
-                self.assertEqual(render(event, {'category': 'disk'}, 'it')[field],
-                                 render(event, {'category': 'disk'}, 'en')[field])
-        translated['runtime']['notifications']['templates']['error_resolved']['title'] = 'Synthetic observation: {category}'
-        _, render = renderer(catalog, translated)
-        self.assertEqual(render('error_resolved', {'category': 'disk'}, 'it')['title'], 'Synthetic observation: disk')
 
     def test_rich_backup_icon_tracks_outcome_and_digest_default_is_neutral(self):
         tree = ast.parse((SCRIPTS / 'notification_templates.py').read_text())
@@ -465,8 +393,6 @@ class OutcomeWording(unittest.TestCase):
         tree = ast.parse((SCRIPTS / 'notification_templates.py').read_text())
         icon_map = ast.literal_eval(next(n.value for n in tree.body if isinstance(n, ast.Assign)
             and any(isinstance(t, ast.Name) and t.id == 'EVENT_EMOJI' for t in n.targets)))
-        for event in ('error_resolved', 'system_restore_completed'):
-            self.assertNotIn('✅', icon_map[event], event)
         self.assertNotIn('✅', icon_map['backup_complete'])  # buffered digest has no outcome metadata
 
 

.github/scripts/tests/test_notification_packaged_recovery.py

@@ -1,83 +0,0 @@
-"""Execute the build's literal backend copies, then a clean shipped-only runtime."""
-import os
-from pathlib import Path
-import re
-import subprocess
-import sys
-import tempfile
-import unittest
-
-ROOT = Path(__file__).resolve().parents[3]
-PROBE = r'''
-import ast, pathlib, sys, types, typing
-stage = pathlib.Path(sys.argv[1])
-sys.path.insert(0, str(stage))
-assert 'health_recovery' not in sys.modules
-import notification_templates as templates
-from notification_channels import EmailChannel
-import health_recovery
-for module in (templates, health_recovery, sys.modules['notification_channels']):
-    assert pathlib.Path(module.__file__).parent == stage
-assert not any('projects/proxmox' in p for p in sys.path)
-templates._get_hostname = lambda: 'node-a'
-channel = object.__new__(EmailChannel)
-channel.subject_prefix = '[ProxMenux]'
-neutral = {'hostname': 'node-a', 'category': 'cpu', 'reason': 'CPU high', '_event_type': 'error_resolved'}
-templates.render_template('error_resolved', neutral)
-templates.enrich_with_emojis('error_resolved', 'Observation', 'Body', neutral)
-channel._format_html('Observation', 'Body', 'OK', neutral)
-templates.render_template('node_reconnect', {'hostname': 'node-a'})
-now = 1000.0
-calls = []
-ns = dict(vars(typing), time=types.SimpleNamespace(time=lambda: now),
-    os=types.SimpleNamespace(cpu_count=lambda: 4),
-    psutil=types.SimpleNamespace(cpu_percent=lambda **kw: 20, cpu_count=lambda: 4),
-    health_persistence=types.SimpleNamespace(resolve_error=lambda *a, **kw: calls.append((a, kw))))
-tree = ast.parse((stage/'health_monitor.py').read_text())
-owner = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'HealthMonitor')
-node = next(n for n in owner.body if isinstance(n, ast.FunctionDef) and n.name == '_check_cpu_with_hysteresis')
-exec(compile(ast.Module(body=[node], type_ignores=[]), 'shipped_cpu', 'exec'), ns)
-monitor = types.SimpleNamespace(state_history={'cpu_usage': [{'value':20,'time':now-i*10} for i in range(1,11)]},
-    CPU_WARNING=85, CPU_CRITICAL=95, CPU_RECOVERY=75, CPU_WARNING_DURATION=300,
-    CPU_CRITICAL_DURATION=300, CPU_RECOVERY_DURATION=120, _check_cpu_temperature=lambda:None)
-assert ns['_check_cpu_with_hysteresis'](monitor)['status'] == 'OK'
-assert len(calls) == 1 and calls[0][1]['check_evidence']['value'] == 20
-proof = calls[0][1]['check_evidence']
-native = dict(neutral, error_key='cpu_usage', check_evidence=proof, is_recovery=True, recovery_outcome='resolved')
-# Actual body/icon/email consumers from the package, no source-module fixture rescue.
-import unittest.mock
-with unittest.mock.patch('health_recovery.time.time', return_value=now):
-    result = templates.render_template('error_resolved', native)
-    title, body = result['title'], result['body']
-    assert 'Resolved' in title and 'fresh health check' in body, (title, body, proof)
-    rich_title, rich_body = templates.enrich_with_emojis('error_resolved', title, body, native)
-    assert rich_title.startswith('✅')
-    assert 'background:#f0fdf4;' in channel._format_html(title, body, 'OK', native)
-print('shipped-only neutral body/icon/email + CPU native measurement/proof consumers PASS')
-'''
-
-
-class PackagedRecoveryTests(unittest.TestCase):
-    def test_actual_copy_manifest_supports_isolated_recovery_runtime(self):
-        source = ROOT / 'AppImage/scripts'
-        build = (source / 'build_appimage.sh').read_text()
-        lines = [line for line in build.splitlines()
-                 if re.match(r'^cp "\$SCRIPT_DIR/[^"/]+\.py" "\$APP_DIR/usr/bin/"', line)]
-        self.assertTrue(lines)
-        catalog_copy = re.search(r'^for locale in en de es fr it pt sk sv; do\n.*?^done$', build, re.MULTILINE | re.DOTALL)
-        if catalog_copy is None:
-            self.fail('The shipped locale-copy loop was not found in the actual build script')
-        lines.append(catalog_copy.group(0))
-        with tempfile.TemporaryDirectory(prefix='shipped-recovery-') as directory:
-            stage = Path(directory) / 'usr/bin'
-            stage.mkdir(parents=True)
-            copied = subprocess.run(['/bin/bash'], input='set -e\n'+'\n'.join(lines)+'\n', text=True,
-                capture_output=True, env={**os.environ, 'SCRIPT_DIR': str(source), 'APPIMAGE_ROOT': str(source.parent), 'APP_DIR': directory})
-            self.assertEqual(copied.returncode, 0, copied.stderr)
-            result = subprocess.run([sys.executable, '-I', '-B', '-c', PROBE, str(stage)],
-                cwd=directory, text=True, capture_output=True)
-            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
-
-
-if __name__ == '__main__':
-    unittest.main()

.github/scripts/tests/test_notification_pve92.py

@@ -103,9 +103,7 @@ class PVE92Tests(unittest.TestCase):
                 'warnings_block': ''}
         slovak = templates._load_runtime_catalog('sk')
         english = templates._load_runtime_catalog('en')
-        for event, field in (('error_resolved', 'title'), ('error_resolved', 'body'),
-                             ('system_restore_completed', 'body'),
-                             ('backup_complete', 'title'), ('backup_complete', 'body')):
+        for event, field in (('system_restore_completed', 'body'),):
             with self.subTest(event=event, field=field):
                 value = slovak['templates'][event][field]
                 result = templates.render_template(event, data, 'sk')

.github/scripts/tests/test_notification_recovery_corrections.py

@@ -1,225 +0,0 @@
-"""Native initializer, measurement methods, SQL writers/readers and collector."""
-import unittest
-from notification_recovery_fixture import case, Clock, BASE, cpu, sql, poll
-
-class RecoveryCorrectionTests(unittest.TestCase):
-    def test_supported_low_warning_current_violation_is_neutral(self):
-        with case() as store:
-            Clock.epoch = BASE - 600
-            initial = cpu(store, 80, [{'value':80, 'time':Clock.epoch-i*5} for i in range(1,26)], warning=50)
-            self.assertEqual(initial['status'], 'WARNING')
-            first = sql(store, 'SELECT first_seen FROM errors')[0][0]
-            Clock.epoch = BASE
-            result = cpu(store, 60, [{'value':60, 'time':BASE-i*5} for i in range(1,26)], warning=50)
-            events, _ = poll(store, first, reason=initial['reason'])
-            # Preserve operational clear/hysteresis behavior, not its factual claim.
-            self.assertEqual(result['status'], 'OK')
-            self.assertFalse(events[0]['is_recovery'], events)
-            self.assertIsNone(store.get_recovery_evidence('cpu_usage', first))
-
-    def test_original_policy_survives_repeated_native_updates(self):
-        import json
-        for later_warning in (85, 40):
-            with self.subTest(later_warning=later_warning), case() as store:
-                Clock.epoch = BASE - 600
-                cpu(store, 80, [{'value':80, 'time':Clock.epoch-i*5} for i in range(1,26)], warning=50)
-                first = sql(store, 'SELECT first_seen FROM errors')[0][0]
-                for step in (400, 200):
-                    Clock.epoch = BASE-step
-                    cpu(store, 90, [{'value':90, 'time':Clock.epoch-i*5} for i in range(1,26)], warning=later_warning)
-                original = json.loads(sql(store, 'SELECT details FROM errors')[0][0])['cpu_policy']
-                self.assertEqual(original['warning'], 50)
-                Clock.epoch = BASE
-                cpu(store, 20, warning=later_warning)
-                self.assertIsNone(store.get_recovery_evidence('cpu_usage', first))
-                self.assertFalse(poll(store, first)[0][0]['is_recovery'])
-
-    def test_clock_rollback_new_row_generic_clear_cannot_inherit_old_proof(self):
-        for rollback, reuse_first in ((True,False),(False,False),(True,True)):
-            with self.subTest(rollback=rollback,reuse_first=reuse_first),case() as store:
-                Clock.epoch = BASE-600
-                cpu(store, 90, [{'value':90, 'time':Clock.epoch-i*5} for i in range(1,26)])
-                first = sql(store, 'SELECT first_seen FROM errors')[0][0]
-                Clock.epoch = BASE; Clock.tick = .001
-                try: cpu(store)
-                finally: Clock.tick = 0
-                self.assertTrue(store.get_recovery_evidence('cpu_usage', first))
-                store.acknowledge_error('cpu_usage', suppression_hours=-1)
-                store.clear_error('cpu_usage')
-                Clock.epoch = BASE-100 if rollback else BASE+1
-                store.record_error('cpu_usage','cpu','WARNING','new incident after clock step',{})
-                second = sql(store, 'SELECT first_seen FROM errors')[0][0]
-                self.assertNotEqual(first, second)
-                if reuse_first:
-                    # Restored malformed snapshot with reused wall-clock identity;
-                    # native row id and latest closure still prevent replay.
-                    sql(store,'UPDATE errors SET first_seen=?',(first,));second=first
-                Clock.epoch = BASE+.0005 if rollback else BASE+2
-                store.clear_error('cpu_usage'); Clock.epoch = BASE+3
-                self.assertIsNone(store.get_recovery_evidence('cpu_usage', second))
-                self.assertFalse(poll(store, second)[0][0]['is_recovery'])
-
-    def test_malformed_native_and_manual_proof_is_neutral_at_all_consumers(self):
-        import copy, json, time
-        from unittest.mock import patch
-        from notification_fixture import LANGUAGES
-        from notification_final_fixture import deliver
-        with case() as store:
-            Clock.epoch = BASE-600
-            cpu(store, 90, [{'value':90,'time':Clock.epoch-i*5} for i in range(1,26)])
-            first = sql(store,'SELECT first_seen FROM errors')[0][0]
-            Clock.epoch = BASE; cpu(store)
-            event_data = json.loads(sql(store,"SELECT data FROM events WHERE event_type='resolved'")[0][0])
-            data = poll(store,first)[0][0]
-            for field, bad in [('checked_at','bad'),('checked_at',True),('checked_at',float('inf')),
-                               ('checked_at',10**400),('value',60),('max_sample',float('nan')),
-                               ('normal_samples',True),('normal_samples',9),('checked_at',BASE+1),('checked_at',BASE-7201),
-                               ('policy',{'warning':False,'critical':95,'recovery':75}),
-                               ('policy',{'warning':96,'critical':95,'recovery':75}),
-                               ('policy',{'warning':85,'critical':95}),
-                               ('policy',{'warning':85,'critical':95,'recovery':float('nan')}),
-                               ('policy',{'warning':85,'critical':95,'recovery':75,'extra':0})]:
-                broken = copy.deepcopy(event_data)
-                broken['check_evidence'][field] = bad
-                if field == 'value': broken['check_evidence']['policy']['warning'] = 50
-                sql(store,"UPDATE events SET data=? WHERE event_type='resolved'",(json.dumps(broken),))
-                with self.subTest(field=field,bad=str(bad)),patch('health_recovery.time.time',return_value=BASE):
-                    self.assertIsNone(store.get_recovery_evidence('cpu_usage',first))
-                    for language in LANGUAGES:
-                        for manual in (False,True):
-                            result = deliver('error_resolved',{**data,'check_evidence':broken['check_evidence']},'OK',language,manual=manual)
-                            self.assertNotIn('background:#f0fdf4;', result['html'])
-            # Caller content is trusted, not authenticated native proof; even
-            # well-shaped assertions require a valid time/type/numeric contract.
-            for proof in ({'check':'cpu_usage'}, {'check':'cpu_usage','checked_at':time.time()+1}):
-                self.assertNotIn('background:#f0fdf4;', deliver('error_resolved',{**data,'check_evidence':proof},'OK',manual=True)['html'])
-
-    def test_exact_service_active_native_clear_reaches_recovery_consumers(self):
-        from unittest.mock import patch
-        from notification_recovery_fixture import service
-        from notification_fixture import LANGUAGES
-        from notification_final_fixture import deliver
-        with case() as store:
-            Clock.epoch = BASE-600
-            service(store,3,'inactive\n')
-            first = sql(store,'SELECT first_seen FROM errors')[0][0]
-            Clock.epoch = BASE
-            result, calls = service(store)
-            self.assertEqual(result['status'],'OK')
-            self.assertEqual(calls,[(['systemctl','is-active','pvedaemon'], {'capture_output':True,'text':True,'timeout':2})])
-            proof = store.get_recovery_evidence('pve_service_pvedaemon',first)
-            self.assertTrue(proof)
-            data = poll(store,first,'pve_service_pvedaemon','pve_services','PVE service pvedaemon is inactive',details={'service':'pvedaemon'})[0][0]
-            self.assertTrue(data['is_recovery'])
-            with patch('health_recovery.time.time',return_value=BASE):
-                for language in LANGUAGES:
-                    for manual in (False,True):
-                        rendered = deliver('error_resolved',data,'OK',language,manual=manual)
-                        self.assertIn('background:#f0fdf4;',rendered['html'])
-                        self.assertIn('pvedaemon', rendered['text'])
-
-    def test_cpu_positive_default_low_policy_and_neutral_history_controls(self):
-        from notification_recovery_fixture import record
-        for warning,current,history,legacy,expected in (
-            (85,20,None,False,True), (50,20,None,False,True),
-            (85,20,None,True,False), (85,99,[],False,False),
-            (85,20,[],False,False),
-            (85,20,[{'value':20,'time':BASE+i*5} for i in range(1,10)],False,False),
-            (85,20,[{'value':20,'time':BASE-121-i} for i in range(12)],False,False),
-            (85,99,None,False,False),
-            (85,float('nan'),None,False,False), (85,float('inf'),None,False,False),
-            (85,True,None,False,False), (85,10**400,None,False,False)):
-            with self.subTest(warning=warning,current=str(current),legacy=legacy), case() as store:
-                if legacy: first = record(store)
-                else:
-                    Clock.epoch=BASE-600
-                    cpu(store,90,[{'value':90,'time':Clock.epoch-i*5} for i in range(1,26)],warning=warning)
-                    first=sql(store,'SELECT first_seen FROM errors')[0][0]
-                Clock.epoch=BASE; cpu(store,current,history,warning=warning)
-                self.assertEqual(bool(store.get_recovery_evidence('cpu_usage',first)),expected)
-
-    def test_service_unavailable_removed_overall_ok_and_ack_controls(self):
-        from notification_recovery_fixture import service, record
-        for rc,stdout,raised in ((3,'inactive\n',False),(4,'unknown\n',False),(0,'active extra\n',False),(1,'active\n',False),(0,'',True)):
-            with self.subTest(rc=rc,stdout=stdout,raised=raised),case() as store:
-                Clock.epoch=BASE-600; service(store,3,'inactive\n')
-                first=sql(store,'SELECT first_seen FROM errors')[0][0]
-                Clock.epoch=BASE; service(store,rc,stdout,raised)
-                self.assertIsNone(store.get_recovery_evidence('pve_service_pvedaemon',first))
-                self.assertFalse(poll(store,first,'pve_service_pvedaemon','pve_services')[0])
-        for removed in ('pvedaemon','corosync'):
-            with self.subTest(removed=removed),case() as store:
-                first=record(store,'pve_service_'+removed,'pve_services','service inactive',{'service':removed})
-                result,calls=service(store,services=(),clustered=False)
-                self.assertEqual(result['status'],'OK'); self.assertEqual(calls,[])
-                store.clear_error('pve_service_'+removed)
-                self.assertFalse(poll(store,first,'pve_service_'+removed,'pve_services')[0][0]['is_recovery'])
-        with case() as store:
-            first=record(store,'pve_service_corosync','pve_services','corosync inactive',{'service':'corosync'})
-            result,calls=service(store,services=('pvedaemon',),clustered=False)
-            self.assertEqual(result['status'],'OK')
-            self.assertEqual([c[0][-1] for c in calls],['pvedaemon'])
-            self.assertTrue(store.is_error_active('pve_service_corosync'))
-            self.assertIsNone(store.get_recovery_evidence('pve_service_corosync',first))
-        with case() as store:
-            first=record(store,'pve_service_pvedaemon','pve_services','service inactive')
-            store.acknowledge_error('pve_service_pvedaemon',suppression_hours=-1)
-            store.clear_error('pve_service_pvedaemon',check_evidence={'check':'pve_service_pvedaemon','checked_at':BASE,'service':'pvedaemon','state':'active','returncode':0})
-            self.assertEqual(sql(store,'SELECT id FROM errors'),[])
-            self.assertIsNone(store.get_recovery_evidence('pve_service_pvedaemon',first))
-
-    def test_native_binding_latest_closure_rollbacks_and_consistent_ack_read(self):
-        import contextlib, json, sqlite3
-        for mutation in ('row_id','first_seen','closure','last_seen','latest_clear','latest_resolve','ack','event_insert_failure','ack_before_join'):
-            with self.subTest(mutation=mutation),case() as store:
-                Clock.epoch=BASE-600
-                cpu(store,90,[{'value':90,'time':Clock.epoch-i*5} for i in range(1,26)])
-                first=sql(store,'SELECT first_seen FROM errors')[0][0]
-                Clock.epoch=BASE
-                if mutation=='event_insert_failure':
-                    sql(store,"CREATE TRIGGER fail_resolve BEFORE INSERT ON events WHEN NEW.event_type='resolved' BEGIN SELECT RAISE(ABORT,'fixture'); END")
-                    self.assertEqual(cpu(store)['status'],'UNKNOWN')
-                    self.assertIsNone(sql(store,'SELECT resolved_at FROM errors')[0][0])
-                    continue
-                cpu(store); self.assertTrue(store.get_recovery_evidence('cpu_usage',first))
-                if mutation in ('row_id','first_seen','closure'):
-                    data=json.loads(sql(store,"SELECT data FROM events WHERE event_type='resolved'")[0][0])
-                    field={'row_id':'id','first_seen':'first_seen','closure':'resolved_at'}[mutation]
-                    data['incident'][field]='wrong'
-                    sql(store,"UPDATE events SET data=? WHERE event_type='resolved'",(json.dumps(data),))
-                elif mutation=='last_seen': sql(store,'UPDATE errors SET last_seen=?',(Clock.fromtimestamp(BASE+1).isoformat(),))
-                elif mutation.startswith('latest_'):
-                    sql(store,"INSERT INTO events(event_type,error_key,timestamp,data) VALUES(?,'cpu_usage',?,'{}')",('cleared' if mutation=='latest_clear' else 'resolved',Clock.now().isoformat()))
-                elif mutation=='ack': store.acknowledge_error('cpu_usage',suppression_hours=-1)
-                else:
-                    original=store._db_connection; triggered=[]
-                    class Proxy:
-                        def __init__(self,connection):self.connection=connection
-                        def __getattr__(self,name):return getattr(self.connection,name)
-                        def execute(self,query,args=()):
-                            if 'FROM errors e JOIN events' in query and not triggered:
-                                triggered.append(True)
-                                store.acknowledge_error('cpu_usage',suppression_hours=-1)
-                            return self.connection.execute(query,args)
-                    @contextlib.contextmanager
-                    def interleaved(**kw):
-                        with original(**kw) as conn: yield Proxy(conn)
-                    store._db_connection=interleaved
-                self.assertIsNone(store.get_recovery_evidence('cpu_usage',first))
-                if mutation=='ack_before_join': self.assertTrue(triggered)
-        with case() as store:
-            sql(store,"INSERT INTO errors(error_key,category,severity,reason,first_seen,last_seen) VALUES('cpu_usage','cpu','WARNING','fixture','x','x')")
-            with self.assertRaises(sqlite3.IntegrityError):
-                sql(store,"INSERT INTO errors(error_key,category,severity,reason,first_seen,last_seen) VALUES('cpu_usage','cpu','WARNING','duplicate','x','x')")
-
-    def test_malformed_history_declines_proof_without_changing_operational_clear(self):
-        with case() as store:
-            Clock.epoch=BASE-600
-            cpu(store,90,[{'value':90,'time':Clock.epoch-i*5} for i in range(1,26)])
-            first=sql(store,'SELECT first_seen FROM errors')[0][0]
-            Clock.epoch=BASE
-            result=cpu(store,20,[{'value':10**400,'time':BASE-i*5} for i in range(1,10)])
-            self.assertEqual(result['status'],'OK')
-            self.assertIsNone(store.get_recovery_evidence('cpu_usage',first))
-
-if __name__ == '__main__': unittest.main()

.github/scripts/tests/test_notification_recovery_evidence.py

@@ -1,75 +0,0 @@
-"""Fresh existing-check provenance; native initializer and disposable SQLite."""
-import json
-import time
-import unittest
-from unittest.mock import patch
-from notification_fixture import templates, LANGUAGES
-from notification_final_fixture import deliver
-from notification_recovery_fixture import case, Clock, BASE, cpu, sql, poll
-
-
-def original_cpu(store):
-    Clock.epoch = BASE-600
-    result = cpu(store, 90, [{'value':90,'time':Clock.epoch-i*5} for i in range(1,26)])
-    assert result['status'] == 'WARNING'
-    Clock.epoch = BASE
-    return sql(store, 'SELECT first_seen FROM errors')[0][0]
-
-
-class RecoveryEvidenceTests(unittest.TestCase):
-    def test_cpu_success_provenance_is_persisted_only_after_normal_samples(self):
-        with case() as store:
-            first = original_cpu(store)
-            self.assertEqual(cpu(store)['status'], 'OK')
-            proof = store.get_recovery_evidence('cpu_usage', first)
-            self.assertTrue(proof)
-            self.assertEqual(proof['check'], 'cpu_usage')
-            self.assertEqual(proof['checked_at'], BASE)
-            # A generic closure never gains proof; use another actual native row.
-            store.record_error('pve_service_test','pve_services','CRITICAL','inactive')
-            store.resolve_error('pve_service_test','No longer present')
-            self.assertFalse(json.loads(sql(store,"SELECT data FROM events ORDER BY id DESC LIMIT 1")[0][0]).get('check_evidence'))
-
-    def test_recovery_query_requires_fresh_same_incident_proof(self):
-        with case() as store:
-            first = original_cpu(store); cpu(store)
-            proof = store.get_recovery_evidence('cpu_usage',first)
-            self.assertTrue(proof)
-            self.assertIsNone(store.get_recovery_evidence('cpu_usage','different incident'))
-            saved = sql(store,'SELECT last_seen,resolved_at FROM errors')[0]
-            for field,value in [('acknowledged',1),('resolved_at',None),('last_seen',Clock.fromtimestamp(BASE+1).isoformat())]:
-                sql(store,f'UPDATE errors SET {field}=?',(value,))
-                self.assertIsNone(store.get_recovery_evidence('cpu_usage',first))
-                sql(store,'UPDATE errors SET acknowledged=0,last_seen=?,resolved_at=?',saved)
-            data = json.loads(sql(store,"SELECT data FROM events WHERE event_type='resolved'")[0][0])
-            for bad in (None,{'check':'cpu_usage','checked_at':BASE-7201}, {'check':'storage_removed','checked_at':BASE}, {'check':'cpu_usage','checked_at':float('inf')}, {'check':'cpu_usage','checked_at':10**400}):
-                sql(store,"UPDATE events SET data=? WHERE event_type='resolved'",(json.dumps({**data,'check_evidence':bad}),))
-                self.assertIsNone(store.get_recovery_evidence('cpu_usage',first))
-
-    def test_poller_and_all_consumers_distinguish_proven_recovery_from_disappearance(self):
-        for proved in (False,True):
-            with case() as store:
-                first = original_cpu(store)
-                if proved: cpu(store)
-                else: store.resolve_error('cpu_usage','No longer present')
-                data = poll(store,first,reason='CPU high')[0][0]
-                self.assertEqual(data['is_recovery'],proved)
-                self.assertEqual(data['recovery_outcome'],'resolved' if proved else 'no_longer_reported')
-                with patch('health_recovery.time.time',return_value=BASE):
-                    for lang in LANGUAGES:
-                        for manual in (False,True):
-                            result = deliver('error_resolved',data,'OK',lang,manual=manual)
-                            self.assertEqual('background:#f0fdf4;' in result['html'],proved)
-                            if proved:
-                                self.assertIn(templates.runtime_message('healthRecovery.title',lang,hostname='node-a',category='cpu',entity_suffix=''),result['title'])
-                            self.assertEqual(result['text'].count(data['reason']),1)
-
-    def test_manual_recovery_flag_alone_is_not_authoritative_evidence(self):
-        data={'hostname':'node-a','category':'cpu','reason':'Observation disappeared','duration':'1h','original_severity':'WARNING','recovery_outcome':'resolved'}
-        for lang in LANGUAGES:
-            for manual in (False,True):
-                result=deliver('error_resolved',data,'OK',lang,manual=manual)
-                self.assertNotIn('background:#f0fdf4;',result['html'])
-                self.assertNotIn(templates.runtime_message('healthRecovery.body',lang,**data),result['body'])
-
-if __name__=='__main__':unittest.main()

.github/scripts/tests/test_notification_recovery_order.py

@@ -1,170 +0,0 @@
-"""Durable native observation order, not wall-clock row identity, admits proof."""
-import datetime
-import json
-import re
-import sys
-import types
-import typing
-import unittest
-from unittest.mock import patch
-from notification_recovery_fixture import case, Clock, BASE, cpu, service, sql, extract, scripts, TIME
-from notification_fixture import LANGUAGES
-from notification_final_fixture import deliver
-
-
-def collector(store):
-    events = []
-    target = types.SimpleNamespace(_hostname='alias {rack.location}',
-        _ENTITY_MAP={'cpu':('node',''), 'pve_services':('node','')}, _first_poll_done=False,
-        _known_errors={}, _notified_severity={}, _last_notified={}, SAME_ERROR_COOLDOWN=86400,
-        _get_cooldown_from_db=lambda *a:BASE-1, _queue=types.SimpleNamespace(put=events.append),
-        _save_known_errors_meta=lambda:None)
-    ns = dict(vars(typing), time=TIME, json=json, re=re,
-        NotificationEvent=lambda *a, **kw:types.SimpleNamespace(event_type=a[0], severity=a[1], data=a[2], **kw),
-        startup_grace=types.SimpleNamespace(should_suppress_category=lambda *a:False))
-    target._guest_storage_error_is_now_foreign = extract(scripts/'notification_events.py',
-        '_guest_storage_error_is_now_foreign', 'PollingCollector', ns)
-    poll = extract(scripts/'notification_events.py', '_check_persistent_health', 'PollingCollector', ns)
-    def tick():
-        with patch.dict(sys.modules, {'health_persistence':types.SimpleNamespace(health_persistence=store),
-                'datetime':types.SimpleNamespace(**{**vars(datetime), 'datetime':Clock})}):
-            poll(target)
-    return target, events, tick
-
-
-def abnormal(store, kind, value=99):
-    if kind == 'cpu':
-        return cpu(store, value, [{'value':value, 'time':Clock.epoch-i*5} for i in range(1,26)])
-    return service(store,3,'inactive\n')[0]
-
-
-def normal(store, kind):
-    return cpu(store) if kind == 'cpu' else service(store)[0]
-
-
-def initial_closure(store, kind):
-    key = 'cpu_usage' if kind == 'cpu' else 'pve_service_pvedaemon'
-    Clock.epoch = BASE-600
-    abnormal(store,kind,90)
-    target, events, tick = collector(store)
-    tick()
-    assert target._first_poll_done and key in target._known_errors and not events
-    snapshot = json.loads(json.dumps(target._known_errors))
-    Clock.epoch = BASE
-    assert normal(store,kind)['status'] == 'OK'
-    first = snapshot[key]['first_seen']
-    assert store.get_recovery_evidence(key,first)
-    return key, first, snapshot, target, events, tick
-
-
-class RecoveryOrderTests(unittest.TestCase):
-    def assert_consumers(self, data, expected):
-        with patch('health_recovery.time.time', return_value=Clock.epoch):
-            for language in LANGUAGES:
-                for manual in (False,True):
-                    with self.subTest(language=language, manual=manual):
-                        result = deliver('error_resolved',data,'OK',language,manual=manual)
-                        self.assertEqual('background:#f0fdf4;' in result['html'],expected)
-                        self.assertIn('alias {rack.location}',result['text'])
-                quiet = deliver('error_resolved',data,'OK',language,quiet=True)
-                self.assertEqual(len(quiet['buffered']),1)
-                # The existing quiet digest stores the rendered title, not
-                # health body/proof metadata. Assert its exact outcome label.
-                from notification_fixture import templates
-                label = templates.render_template('error_resolved',data,language)['title'].split(': ',1)[-1]
-                self.assertIn(label, quiet['body'])
-                self.assertIn(label, quiet['buffered'][0][2])
-
-    def assert_superseded(self, kind):
-        for offset in (-100,0,1):
-            for value in ((90,99) if kind == 'cpu' else (99,)):
-                with self.subTest(kind=kind, offset=offset, value=value), case() as store:
-                    key, first, snapshot, target, events, tick = initial_closure(store,kind)
-                    oldrow = sql(store,'SELECT id,first_seen,resolved_at FROM errors')[0]
-                    prior = sql(store,'SELECT id,event_type FROM events ORDER BY id')
-                    Clock.epoch = BASE+offset
-                    renewed = abnormal(store,kind,value)
-                    self.assertEqual(renewed['status'],'WARNING' if kind == 'cpu' and value == 90 else 'CRITICAL')
-                    self.assertEqual(sql(store,'SELECT id,first_seen,resolved_at FROM errors')[0],oldrow)
-                    later = sql(store,'SELECT id,event_type FROM events ORDER BY id')[-1]
-                    self.assertGreater(later[0],prior[-1][0])
-                    self.assertEqual(later[1],'escalated' if kind == 'cpu' and value == 99 else 'updated')
-                    if kind == 'cpu':
-                        policy=json.loads(sql(store,'SELECT details FROM errors')[0][0])['cpu_policy']
-                        self.assertEqual(policy,{'warning':85,'critical':95,'recovery':75})
-                    Clock.epoch = BASE+2
-                    tick()
-                    self.assertEqual(len(events),1)
-                    data = events[0].data
-                    self.assertFalse(data['is_recovery'])
-                    self.assertIsNone(store.get_recovery_evidence(key,first))
-                    self.assert_consumers(data,False)
-                    # Existing operations do not rearm the resolved row, so a
-                    # later normal check cannot establish a NEW native closure.
-                    before = sql(store,'SELECT id FROM events ORDER BY id')
-                    Clock.epoch = BASE+3
-                    self.assertEqual(normal(store,kind)['status'],'OK')
-                    self.assertEqual(sql(store,'SELECT id FROM events ORDER BY id'),before)
-                    self.assertIsNone(store.get_recovery_evidence(key,first))
-
-    def test_cpu_superseded_same_row_is_neutral_at_all_consumers(self):
-        self.assert_superseded('cpu')
-
-    def test_service_superseded_same_row_is_neutral_at_all_consumers(self):
-        self.assert_superseded('service')
-
-    def test_fresh_closure_and_repeated_noop_clear_keep_genuine_proof(self):
-        for kind in ('cpu','service'):
-            with self.subTest(kind=kind),case() as store:
-                key, first, snapshot, target, events, tick = initial_closure(store,kind)
-                before = sql(store,'SELECT id,event_type FROM events ORDER BY id')
-                Clock.epoch = BASE+1
-                for _ in range(3):
-                    store.clear_error(key)
-                    store.resolve_error(key,'generic repeat')
-                    normal(store,kind)
-                self.assertEqual(sql(store,'SELECT id,event_type FROM events ORDER BY id'),before)
-                self.assertTrue(store.get_recovery_evidence(key,first))
-                tick()
-                self.assertEqual(len(events),1)
-                self.assertTrue(events[0].data['is_recovery'])
-                self.assert_consumers(events[0].data,True)
-                tick()
-                self.assertEqual(len(events),1)
-
-    def test_later_actual_generic_closure_blocks_older_proof_even_clock_rollback(self):
-        for kind in ('cpu','service'):
-            with self.subTest(kind=kind),case() as store:
-                key, first, *_ = initial_closure(store,kind)
-                Clock.epoch = BASE-100
-                with store._db_connection() as conn:
-                    store._record_event(conn.cursor(),'cleared',key,{'reason':'generic actual closure','check_evidence':None})
-                    conn.commit()
-                self.assertIsNone(store.get_recovery_evidence(key,first))
-
-    def test_explicit_acknowledged_row_suppresses_native_proof(self):
-        for kind in ('cpu','service'):
-            with self.subTest(kind=kind),case() as store:
-                key, first, snapshot, target, events, tick = initial_closure(store,kind)
-                store.acknowledge_error(key,suppression_hours=-1)
-                self.assertEqual(sql(store,'SELECT acknowledged FROM errors')[0][0],1)
-                self.assertIsNone(store.get_recovery_evidence(key,first))
-                tick()
-                self.assertEqual(events,[])
-
-    def test_new_incarnation_abnormal_order_blocks_previous_closure(self):
-        for kind in ('cpu','service'):
-            with self.subTest(kind=kind),case() as store:
-                key, first, *_ = initial_closure(store,kind)
-                old_id = sql(store,'SELECT id FROM errors')[0][0]
-                store.acknowledge_error(key,suppression_hours=-1)
-                store.clear_error(key)
-                Clock.epoch = BASE-100
-                abnormal(store,kind,99)
-                self.assertGreater(sql(store,'SELECT id FROM errors')[0][0],old_id)
-                self.assertEqual(sql(store,'SELECT event_type FROM events ORDER BY id DESC')[0][0],'new')
-                self.assertIsNone(store.get_recovery_evidence(key,first))
-
-
-if __name__ == '__main__':
-    unittest.main()