fix(notifications): retain diagnostic context across final consumers

.github/scripts/tests/notification_final_fixture.py

@@ -0,0 +1,88 @@
+"""Inert infrastructure for actual endpoint/manual/queued/SQLite release seams."""
+import datetime
+import sqlite3
+import tempfile
+import threading
+import time
+import types
+import typing
+from html.parser import HTMLParser
+from pathlib import Path
+from notification_fixture import templates, SCRIPTS, EmailChannel, extract
+
+
+def visible(markup):
+    class Text(HTMLParser):
+        def __init__(self): super().__init__(); self.parts = []; self.tags = []
+        def handle_data(self, data): self.parts.append(data)
+        def handle_starttag(self, tag, attrs): self.tags.append(tag)
+    parser = Text(); parser.feed(markup)
+    return '\n'.join(p.strip() for p in parser.parts if p.strip()), parser.tags
+
+
+def restore_event(warnings):
+    events = []
+    ns = {'request': types.SimpleNamespace(remote_addr='127.0.0.1', get_json=lambda **kw: {
+        'hostname':'node-a', 'guests':'3', 'stubs':'1', 'stale_nodes':'0',
+        'components':'2', 'duration':'2m', 'warnings':warnings}),
+        'notification_manager':types.SimpleNamespace(emit_event=lambda **kw:events.append(kw)),
+        'jsonify':lambda value:value}
+    handler = extract(SCRIPTS/'flask_notification_routes.py', 'internal_restore_event', None, ns)
+    response, status = handler()
+    assert status == 200 and len(events) == 1
+    return events[0]
+
+
+def deliver(event_type, data, severity='INFO', language='en', manual=False, quiet=False, quiet_before=()):
+    """Execute real dispatch/manual and optional real SQLite buffer+flush."""
+    captured = []
+    channel = object.__new__(EmailChannel); channel.subject_prefix = '[ProxMenux]'
+    def sink(title, body, severity, data):
+        markup = channel._format_html(title, body, severity, data)
+        text, tags = visible(markup)
+        captured.append(dict(title=title, body=body, severity=severity, data=dict(data), html=markup, text=text, tags=tags))
+        return {'success':True}
+    manager = types.SimpleNamespace(_config={'email.rich_format':'true'}, _lock=threading.RLock(),
+        _channels={'email':types.SimpleNamespace(send=sink)},
+        _group_limiter=types.SimpleNamespace(allow=lambda group:True),
+        _claim_delivery=lambda event:'inert', _finish_delivery_claim=lambda *a,**kw:None,
+        _notification_language=lambda:language, _build_ai_config=lambda:{'ai_enabled':'false'},
+        _in_quiet_hours=lambda channel:quiet, _should_buffer_for_digest=lambda *a:False,
+        _record_history=lambda *a:None, _stats={'total_sent':0,'total_errors':0},
+        is_event_enabled=lambda event:True)
+    ns = dict(vars(typing), NotificationEvent=types.SimpleNamespace, TEMPLATES=templates.TEMPLATES, render_template=templates.render_template,
+        resolve_notification_hostname=lambda host, config:host or 'node-a',
+        enrich_with_emojis=templates.enrich_with_emojis, datetime=datetime.datetime,
+        _should_bypass_ai=lambda event:True, _AI_BYPASS_EVENTS=frozenset({'backup_complete','backup_fail'}))
+    for name in ('_dispatch_to_channels', '_dispatch_event', 'send_notification'):
+        setattr(manager, name, types.MethodType(extract(SCRIPTS/'notification_manager.py', name, 'NotificationManager', ns), manager))
+    with tempfile.TemporaryDirectory(prefix='notification-final-') as scratch:
+        db = Path(scratch)/'pending.sqlite'
+        rows = []
+        if quiet:
+            conn = sqlite3.connect(db)
+            conn.execute('CREATE TABLE quiet_pending (id INTEGER PRIMARY KEY, channel TEXT, event_type TEXT, event_group TEXT, severity TEXT, ts INTEGER, title TEXT, body TEXT)')
+            conn.commit(); conn.close()
+            qns = dict(vars(typing), sqlite3=sqlite3, DB_PATH=db, time=time, datetime=datetime.datetime,
+                _resolve_display_hostname=lambda config:'node-a', runtime_message=templates.runtime_message,
+                EVENT_EMOJI=templates.EVENT_EMOJI, CATEGORY_EMOJI=templates.CATEGORY_EMOJI)
+            for name in ('_buffer_quiet_event', '_flush_quiet_for_channel', '_compose_digest_body'):
+                setattr(manager,name,types.MethodType(extract(SCRIPTS/'notification_manager.py',name,'NotificationManager',qns),manager))
+            for earlier in quiet_before:
+                manager._dispatch_event(types.SimpleNamespace(**earlier,source='inert',entity_type='node',entity_id='',event_id='inert',fingerprint='inert'))
+        if manual:
+            assert not quiet
+            result = manager.send_notification(event_type, severity, '', '', dict(data))
+            assert result['success']
+        else:
+            event = types.SimpleNamespace(event_type=event_type, severity=severity, data=dict(data), source='inert', entity_type='node', entity_id='', event_id='inert', fingerprint='inert')
+            manager._dispatch_event(event)
+        if quiet:
+            conn = sqlite3.connect(db); rows = conn.execute('SELECT event_type,severity,title,body FROM quiet_pending').fetchall(); conn.close()
+            assert len(rows)==1+len(quiet_before) and not captured
+            manager._flush_quiet_for_channel('email', manager._channels['email'])
+            conn=sqlite3.connect(db); remaining=conn.execute('SELECT count(*) FROM quiet_pending').fetchone()[0]; conn.close()
+            assert remaining==0
+        assert len(captured)==1
+        if quiet: captured[0]['buffered']=rows
+        return captured[0]

