fix(monitor): reactivate resolved health conditions

.github/scripts/run_offline_qualified.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Run only the independently qualified, host-independent tests/ fixtures.
+"""Run only the independently qualified, host-independent fixtures.
 
 No filesystem discovery widens this manifest. Run from the repository root;
 Python and Node have separate CI jobs so either failure blocks the check.
@@ -10,6 +10,7 @@ import os
 from pathlib import Path
 import subprocess
 import sys
+import tempfile
 
 PYTHON_FILES = (
     "tests/test_audit_presentation.py",
@@ -20,6 +21,7 @@ PYTHON_FILES = (
     "tests/storage/test_nvme_status_message.py",
     "tests/test_fastfetch_config_generation.py",
     "tests/test_notification_active_payloads.py",
+    "AppImage/scripts/tests/test_health_persistence_recurrence.py",
 )
 NODE_FILES = (
     "tests/lxc_updates/test_docker_delegated_ui.cjs",
@@ -69,6 +71,10 @@ def main() -> int:
         if args.lane == "python":
             folder, filename = str(Path(path).parent), Path(path).name
             command = [sys.executable, "-I", "-B", "-c", PYTHON_RUN, folder, filename]
+            with tempfile.TemporaryDirectory(prefix="qualified-offline-") as scratch:
+                environment = os.environ.copy()
+                environment["TMPDIR"] = scratch
+                result = subprocess.run(command, cwd=root, env=environment, check=False)
         else:
             command = ["node", path]
             result = subprocess.run(command, cwd=root, env=os.environ.copy(), check=False)

.github/workflows/test-offline-qualified.yml

@@ -11,6 +11,7 @@ on:
       - 'AppImage/package.json'
       - 'AppImage/package-lock.json'
       - 'AppImage/scripts/*.py'
+      - 'AppImage/scripts/tests/test_health_persistence_recurrence.py'
       - 'AppImage/app/**'
       - 'AppImage/components/**'
       - 'AppImage/hooks/**'
@@ -30,6 +31,7 @@ on:
       - 'AppImage/package.json'
       - 'AppImage/package-lock.json'
       - 'AppImage/scripts/*.py'
+      - 'AppImage/scripts/tests/test_health_persistence_recurrence.py'
       - 'AppImage/app/**'
       - 'AppImage/components/**'
       - 'AppImage/hooks/**'
@@ -54,7 +56,7 @@ jobs:
       - uses: actions/setup-python@v5
         with:
           python-version: '3.11'
-      - name: Run eight qualified Python files
+      - name: Run nine qualified Python files
         env:
           PYTHONDONTWRITEBYTECODE: '1'
         run: python3 -I -B .github/scripts/run_offline_qualified.py --lane python

AppImage/scripts/health_persistence.py

@@ -630,6 +630,12 @@ class HealthPersistence:
 
             event_info = {'type': 'updated', 'needs_notification': False}
 
+            if existing and existing[1] == 0 and existing[2] is not None:
+                # A resolved, non-dismissed condition returning is a new episode.
+                # Keep event history and reuse normal insertion/suppression policy.
+                cursor.execute('DELETE FROM errors WHERE id = ?', (existing[0],))
+                existing = None
+
             if existing:
                 (err_id, ack, resolved_at, old_cat, old_severity, first_seen,
                  notif_sent, stored_suppression, acknowledged_at) = existing

AppImage/scripts/tests/test_health_persistence_recurrence.py

@@ -0,0 +1,495 @@
+"""Offline recurrence tests against the complete, real persistence module.
+
+From the repository root (stdlib only; no broad test discovery):
+    TMPDIR=/path/to/private/scratch python3 -I -B -m unittest discover \
+        -s AppImage/scripts/tests -p test_health_persistence_recurrence.py -v
+
+TMPDIR must already exist and must not be the system temporary directory.
+Only the data-directory Path boundary and datetime clock are redirected, before
+module execution (including the real global constructor). No persistence method
+is stubbed. Every test has its own real SQLite DB. Audit guards constrain DB
+connections and file opens, restrict mkdir/remove/rmdir/rename to the fixture,
+and reject subprocess/socket operations. Metadata/existence probes (for example
+os.path.exists) are not comprehensively guarded; synthetic keys avoid VM/disk
+resource-existence checks. Read opens are limited to fixture/source/stdlib paths
+except the nonexistent synthetic traceback filename described in the guard.
+These are synthetic recorder tests, not transport, UI or deployment validation.
+"""
+
+import contextlib
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import sqlite3
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+from unittest.mock import patch
+
+
+# Use the concrete platform class while pathlib.Path is temporarily replaced.
+# Python 3.11's abstract Path.__new__ consults the patched module-level name.
+Path = type(Path())
+ROOT = Path(__file__).resolve().parents[3]
+SOURCE = ROOT / 'AppImage/scripts/health_persistence.py'
+_AUDIT_SCOPE = None
+
+
+class BoundaryViolation(RuntimeError):
+    pass
+
+
+def _audit_guard(event, args):
+    scope = _AUDIT_SCOPE
+    if scope is None:
+        return
+
+    def reject():
+        scope['blocked'].append((event, repr(args)))
+        raise BoundaryViolation('Offline fixture rejected ' + event)
+
+    def inside(path, directory):
+        return Path(os.fsdecode(path)).resolve().is_relative_to(directory)
+
+    if event.startswith(('socket.', 'subprocess.', 'os.system', 'os.exec',
+                         'os.spawn', 'os.fork', 'pty.')):
+        reject()
+    if event == 'sqlite3.connect':
+        if Path(os.fsdecode(args[0])).resolve() != scope['db']:
+            reject()
+        scope['connections'].append(str(args[0]))
+    if event == 'open':
+        path, mode, flags = args
+        if isinstance(path, int):
+            reject()
+        writing = (isinstance(mode, str) and any(c in mode for c in 'wax+')) or (
+            isinstance(flags, int) and flags & (
+                os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_TRUNC))
+        if writing:
+            if not inside(path, scope['root']):
+                reject()
+        elif not any(inside(path, directory) for directory in (
+                scope['root'], ROOT, Path(sys.base_prefix) / 'lib')):
+            # unittest traceback formatting may ask tokenize to read this exact
+            # synthetic name. Permit only the harmless missing-file read; an
+            # existing file or even a dangling symlink must still be rejected.
+            if not (path == '<unknown>' and mode == 'rb' and flags == 0
+                    and not os.path.lexists(path)):
+                reject()
+    if event in ('os.mkdir', 'os.remove', 'os.rmdir', 'os.rename'):
+        paths = args[:2] if event == 'os.rename' else args[:1]
+        if not all(inside(path, scope['root']) for path in paths):
+            reject()
+
+
+sys.addaudithook(_audit_guard)
+
+
+class FrozenDatetime(datetime):
+    current = datetime(2026, 10, 1, 12)
+
+    @classmethod
+    def now(cls, tz=None):
+        if tz is not None:
+            raise AssertionError('Unexpected timezone request')
+        return cls.current
+
+
+class HealthPersistenceRecurrenceTests(unittest.TestCase):
+    def setUp(self):
+        global _AUDIT_SCOPE
+        scratch = os.environ.get('TMPDIR')
+        if not scratch or not Path(scratch).is_dir():
+            self.fail('Set TMPDIR to an existing private scratch directory')
+        if Path(scratch).resolve() in (Path('/tmp'), Path('/var/tmp')):
+            self.fail('System temporary directories are not allowed')
+        temporary = tempfile.TemporaryDirectory(prefix='health-recurrence-', dir=scratch)
+        self.addCleanup(temporary.cleanup)
+        self.data_dir = Path(temporary.name)
+        self.scope = {'root': self.data_dir, 'db': self.data_dir / 'health_monitor.db',
+                      'blocked': [], 'connections': []}
+        _AUDIT_SCOPE = self.scope
+        self.addCleanup(self.finish_isolation)
+        FrozenDatetime.current = datetime(2026, 10, 1, 12)
+
+        def redirected_path(*args, **kwargs):
+            if args == ('/usr/local/share/proxmenux',):
+                return self.data_dir
+            return Path(*args, **kwargs)
+
+        spec = importlib.util.spec_from_file_location('offline_health_persistence', SOURCE)
+        assert spec is not None and spec.loader is not None
+        self.module = importlib.util.module_from_spec(spec)
+        with patch('pathlib.Path', redirected_path), \
+                patch('datetime.datetime', FrozenDatetime), \
+                contextlib.redirect_stdout(io.StringIO()):
+            spec.loader.exec_module(self.module)
+        self.db = self.module.health_persistence
+        self.assertEqual(self.db.db_path, self.scope['db'])
+
+    def finish_isolation(self):
+        global _AUDIT_SCOPE
+        _AUDIT_SCOPE = None
+        self.assertEqual(self.scope['blocked'], [], 'Even swallowed boundary violations fail')
+        self.assertTrue(self.scope['connections'], 'The real SQLite constructor must run')
+
+    def advance(self, **kwargs):
+        FrozenDatetime.current += timedelta(**kwargs)
+
+    def record(self, key='security_updates', category='updates', severity='INFO',
+               reason='Synthetic condition', details=None):
+        return self.db.record_error(key, category, severity, reason, details)
+
+    def snapshot(self):
+        # Independent fresh SQLite connection verifies durable, committed state.
+        conn = sqlite3.connect(str(self.db.db_path))
+        conn.row_factory = sqlite3.Row
+        try:
+            rows = [dict(row) for row in conn.execute('SELECT * FROM errors ORDER BY id')]
+            events = [dict(row) for row in conn.execute('SELECT * FROM events ORDER BY id')]
+            for event in events:
+                event['data'] = json.loads(event['data'])
+            return rows, events
+        finally:
+            conn.close()
+
+    def test_notified_resolved_condition_returns_as_a_fresh_episode(self):
+        key = 'security_updates'
+        self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True})
+        self.db.mark_notified(key)
+        self.advance(minutes=1)
+        self.db.resolve_error(key, 'Synthetic condition absent')
+        self.assertEqual(self.db.get_active_errors(), [])
+        self.assertEqual(self.db.get_unnotified_errors(), [])
+        self.advance(minutes=1)
+        details = {'name': 'offline fixture', 'synthetic': True, 'pending': 3}
+        result = self.record(reason='Synthetic condition returned', details=details)
+        rows, events = self.snapshot()
+        self.assertIsNone(rows[0]['resolved_at'], 'Resolved condition must reactivate')
+        self.assertEqual(result, {'type': 'new', 'needs_notification': True})
+        row = rows[0]
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat())
+        self.assertEqual(row['last_seen'], FrozenDatetime.current.isoformat())
+        self.assertEqual(row['notification_sent'], 0)
+        self.assertEqual(row['acknowledged'], 0)
+        self.assertIsNone(row['resolution_type'])
+        self.assertIsNone(row['resolution_reason'])
+        self.assertEqual(json.loads(row['details']), details)
+        self.assertTrue(self.db.is_error_active(key, 'updates'))
+        self.assertEqual(self.db.get_error_by_key(key)['reason'], 'Synthetic condition returned')
+        self.assertEqual([e['error_key'] for e in self.db.get_active_errors('updates')], [key])
+        self.assertEqual([e['error_key'] for e in self.db.get_unnotified_errors()], [key])
+        self.assertEqual([event['event_type'] for event in events], ['new', 'resolved', 'new'])
+        self.assertEqual(events[-1]['data']['details'], details)
+        self.assertEqual(events[-1]['data']['entity'], 'offline fixture')
+
+    def test_resolve_and_clear_recurrence_matrix(self):
+        for severity in ('INFO', 'WARNING', 'CRITICAL'):
+            for route in ('resolve_error', 'clear_error'):
+                for notified in (False, True):
+                    key = f'fixture_{severity}_{route}_{notified}'
+                    with self.subTest(severity=severity, route=route, notified=notified):
+                        self.record(key=key, severity=severity)
+                        if notified:
+                            self.db.mark_notified(key)
+                        self.advance(minutes=1)
+                        getattr(self.db, route)(key)
+                        self.assertFalse(self.db.is_error_active(key))
+                        self.advance(minutes=1)
+                        self.assertEqual(self.record(key=key, severity=severity),
+                                         {'type': 'new', 'needs_notification': True})
+                        row = self.db.get_error_by_key(key)
+                        self.assertIsNotNone(row)
+                        for field in ('resolved_at', 'resolution_type', 'resolution_reason',
+                                      'acknowledged_at'):
+                            self.assertIsNone(row[field], field)
+                        self.assertEqual(row['notification_sent'], 0)
+                        self.assertEqual(row['acknowledged'], 0)
+                        self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat())
+                        self.assertEqual(row['last_seen'], row['first_seen'])
+                        self.assertIn(key, [e['error_key'] for e in self.db.get_unnotified_errors()])
+                        rows, events = self.snapshot()
+                        self.assertEqual(sum(row['error_key'] == key for row in rows), 1)
+                        self.assertEqual([e['event_type'] for e in events if e['error_key'] == key],
+                                         ['new', 'resolved' if route == 'resolve_error' else 'cleared', 'new'])
+
+    def test_repeated_polls_after_reopening_rearm_only_once(self):
+        key = 'security_updates'
+        self.record()
+        self.db.mark_notified(key)
+        self.db.resolve_error(key)
+        self.advance(minutes=1)
+        self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True})
+        first_seen = self.db.get_error_by_key(key)['first_seen']
+        for poll in range(4):
+            self.advance(minutes=1)
+            self.assertEqual(self.record(), {'type': 'updated', 'needs_notification': False})
+            self.assertEqual(self.db.get_error_by_key(key)['first_seen'], first_seen)
+            self.assertEqual(self.db.get_error_by_key(key)['notification_sent'], int(poll > 0))
+            if poll == 0:
+                self.db.mark_notified(key)
+        rows, events = self.snapshot()
+        self.assertEqual(len(rows), 1)
+        self.assertEqual([e['event_type'] for e in events],
+                         ['new', 'resolved', 'new'] + ['updated'] * 4)
+        self.assertEqual(self.db.get_unnotified_errors(), [])
+        # A second real constructor sees exactly the same committed episode/history.
+        with contextlib.redirect_stdout(io.StringIO()):
+            self.db = self.module.HealthPersistence()
+        self.assertEqual(self.snapshot(), (rows, events))
+        self.assertTrue(self.db.is_error_active(key))
+        self.assertEqual(self.db.get_unnotified_errors(), [])
+
+    def test_same_key_can_resolve_and_recur_twice(self):
+        key = 'security_updates'
+        self.record()
+        for _ in range(2):
+            self.db.mark_notified(key)
+            self.advance(minutes=1)
+            self.db.resolve_error(key, 'Synthetic recovery')
+            resolved = self.snapshot()
+            self.db.resolve_error(key, 'Duplicate recovery poll')
+            self.assertEqual(self.snapshot(), resolved)
+            self.advance(minutes=1)
+            self.assertEqual(self.record(), {'type': 'new', 'needs_notification': True})
+            self.assertEqual(self.db.get_error_by_key(key)['notification_sent'], 0)
+        rows, events = self.snapshot()
+        self.assertEqual(len(rows), 1)
+        self.assertEqual([e['event_type'] for e in events],
+                         ['new', 'resolved', 'new', 'resolved', 'new'])
+        self.assertEqual([e['data']['reason'] for e in events if e['event_type'] == 'resolved'],
+                         ['Synthetic recovery', 'Synthetic recovery'])
+
+    def test_active_polls_preserve_existing_episode_and_notification_state(self):
+        for severity in ('INFO', 'WARNING'):
+            for notified in (False, True):
+                key = f'fixture_active_{severity}_{notified}'
+                with self.subTest(severity=severity, notified=notified):
+                    self.record(key=key, severity=severity)
+                    if notified:
+                        self.db.mark_notified(key)
+                    before = self.db.get_error_by_key(key)
+                    self.advance(minutes=1)
+                    self.assertEqual(self.record(key=key, severity=severity, reason='Next poll'),
+                                     {'type': 'updated', 'needs_notification': False})
+                    after = self.db.get_error_by_key(key)
+                    for field in ('first_seen', 'notification_sent', 'acknowledged',
+                                  'resolved_at', 'resolution_type', 'resolution_reason'):
+                        self.assertEqual(after[field], before[field], field)
+                    self.assertEqual(after['last_seen'], FrozenDatetime.current.isoformat())
+                    self.assertEqual(after['reason'], 'Next poll')
+
+    def test_active_warning_escalation_is_unchanged(self):
+        self.record(severity='WARNING')
+        self.db.mark_notified('security_updates')
+        before = self.db.get_error_by_key('security_updates')
+        self.advance(minutes=1)
+        self.assertEqual(self.record(severity='CRITICAL'),
+                         {'type': 'escalated', 'needs_notification': True})
+        after = self.db.get_error_by_key('security_updates')
+        self.assertEqual(after['first_seen'], before['first_seen'])
+        self.assertEqual(after['notification_sent'], 1)
+        self.assertIsNone(after['resolved_at'])
+
+    def test_acknowledged_rows_within_window_are_untouched(self):
+        for resolved in (False, True):
+            key = f'fixture_timed_{resolved}'
+            with self.subTest(resolved=resolved):
+                self.record(key=key, severity='WARNING')
+                self.db.mark_notified(key)
+                self.db.acknowledge_error(key, suppression_hours=24)
+                if resolved:
+                    self.db.resolve_error(key)
+                before = self.snapshot()
+                self.advance(hours=23)
+                self.assertEqual(self.record(key=key, severity='CRITICAL'),
+                                 {'type': 'skipped_acknowledged', 'needs_notification': False})
+                self.assertEqual(self.snapshot(), before)
+                self.assertTrue(self.db.is_error_acknowledged(key))
+                self.assertTrue(self.db.is_error_active(key))
+
+    def test_permanent_dismissal_survives_recovery_and_recurrence(self):
+        for resolved_before_dismiss in (False, True):
+            key = f'fixture_permanent_{resolved_before_dismiss}'
+            with self.subTest(resolved_before_dismiss=resolved_before_dismiss):
+                self.db.set_setting('suppress_updates', '24')
+                self.record(key=key, severity='WARNING')
+                self.db.mark_notified(key)
+                if resolved_before_dismiss:
+                    self.db.resolve_error(key)
+                self.db.acknowledge_error(key, suppression_hours=-1)
+                before = self.snapshot()
+                self.db.clear_error(key)
+                self.db.resolve_error(key)
+                self.advance(days=30)
+                self.db.set_setting('suppress_updates', '48')
+                self.assertEqual(self.record(key=key, severity='CRITICAL'),
+                                 {'type': 'skipped_acknowledged', 'needs_notification': False})
+                self.assertEqual(self.snapshot(), before)
+                self.assertTrue(self.db.is_error_permanently_acknowledged(key))
+                self.assertEqual(self.db.get_active_errors(), [])
+                self.assertEqual(self.db.get_unnotified_errors(), [])
+
+    def test_expired_resolved_dismissal_rearms_once(self):
+        key = 'security_updates'
+        self.record(severity='WARNING')
+        self.db.mark_notified(key)
+        self.db.acknowledge_error(key, suppression_hours=24)
+        self.db.resolve_error(key)
+        self.advance(hours=25)
+        self.assertEqual(self.record(severity='WARNING'),
+                         {'type': 'new', 'needs_notification': True})
+        row = self.db.get_error_by_key(key)
+        self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat())
+        self.assertIsNone(row['resolved_at'])
+        self.assertEqual(row['acknowledged'], 0)
+        self.assertEqual(row['notification_sent'], 0)
+        self.assertEqual(self.record(severity='WARNING'),
+                         {'type': 'updated', 'needs_notification': False})
+        self.assertEqual([e['event_type'] for e in self.snapshot()[1]],
+                         ['new', 'acknowledged', 'resolved', 'new', 'updated'])
+
+    def test_recurrence_uses_current_category_auto_suppression(self):
+        for category, setting in (('updates', 'suppress_updates'), ('memory', 'suppress_memory')):
+            for hours in (48, -1):
+                key = f'fixture_auto_{category}_{hours}'
+                with self.subTest(category=category, hours=hours):
+                    # Establish each case independently, even if an earlier
+                    # subtest failed before it could restore shared settings.
+                    self.db.set_setting(setting, '24')
+                    self.record(key=key, category=category, severity='WARNING')
+                    self.db.mark_notified(key)
+                    self.db.resolve_error(key)
+                    self.db.set_setting(setting, str(hours))
+                    self.advance(minutes=1)
+                    self.assertEqual(self.record(key=key, category=category, severity='WARNING'),
+                                     {'type': 'auto_suppressed', 'needs_notification': False})
+                    rows, events = self.snapshot()
+                    row = next(row for row in rows if row['error_key'] == key)
+                    self.assertIsNone(row['resolved_at'])
+                    self.assertIsNone(row['resolution_type'])
+                    self.assertIsNone(row['resolution_reason'])
+                    self.assertEqual(row['notification_sent'], 0)
+                    self.assertEqual(row['acknowledged'], 1)
+                    self.assertEqual(row['suppression_hours'], hours)
+                    self.assertEqual(row['first_seen'], FrozenDatetime.current.isoformat())
+                    self.assertEqual(row['acknowledged_at'], row['first_seen'])
+                    self.assertEqual([e['event_type'] for e in events if e['error_key'] == key],
+                                     ['new', 'resolved', 'auto_suppressed'])
+                    before = self.snapshot()
+                    self.assertEqual(self.record(key=key, category=category, severity='WARNING'),
+                                     {'type': 'skipped_acknowledged', 'needs_notification': False})
+                    self.assertEqual(self.snapshot(), before)
+                    self.assertNotIn(key, [e['error_key'] for e in self.db.get_active_errors()])
+                    self.assertNotIn(key, [e['error_key'] for e in self.db.get_unnotified_errors()])
+                    self.db.set_setting(setting, '24')
+
+    def test_critical_cpu_recurrence_bypasses_category_auto_suppression(self):
+        key = 'cpu_temperature'
+        self.record(key=key, category='temperature', severity='WARNING')
+        self.db.mark_notified(key)
+        self.db.resolve_error(key)
+        self.db.set_setting('suppress_cpu', '-1')
+        self.advance(minutes=1)
+        self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'),
+                         {'type': 'new', 'needs_notification': True})
+        row = self.db.get_error_by_key(key)
+        self.assertEqual(row['acknowledged'], 0)
+        self.assertEqual(row['notification_sent'], 0)
+        self.assertIsNone(row['resolved_at'])
+        self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'),
+                         {'type': 'updated', 'needs_notification': False})
+
+    def test_critical_cpu_override_of_permanent_dismissal_is_unchanged(self):
+        key = 'cpu_temperature'
+        self.record(key=key, category='temperature', severity='WARNING')
+        self.db.acknowledge_error(key, suppression_hours=-1)
+        self.db.clear_error(key)
+        self.db.resolve_error(key)
+        self.db.set_setting('suppress_cpu', '-1')
+        self.advance(minutes=1)
+        self.assertEqual(self.record(key=key, category='temperature', severity='CRITICAL'),
+                         {'type': 'new', 'needs_notification': True})
+        row = self.db.get_error_by_key(key)
+        self.assertEqual(row['acknowledged'], 0)
+        self.assertEqual(row['notification_sent'], 0)
+        self.assertIsNone(row['resolved_at'])
+        self.assertEqual([e['event_type'] for e in self.snapshot()[1]],
+                         ['new', 'acknowledged', 'new'])
+
+    def test_reopening_one_key_preserves_other_rows_and_delivery_state(self):
+        self.record()
+        self.record(key='security_updates_other', severity='WARNING')
+        self.db.mark_notified('security_updates_other')
+        self.db.resolve_error('security_updates')
+        other = self.db.get_error_by_key('security_updates_other')
+        conn = sqlite3.connect(str(self.db.db_path))
+        try:
+            conn.execute('INSERT INTO notification_delivery_claims VALUES (?, ?, ?)',
+                         ('health_security_updates', 'in-flight-token', 123))
+            conn.execute('INSERT INTO notification_last_sent VALUES (?, ?, ?)',
+                         ('health_security_updates', 123, 1))
+            conn.commit()
+            self.advance(minutes=1)
+            self.record()
+            self.assertEqual(self.db.get_error_by_key('security_updates_other'), other)
+            self.assertEqual(conn.execute('SELECT * FROM notification_delivery_claims').fetchall(),
+                             [('health_security_updates', 'in-flight-token', 123)])
+            self.assertEqual(conn.execute('SELECT * FROM notification_last_sent').fetchall(),
+                             [('health_security_updates', 123, 1)])
+        finally:
+            conn.close()
+
+    def test_failed_reopening_rolls_back_without_losing_resolved_episode(self):
+        self.record()
+        self.db.mark_notified('security_updates')
+        self.db.resolve_error('security_updates')
+        before = self.snapshot()
+        conn = sqlite3.connect(str(self.db.db_path))
+        try:
+            conn.execute('''CREATE TRIGGER fixture_reject_new_episode BEFORE INSERT ON errors
+                            BEGIN SELECT RAISE(ABORT, 'fixture rejects insert'); END''')
+            conn.commit()
+        finally:
+            conn.close()
+        self.advance(minutes=1)
+        with self.assertRaisesRegex(sqlite3.IntegrityError, 'fixture rejects insert'):
+            self.record()
+        self.assertEqual(self.snapshot(), before)
+
+    def test_failed_new_event_rolls_back_reopened_row_and_history(self):
+        self.record()
+        self.db.mark_notified('security_updates')
+        self.db.resolve_error('security_updates')
+        before = self.snapshot()
+        old_id = before[0][0]['id']
+        self.assertEqual([event['event_type'] for event in before[1]], ['new', 'resolved'])
+        conn = sqlite3.connect(str(self.db.db_path))
+        try:
+            # Fail at the later durable-event insert, only after the old row
+            # has been replaced by the fresh episode within this transaction.
+            conn.execute(f'''CREATE TRIGGER fixture_reject_new_event BEFORE INSERT ON events
+                             WHEN NEW.event_type = 'new' AND NEW.error_key = 'security_updates'
+                             BEGIN
+                                 SELECT RAISE(ABORT, 'fixture rejects new event')
+                                 WHERE EXISTS (
+                                     SELECT 1 FROM errors
+                                     WHERE error_key = NEW.error_key AND id <> {old_id}
+                                       AND resolved_at IS NULL AND first_seen = NEW.timestamp
+                                 );
+                             END''')
+            conn.commit()
+        finally:
+            conn.close()
+        self.advance(minutes=1)
+        with self.assertRaisesRegex(sqlite3.IntegrityError, 'fixture rejects new event'):
+            self.record()
+        self.assertEqual(self.snapshot(), before)
+
+
+if __name__ == '__main__':
+    unittest.main()

CONTRIBUTING.md

@@ -863,8 +863,11 @@ publication workflow and has read-only repository permissions.
 #### Qualified offline `tests/` CI lanes
 
 `.github/workflows/test-offline-qualified.yml` runs **only** the explicitly listed
-eight Python files (including the inert active-notification payload fixture) and eleven
+nine Python files (including the inert active-notification payload fixture and
+the focused health recurrence tests) and eleven
 Node files in `.github/scripts/run_offline_qualified.py`.
+The Python runner gives each test file a private temporary directory through
+`TMPDIR`. It removes that directory after the test file finishes.
 Run the same allowlist locally from the root of a clean, disposable checkout
 with no `AppImage/node_modules`. Do not run this over a normal Monitor install:
 the commands refuse an existing directory, file, or symlink (including a
@@ -892,7 +895,7 @@ React peer constraint against locked React 19; it does not resolve that mismatch
 tests, skipped Python tests, or a nonzero exit from either lane. Node fixtures
 use top-level assertions, not a runner that reports case/skip totals.
 These are local fixture tests, not the whole `tests/` tree, `AppImage/tests/`,
-`AppImage/scripts/tests/`, browser smoke, a live Proxmox host, or an AppImage
+the whole `AppImage/scripts/tests/` tree, browser smoke, a live Proxmox host, or an AppImage
 build. The CI workflow has read-only permissions and runs on relevant PRs and
 main/develop pushes, or manually. Review the input paths and the test's imports
 before extending the explicit manifest.