fix(notifications): preserve VM and CT identities in AI diagnoses

.github/scripts/tests/notification_final_fixture.py

@@ -1,4 +1,5 @@
 """Inert infrastructure for actual endpoint/manual/queued/SQLite release seams."""
+import ast
 import datetime
 import sqlite3
 import tempfile
@@ -36,6 +37,10 @@ def restore_event(warnings):
 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 = []
+    delivery_errors = []
+    def record_history(event_type, channel, title, body, severity, success, error, source):
+        if not success:
+            delivery_errors.append(f'{channel}: {error}')
     channel = object.__new__(EmailChannel); channel.subject_prefix = '[ProxMenux]'
     def sink(title, body, severity, data):
         markup = channel._format_html(title, body, severity, data)
@@ -48,12 +53,25 @@ def deliver(event_type, data, severity='INFO', language='en', manual=False, quie
         _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},
+        _record_history=record_history, _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'}))
+        format_backup_diagnostics_with_ai=templates.format_backup_diagnostics_with_ai,
+        format_with_ai_full=templates.format_with_ai_full,
+        # AI is disabled in this fixture. Execute the actual formatting helpers
+        # synchronously, without a worker pool, providers or operational imports.
+        _format_with_ai_bounded=lambda function,*args,**kwargs:function(*args,**kwargs),
+        enrich_context_for_ai=lambda **kwargs:'')
+    manager_path = SCRIPTS/'notification_manager.py'
+    for node in ast.parse(manager_path.read_text()).body:
+        if isinstance(node, ast.Assign):
+            for target in node.targets:
+                if isinstance(target, ast.Name) and target.id in ('_AI_BYPASS_EVENTS', '_AI_NO_SUGGESTION_EVENTS'):
+                    ns[target.id] = frozenset(ast.literal_eval(node.value.args[0]))
+    for name in ('_should_bypass_ai', '_format_backup_with_ai_bounded'):
+        extract(manager_path, name, None, ns)
     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:
@@ -73,10 +91,11 @@ def deliver(event_type, data, severity='INFO', language='en', manual=False, quie
         if manual:
             assert not quiet
             result = manager.send_notification(event_type, severity, '', '', dict(data))
-            assert result['success']
+            assert result['success'], result.get('errors')
         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)
+        assert not delivery_errors, '; '.join(delivery_errors)
         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

AppImage/scripts/notification_templates.py

@@ -2308,6 +2308,12 @@ def render_template(event_type: str, data: Dict[str, Any],
             'diagnostics': '\n'.join(dict.fromkeys(line for line in analysis_lines if line.strip())),
             'failed_ids': [str(guest['vmid']) for guest in (parsed_backup or {}).get('vms', [])
                            if str(guest.get('status', '')).lower() == 'error'],
+            'guest_types': {
+                str(guest['vmid']): 'VM' if guest['type'] == 'qemu' else 'CT'
+                for guest in (parsed_backup or {}).get('vms', [])
+                if str(guest.get('status', '')).lower() == 'error'
+                and guest.get('type') in ('qemu', 'lxc')
+            },
         }
 
     if backup_diagnostics:
@@ -3390,6 +3396,8 @@ and one brief, actionable check when supported by the evidence. {emoji_rule}
 Do not reproduce shell commands, credentials, timestamps, raw logs, or a backup inventory.
 Never invent successes, failed guest IDs, causes, available space or quota amounts.
 Keep the scope of each error: a cause recorded for one guest is not proof for other guests.
+Use the authoritative Guest identities from the context, with VM/CT labels exactly
+as given. Proxmox's generic "Backup of VM" log wording can also refer to a CT.
 "Disk quota exceeded" means a quota prevented a write, not proof that the Proxmox
 host disk is full. If the error explicitly concerns a PBS connection/write, explain
 the destination quota and suggest checking that account/datastore quota.