.github/scripts/tests/test_notification_final_corrections.py

@@ -0,0 +1,146 @@
+"""Final review contracts at actual locale/manual/queued/quiet email seams.
+All operational dependencies are inert; no manager or route module import.
+"""
+import ast
+import copy
+import html
+import unittest
+from unittest.mock import patch
+from notification_fixture import templates, SCRIPTS, LANGUAGES, receive
+from notification_final_fixture import deliver, restore_event
+
+# Literal native send_notification body, produced by pinned PVE Perl helpers.
+NATIVE_REPORT = 'Details\n=======\n' + '''VMID    Name    Status    Time     Size     Filename
+100     web     ok        1m 1s    1 GiB    vm/100/2026-09-29T17:00:00Z
+
+Total running time: 1m 1s
+Total size: 1 GiB
+
+Logs
+====
+vzdump --all 1 --storage PBS --mode snapshot
+
+100: no log available
+'''
+
+
+
+class FinalCorrectionsTests(unittest.TestCase):
+    def test_runtime_backup_assertions_accept_generated_and_missing_slovak_title(self):
+        path = SCRIPTS / 'tests/test_notification_runtime_i18n.py'
+        tree = ast.parse(path.read_text())
+        owner = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'RuntimeCatalogTests')
+        method = next(n for n in owner.body if isinstance(n, ast.FunctionDef) and n.name == 'test_special_formatters_digest_and_test_message_are_slovak')
+        start = next(i for i,n in enumerate(method.body) if isinstance(n,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='backup' for t in n.targets))
+        stop = next(i for i,n in enumerate(method.body) if isinstance(n,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='manager' for t in n.targets))
+        block = ast.Module(body=method.body[start:stop], type_ignores=[])
+        english = templates._load_runtime_catalog('en')
+        for title in (None, '{hostname}: GENERATED_SK potvrdené'):
+            sk = copy.deepcopy(templates._load_runtime_catalog('sk'))
+            if title is not None: sk.setdefault('backup', {})['confirmedTitle'] = title
+            else: sk.get('backup', {}).pop('confirmedTitle', None)
+            with patch.object(templates, '_load_runtime_catalog', side_effect=lambda lang: english if lang == 'en' else sk):
+                exec(compile(block,str(path),'exec'), {'self':self, 'notification_templates':templates})
+
+
+    def test_native_multiline_failure_diagnostics_survive_receiver_dispatch_once(self):
+        subject = 'vzdump backup status (node-a): backup failed: multiple problems'
+        for diagnostic, report in (
+            ('external provider job cleanup failed\njob-abort hook permission denied', NATIVE_REPORT),
+            ('job interrupted\njob-abort hook permission denied', NATIVE_REPORT.replace('ok        ', 'todo      ')),
+            ('unable to initialize external provider\njob-abort hook permission denied', NATIVE_REPORT.replace('100     web     ok        1m 1s    1 GiB    vm/100/2026-09-29T17:00:00Z\n', '')),
+        ):
+            event = receive(diagnostic + '\n' + report, 'error', subject)
+            self.assertEqual((event.event_type, event.severity), ('backup_fail', 'CRITICAL'))
+            for language in LANGUAGES:
+                result = deliver(event.event_type, event.data, event.severity, language)
+                for line in diagnostic.splitlines():
+                    self.assertEqual(result['body'].count(line), 1)
+                    self.assertEqual(result['text'].count(line), 1)
+                if 'ok        ' in report:
+                    self.assertIn('✅ VM web (100)', result['body'])
+
+
+    def test_restore_endpoint_quiet_release_keeps_warning_counts_and_truthful_footer(self):
+        reason = 'missing module zfs <raw> & {literal}'
+        event = restore_event(reason)
+        self.assertEqual(event['severity'], 'WARNING')
+        for language in LANGUAGES:
+            result = deliver(event['event_type'], event['data'], event['severity'], language, quiet=True)
+            self.assertEqual(result['severity'], 'INFO')
+            buffered_body = result['buffered'][0][3]
+            self.assertEqual(result['text'].count(reason), 1)
+            for line in buffered_body.splitlines():
+                if line.strip(): self.assertIn(line.strip(), result['text'])
+            self.assertIn('2m', result['text'])
+            self.assertNotIn(templates.runtime_message('digest.footer', language), result['body'])
+            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):
+        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'])
+                self.assertIn('overflow-wrap:break-word;', result['html'])
+                self.assertEqual(result['text'].count(token), 1)
+        unrelated = deliver('node_reconnect', {'hostname':'node-a'}, 'OK')
+        self.assertNotIn('table-layout:fixed;', unrelated['html'])
+
+
+    def test_manual_failed_and_unconfirmed_backup_keep_short_actionable_reason(self):
+        for outcome, reason in (('failed','PBS permission denied for datastore remote'),
+                                ('unconfirmed','Task status unavailable: upstream API timed out')):
+            for language in LANGUAGES:
+                result = deliver('backup_complete',{'hostname':'node-a','vmid':'100','vmname':'web',
+                    'storage':'PBS','backup_outcome':outcome,'reason':reason},'INFO',language,manual=True)
+                self.assertEqual(result['text'].count(reason), 1)
+                self.assertNotIn('>COMPLETED</span>', result['html'])
+
+
+    def test_backup_reason_threshold_and_raw_body_deduplication(self):
+        for event in ('backup_complete', 'backup_fail'):
+            for length in (79, 80, 81, 120):
+                prefix = '<raw> & {rack.location} '
+                reason = prefix + 'b' * (length - len(prefix))
+                self.assertEqual(len(reason), length)
+                for raw in ('', reason):
+                    data = {'hostname':'node-a','backup_outcome':'failed','reason':reason,'pve_message':raw}
+                    for manual in (False, True):
+                        result = deliver(event,data,'INFO','en',manual=manual)
+                        self.assertEqual(result['text'].count(reason), 1, (event,length,raw,manual))
+                        self.assertNotIn('raw', result['tags'])
+
+
+    def test_quiet_restore_after_preview_limit_is_not_omitted_or_called_info(self):
+        event = restore_event('missing module zfs')
+        earlier = [dict(event_type='service_fail',severity='WARNING',data={'hostname':'node-a','service_name':f'unit-{i}','reason':'recorded'}) for i in range(8)]
+        result = deliver(event['event_type'],event['data'],event['severity'],quiet=True,quiet_before=earlier)
+        self.assertEqual(result['text'].count('missing module zfs'), 1)
+        self.assertNotIn(templates.runtime_message('digest.lead','en',count=9), result['body'])
+        self.assertEqual(result['data']['_count'], 9)
+
+
+    def test_native_error_block_does_not_deduplicate_a_substring_of_inventory(self):
+        event = receive('web\nexternal provider job cleanup failed\n' + NATIVE_REPORT,
+                        'error','vzdump backup status (node-a): backup failed: multiple problems')
+        result = deliver(event.event_type,event.data,event.severity)
+        self.assertEqual(result['body'].splitlines().count('web'), 1)
+
+
+if __name__ == '__main__': unittest.main()

AppImage/scripts/notification_channels.py

@@ -1053,11 +1053,13 @@ class EmailChannel(NotificationChannel):
                 status = 'unconfirmed'
             sev['label'] = _runtime_text(f'email.status.{status}', data)
         group = data.get('_group', 'other')
-        # Scoped inline mail-compatible wrapping: temperature measurements
-        # and backup identities/raw diagnostics. Other events retain layout.
+        # Scoped inline mail-compatible wrapping for authoritative raw-context
+        # bodies, including restore bodies released from quiet hours.
         backup_email = event_type in {'backup_complete', 'backup_fail'}
-        temp_cell_wrap = 'word-wrap:break-word;overflow-wrap:break-word;word-break:break-word;' if event_type == 'temp_high' or backup_email else ''
-        temp_table_layout = 'table-layout:fixed;' if event_type == 'temp_high' or backup_email else ''
+        wrap_body = (event_type in {'temp_high', 'system_restore_completed', 'error_resolved'}
+                     or backup_email or data.get('_restore_summary'))
+        temp_cell_wrap = 'word-wrap:break-word;overflow-wrap:break-word;word-break:break-word;' if wrap_body else ''
+        temp_table_layout = 'table-layout:fixed;' if wrap_body else ''
         backup_title_wrap = temp_cell_wrap if backup_email else ''
         backup_metadata_layout = 'table-layout:fixed;' if backup_email else ''
         section_label = _runtime_text(f'email.groups.{group}', data)