@@ -3464,6 +3472,11 @@ def format_backup_diagnostics_with_ai(title: str, body: str, severity: str,
     context = '\n'.join(lines)
     if len(context) > 12000:
         context = context[:5999] + '\n…\n' + context[-5998:]
+    guest_types = {str(vmid): kind for vmid, kind in analysis.get('guest_types', {}).items()
+                   if kind in ('VM', 'CT') and re.fullmatch(r'\d+', str(vmid))}
+    if guest_types:
+        context = 'Guest identities: ' + ', '.join(
+            f'{guest_types[vmid]} {vmid}' for vmid in sorted(guest_types, key=int)) + '\n' + context
     language = AI_LANGUAGES.get(ai_config.get('ai_language', 'en'), 'English')
     config = {
         **ai_config, 'channel_type': '', 'ai_prompt_mode': 'custom',
@@ -3477,7 +3490,19 @@ def format_backup_diagnostics_with_ai(title: str, body: str, severity: str,
     except Exception:
         result = {}
     explanation = str(result.get('body') or '').strip()
-    if not explanation or explanation == context or len(explanation) > 2000:
+    invalid = not explanation or explanation == context or len(explanation) > 2000
+    if not invalid:
+        references = re.compile(r'\b(?:VM|CT)\s*#?\s*(\d+)\b', re.IGNORECASE)
+        failed_ids = {str(vmid) for vmid in analysis.get('failed_ids', [])}
+        if failed_ids and set(references.findall(explanation)) - failed_ids:
+            invalid = True
+        else:
+            # A fluent diagnosis must not turn a VM into a CT (or vice versa).
+            # Unknown types stay neutral rather than inheriting the model's guess.
+            explanation = references.sub(
+                lambda match: f"{guest_types.get(match.group(1), 'ID')} {match.group(1)}",
+                explanation)
+    if invalid:
         explanation = _backup_quota_explanation(
             diagnostics, analysis, data, ai_config.get('ai_language', 'en'), use_emojis)
         if not explanation:

AppImage/scripts/tests/test_backup_ai_diagnostics.py

@@ -41,6 +41,9 @@ def report(count=11):
                                'null' if failed else f'ct/{vmid}/2026-11-01T06:00:00Z'))
     return ('Proxmox vzdump report\n' + '\n'.join(rows)
             + '\nTotal running time: 00:14:47\nTotal size: 110.487 GiB\n'
+            + '108: 2026-11-01 07:10:47 INFO: Starting Backup of VM 108 (lxc)\n'
+            + '110: 2026-11-01 07:14:47 INFO: Starting Backup of VM 110 (qemu)\n'
+            + '9000: 2026-11-01 07:14:48 INFO: Starting Backup of VM 9000 (lxc)\n'
             + "108: 2026-11-01 07:14:47 ERROR: Backup of VM 108 failed - command '/usr/bin/proxmox-backup-client backup root.pxar:/mnt/vzsnap0' failed: exit code 255\n"
             + "110: 2026-11-01 07:14:48 ERROR: VM 110 qmp command 'backup' failed - backup connect failed: command error: Disk quota exceeded (os error 122)\n"
             + "9000: 2026-11-01 07:14:53 ERROR: Backup of VM 9000 failed - command 'lxc-usernsexec -- proxmox-backup-client backup' failed: exit code 255\n")
@@ -183,6 +186,33 @@ class BackupAIDiagnosticsTests(unittest.TestCase):
                              {'title': rendered['title'], 'body': rendered['body']})
         self.assertEqual(self.prompts, [])
 
+    def test_ai_guest_type_is_corrected_using_authoritative_inventory(self):
+        data, rendered = render()
+        self.assertEqual(data['_backup_analysis']['guest_types'],
+                         {'108': 'CT', '110': 'VM', '9000': 'CT'})
+        with patch.object(self.provider, 'generate', return_value=(
+                '[TITLE]\nIgnored\n[BODY]\nCT 110: cuota excedida. VM 108: falta causa.')) as generate:
+            result = self.explain(data, rendered)
+        explanation = result['body'][len(data['_backup_analysis']['inventory_body']):]
+        self.assertIn('VM 110: cuota excedida', explanation)
+        self.assertIn('CT 108: falta causa', explanation)
+        self.assertIn('Guest identities: CT 108, VM 110, CT 9000', generate.call_args.args[1])
+
+    def test_ai_cannot_introduce_a_failed_guest_outside_the_report(self):
+        data, rendered = render()
+        with patch.object(self.provider, 'generate', return_value=(
+                '[TITLE]\nIgnored\n[BODY]\nVM 999: backup failed.')):
+            result = self.explain(data, rendered)
+        self.assertNotIn('999', result['body'])
+        self.assertIn('Copias 110:', result['body'])
+
+    def test_unknown_guest_type_is_neutral_in_ai_explanation(self):
+        data, rendered = render()
+        data['_backup_analysis']['guest_types'].pop('110')
+        result = self.explain(data, rendered)
+        explanation = result['body'][len(data['_backup_analysis']['inventory_body']):]
+        self.assertIn('ID 110:', explanation)
+
     def test_missing_credentials_use_evidenced_quota_fallback(self):
         data, rendered = render()
         result = self.explain(data, rendered, ai_provider='openai', ai_api_key='')