@@ -1090,8 +1092,14 @@ class EmailChannel(NotificationChannel):
                 ('', html_mod.escape(line.strip()))
                 for line in body.split('\n') if line.strip()
             )
+            # A metadata-only/manual body may be generic. Keep actionable raw
+            # context once, without restoring duplicated inventory metadata.
+            reason = data.get('reason', '')
+            if reason and len(reason) <= 80 and reason not in body:
+                detail_rows.append((html_mod.escape(_runtime_text('email.fields.reason', data)),
+                                    html_mod.escape(reason)))
 
-        if event_type in {'system_restore_completed', 'error_resolved'}:
+        if event_type in {'system_restore_completed', 'error_resolved'} or data.get('_restore_summary'):
             # Observation age/disappearance must not become a green OK row.
             # The endpoint's warnings_block and task counts live in the
             # localized body, not the generic services Event row.
@@ -1129,7 +1137,8 @@ class EmailChannel(NotificationChannel):
         # ── Reason / details block (long text, displayed separately) ──
         reason = data.get('reason', '')
         reason_html = ''
-        if reason and len(reason) > 80 and not (event_type == 'temp_high' and reason in body):
+        if reason and len(reason) > 80 and not (
+                (event_type in {'temp_high', 'error_resolved'} or backup_email) and reason in body):
             reason_html = f'''
 <div style="margin:16px 0 0;padding:12px 16px;border:1px solid #d1d5db;border-radius:6px;">
   <p style="margin:0 0 4px;font-size:11px;font-weight:600;color:#374151;text-transform:uppercase;letter-spacing:0.05em;">{_runtime_text('email.details', data)}</p>

AppImage/scripts/notification_manager.py

@@ -1821,7 +1821,8 @@ class NotificationManager:
             print(f"[NotificationManager] digest cleanup failed for "
                   f"{ch_name}: {e}")
 
-    def _compose_digest_body(self, rows: list, use_icons: bool = False) -> str:
+    def _compose_digest_body(self, rows: list, use_icons: bool = False,
+                             quiet_release: bool = False) -> str:
         """Render a grouped summary body. rows is a list of
         (id, event_type, event_group, ts, title, body) tuples ordered
         by timestamp ASC.
@@ -1830,16 +1831,23 @@ class NotificationManager:
         groups: OrderedDict[str, list] = OrderedDict()
         for _id, ev_type, group, ts, title, body in rows:
             label = group or 'other'
-            groups.setdefault(label, []).append((ts, ev_type, title))
+            groups.setdefault(label, []).append((ts, ev_type, title, body))
 
         language = self._notification_language()
-        lines = [runtime_message('digest.lead', language, count=len(rows))]
+        # The quiet summary title already carries the total; the daily lead
+        # incorrectly calls every buffered WARNING an INFO event.
+        lines = [] if quiet_release else [runtime_message('digest.lead', language, count=len(rows))]
         for group, items in groups.items():
             group_label = runtime_message(f'digest.groups.{group}', language) or group.title()
             group_icon = CATEGORY_EMOJI.get(group, '') if use_icons else ''
             group_prefix = f'{group_icon} ' if group_icon else ''
             lines.append(f"{group_prefix}{group_label}: {len(items)}")
-            for ts, ev_type, title in items[:8]:
+            # Quiet hours can buffer restore warnings, unlike the daily INFO
+            # digest. Keep their complete recorded body, even past the usual
+            # title preview limit, without changing either delivery policy.
+            visible_items = [item for index, item in enumerate(items)
+                             if index < 8 or (quiet_release and item[1] == 'system_restore_completed')]
+            for ts, ev_type, title, body in visible_items:
                 hhmm = datetime.fromtimestamp(ts).strftime('%H:%M')
                 short_title = title.split(': ', 1)[-1] if ': ' in title else title
                 event_icon = (
@@ -1847,10 +1855,15 @@ class NotificationManager:
                 ) if use_icons else ''
                 event_prefix = f'{event_icon} ' if event_icon else ''
                 lines.append(f"  • {event_prefix}{hhmm}  {short_title}")
-            if len(items) > 8:
-                lines.append(runtime_message('digest.more', language, count=len(items) - 8))
+                if quiet_release and ev_type == 'system_restore_completed' and body:
+                    lines.extend(line for line in body.splitlines() if line.strip())
+            if len(items) > len(visible_items):
+                lines.append(runtime_message('digest.more', language, count=len(items) - len(visible_items)))
             lines.append('')
-        lines.append(runtime_message('digest.footer', language))
+        # The daily footer describes live warning delivery, which is not true
+        # for warnings buffered during quiet hours. Do not repeat that claim.
+        if not quiet_release:
+            lines.append(runtime_message('digest.footer', language))
         return '\n'.join(lines).rstrip() + '\n'
 
     # ─── Quiet Hours buffer + flush ────────────────────────────
@@ -1985,13 +1998,14 @@ class NotificationManager:
             'digest.quietTitle', language, hostname=host, count=len(rows),
         )
         use_icons = self._config.get(f'{ch_name}.rich_format', 'false') == 'true'
-        summary_body = self._compose_digest_body(rows, use_icons=use_icons)
+        summary_body = self._compose_digest_body(rows, use_icons=use_icons, quiet_release=True)
 
         result: dict = {'success': False, 'error': ''}
         try:
             result = channel.send(
                 summary_title, summary_body, severity='INFO',
                 data={'_quiet_hours_summary': True, '_count': len(rows),
+                      '_restore_summary': any(row[1] == 'system_restore_completed' for row in rows),
                       '_notification_language': language},
             ) or result
         except Exception as e:

AppImage/scripts/notification_templates.py

@@ -2090,8 +2090,19 @@ def render_template(event_type: str, data: Dict[str, Any],
             diagnostic_lines = [line.strip() for line in pve_message.splitlines()
                                 if re.match(r'^\s*(?:\d+:\s*)?(?:\d{4}-\d{2}-\d{2}\s+\S+\s+)?(?:WARN(?:ING)?:|ERROR:|TASK ERROR)',
                                             line, re.IGNORECASE)]
+            if event_type == 'backup_fail' or data.get('backup_outcome') == 'failed':
+                # Native send_notification puts multiline job/setup errors
+                # before Details, while its subject says only "multiple problems".
+                # Keep that raw block when inventory replaces the producer body;
+                # it is job context, not evidence that every guest failed.
+                error_block = re.match(r'\A(.*?)^Details\r?\n=+\s*$',
+                                       pve_message, re.MULTILINE | re.DOTALL)
+                if error_block:
+                    diagnostic_lines = error_block.group(1).rstrip('\r\n').splitlines() + diagnostic_lines
             if diagnostic_lines:
-                body_text += '\n' + '\n'.join(dict.fromkeys(diagnostic_lines))
+                body_text += '\n' + '\n'.join(
+                    line for line in dict.fromkeys(diagnostic_lines)
+                    if line.strip() and line not in body_text.splitlines())
         else:
             # Couldn't parse -- use PVE raw message as body
             body_text = pve_message.strip()

AppImage/scripts/tests/test_notification_runtime_i18n.py

@@ -341,7 +341,10 @@ class RuntimeCatalogTests(unittest.TestCase):
             },
             language="sk",
         )
-        self.assertIn("Backup complete", backup["title"])
+        expected_title = notification_templates.runtime_message(
+            "backup.confirmedTitle", "sk", hostname="pve01",
+        )
+        self.assertTrue(backup["title"].startswith(expected_title + " — "))
         self.assertIn("pbs-main", backup["title"])
         self.assertIn("VM alpha (100)", backup["title"])
         self.assertNotIn("Backup job finished", backup["title"])