fix(config): stop same-second backups overwriting each other (#564)

* fix(config): stop same-second backups overwriting each other

A backup's version is its identity. save_config_atomic() hands the path
back, rollback_config(backup_version=...) looks that version up, and the
paired secrets backup is found by reusing the same string.

The version was stamped at second granularity, so two saves inside the
same second produced the same filename and the second shutil.copy2()
silently overwrote the first backup. The path a caller was still holding
then pointed at different content, and rolling back to it restored the
wrong config. A user saving twice in quick succession lost a restore
point with no error.

list_backups() made it worse. It parsed the version off Path.stem, which
drops only the last dot-component, so for config.json.backup.20240101_120000
parts was ['config', 'json', 'backup'] and parts[-2] was 'json' -- never
'backup'. The filename branch was unreachable: every backup fell through
to the mtime fallback and reported a second-granularity restamp of its
mtime rather than the name on disk, so a unique filename alone would not
have been enough for rollback to find the right version.

Stamp microseconds, and never overwrite an existing backup -- on a
collision bump a -N suffix rather than lose a restore point. Parse the
version off the exact glob prefix so it round-trips with the filename,
still reading the legacy second-granularity format so restore points that
predate this keep working.

Two tests had encoded the bug:

  - test_multiple_config_changes asserted a rollback produced plugin1=45
    with plugin2=15, a state no single backup ever held -- 45 was only in
    the second backup, 15 only in the first. It passed because the two
    saves collided onto one file, so the first version resolved to the
    second's content. Corrected to the state that backup actually holds.

  - test_backup_rotation asserted against a hardcoded max of 3 while
    setUp configured 5, and still passed: every save in its loop collapsed
    onto a single filename, so there was only ever one backup to count and
    rotation was never exercised. It now asks the manager for its limit
    and overshoots it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(config): fold collision suffix into ordering, close backup-path race

_parse_backup_version() stripped any trailing "-segment" unconditionally,
so a collision-suffixed backup parsed to the exact same timestamp as its
sibling and list_backups() had no deterministic way to order them. Only
strip the suffix when it's numeric, and fold it back in as extra
microseconds so same-tick collisions sort newest-first reliably.

_create_backup() also checked backup_path.exists() before shutil.copy2(),
which two concurrent callers can both pass for the same path -- the second
copy2() then silently destroys the first call's restore point. Reserve
each path (config and, when configured, secrets) with exclusive file
creation instead of a check-then-copy, retrying on a real conflict.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

src/config_manager_atomic.py

@@ -7,9 +7,10 @@ and enable recovery from failed saves.
 
 import json
 import os
+import re
 import shutil
 import tempfile
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Dict, Any, Optional, List, Tuple
 from dataclasses import dataclass
@@ -19,6 +20,19 @@ from src.exceptions import ConfigError
 from src.logging_config import get_logger
 from src.common.permission_utils import ensure_shared_group_ownership
 
+# Version stamp in a backup's filename: config.json.backup.<version>.
+BACKUP_VERSION_FORMAT = "%Y%m%d_%H%M%S_%f"
+
+# Backups written before the format gained microseconds. Still read, never
+# written, so existing restore points on a rig stay usable after an upgrade.
+LEGACY_BACKUP_VERSION_FORMAT = "%Y%m%d_%H%M%S"
+
+# The numeric collision suffix _create_backup() appends to break a same-tick
+# tie: config.json.backup.<version>-<N>. Only digits count as this suffix, so
+# a hand-copied or renamed backup that happens to end in "-something" isn't
+# mistaken for one and silently mis-parsed.
+_BACKUP_COLLISION_SUFFIX_RE = re.compile(r"^(?P<base>.+)-(?P<collision>\d+)$")
+
 
 class SaveResultStatus(Enum):
     """Status of a save operation."""
@@ -246,23 +260,34 @@ class AtomicConfigManager:
         if not self.backup_dir.exists():
             return backups
         
-        # Look for backup files (format: config.json.backup.YYYYMMDD_HHMMSS)
+        # Look for backup files (format: config.json.backup.<version>)
         config_name = self.config_path.name
         backup_pattern = f"{config_name}.backup.*"
-        
+
         for backup_file in self.backup_dir.glob(backup_pattern):
             try:
-                # Extract timestamp from filename
+                # The version reported here is what rollback_config() matches
-                # Format: config.json.backup.20240101_120000
+                # against, so it has to be the exact string in the filename.
-                parts = backup_file.stem.split('.')
+                #
-                if len(parts) >= 3 and parts[-2] == 'backup':
+                # It did not used to be. This read .stem, which drops only the
-                    timestamp_str = parts[-1]
+                # last dot-component, so for config.json.backup.20240101_120000
-                    timestamp = datetime.strptime(timestamp_str, "%Y%m%d_%H%M%S")
+                # parts was ['config', 'json', 'backup'] and parts[-2] was
-                else:
+                # 'json' -- never 'backup'. The filename branch could not be
-                    # Fallback: use file modification time
+                # reached, every backup fell through to the mtime fallback, and
+                # the version was a second-granularity restamp of the mtime
+                # rather than the name on disk. Two backups a second apart could
+                # therefore report the same version, and rollback would pick
+                # whichever the glob happened to yield first.
+                # Strip the exact prefix the glob just matched, so a config
+                # whose own name contains '.backup.' can't shift the split.
+                timestamp_str = backup_file.name[len(f"{config_name}.backup."):]
+                timestamp = self._parse_backup_version(timestamp_str)
+                if timestamp is None:
+                    # Not a version this code wrote (hand-copied, renamed).
+                    # Order it by mtime, but keep the on-disk version string so
+                    # it can still be named in a rollback.
                     timestamp = datetime.fromtimestamp(backup_file.stat().st_mtime)
-                    timestamp_str = timestamp.strftime("%Y%m%d_%H%M%S")
+
-                
                 # Validate backup file
                 is_valid = self._validate_backup_file(backup_file)
                 
@@ -283,6 +308,35 @@ class AtomicConfigManager:
         
         return backups
     
+    @staticmethod
+    def _parse_backup_version(version: str) -> Optional[datetime]:
+        """
+        Parse the ``<version>`` of a ``config.json.backup.<version>`` filename
+        into the time the backup was taken, or None if it is not a version this
+        class wrote.
+
+        Accepts the current microsecond format and the legacy second-granularity
+        one, with or without the ``-N`` suffix _create_backup() appends to break
+        a collision. When that suffix is present, N is folded into the result
+        as extra microseconds so same-tick collisions still sort in the order
+        they were created rather than tying.
+        """
+        if not version:
+            return None
+        base = version
+        collision = 0
+        match = _BACKUP_COLLISION_SUFFIX_RE.match(version)
+        if match:
+            base = match.group('base')
+            collision = int(match.group('collision'))
+        for fmt in (BACKUP_VERSION_FORMAT, LEGACY_BACKUP_VERSION_FORMAT):
+            try:
+                parsed = datetime.strptime(base, fmt)
+            except ValueError:
+                continue
+            return parsed + timedelta(microseconds=collision) if collision else parsed
+        return None
+
     def validate_config_file(self, config_path: Optional[str] = None) -> ValidationResult:
         """
         Validate a configuration file.
@@ -303,19 +357,55 @@ class AtomicConfigManager:
             return None
         
         try:
-            # Generate backup filename with timestamp
+            # Generate backup filename with timestamp.
-            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+            #
+            # This id is the backup's identity: save_config_atomic() returns the
+            # path, rollback_config(backup_version=...) looks the version up, and
+            # the paired secrets backup is found by reusing the same string. At
+            # second granularity two saves inside the same second produced the
+            # same filename, so the second copy2() below silently overwrote the
+            # first backup -- the path a caller was still holding then pointed at
+            # different content, and rolling back to it restored the wrong
+            # config. Microseconds make that collision vanishingly unlikely.
             config_name = self.config_path.name
-            backup_filename = f"{config_name}.backup.{timestamp}"
+            backup_secrets = bool(self.secrets_path and self.secrets_path.exists())
-            backup_path = self.backup_dir / backup_filename
+
-            
+            # exists() then copy2() is two steps: two concurrent callers can
+            # both see the path as free and pick the same one, so the second
+            # copy2() silently destroys the first call's restore point.
+            # Reserve the filename(s) with exclusive creation instead -- that
+            # is atomic, so only one caller can ever win a given timestamp.
+            # Each retry bumps the collision suffix, so this always
+            # terminates and stays compatible with _parse_backup_version().
+            collision = 0
+            while True:
+                timestamp = datetime.now().strftime(BACKUP_VERSION_FORMAT)
+                if collision:
+                    timestamp = f"{timestamp}-{collision}"
+                backup_path = self.backup_dir / f"{config_name}.backup.{timestamp}"
+                secrets_backup_path = (
+                    self.backup_dir / f"{self.secrets_path.name}.backup.{timestamp}"
+                    if backup_secrets else None
+                )
+                try:
+                    backup_path.touch(exist_ok=False)
+                except FileExistsError:
+                    collision += 1
+                    continue
+                if secrets_backup_path is not None:
+                    try:
+                        secrets_backup_path.touch(exist_ok=False)
+                    except FileExistsError:
+                        backup_path.unlink(missing_ok=True)
+                        collision += 1
+                        continue
+                break
+
             # Copy config file to backup
             shutil.copy2(self.config_path, backup_path)
-            
+
             # Also backup secrets file if it exists
-            if self.secrets_path and self.secrets_path.exists():
+            if secrets_backup_path is not None:
-                secrets_backup_filename = f"{self.secrets_path.name}.backup.{timestamp}"
-                secrets_backup_path = self.backup_dir / secrets_backup_filename
                 shutil.copy2(self.secrets_path, secrets_backup_path)
             
             # Rotate old backups

test/conftest.py

@@ -7,7 +7,7 @@ Provides common fixtures for mocking core components and test setup.
 import pytest
 import sys
 from pathlib import Path
-from unittest.mock import Mock, MagicMock
+from unittest.mock import Mock, MagicMock, NonCallableMock
 from typing import Dict, Any, Optional
 
 # Add project root to path
@@ -70,6 +70,61 @@ def pytest_unconfigure(config):
         shutil.rmtree(tmp_dir, ignore_errors=True)
 
 
+# DisplayManager is a process-wide singleton, and the RGBMatrix /
+# RGBMatrixOptions names it builds through are module globals bound once at
+# import to either the emulator or the hardware library. All three are shared
+# by every test module in the run, so a module that leaves a live instance in
+# _instance -- or leaves patch('src.display_manager.RGBMatrix') standing --
+# changes what the NEXT module constructs. None of that is visible when the
+# affected file is run on its own; it surfaces as a full-suite failure that
+# does not reproduce. The full-run failure diff is how a change is confirmed
+# non-regressive, so it has to mean the same thing on every run.
+_PRISTINE_MATRIX_BINDINGS = {}
+
+
+@pytest.fixture(autouse=True, scope="module")
+def _reset_display_manager_globals():
+    """Undo any DisplayManager global state a test module leaves behind.
+
+    Autouse fixtures are set up ahead of the fixtures a test asks for, so this
+    one is finalised after them -- including after a module-scoped fixture that
+    owns a real DisplayManager (test_display_dirty_tracking.py's ``dm``).
+    Module-scoped rather than per-test: files that deliberately share one
+    manager across their own tests keep doing so; only the leak across the
+    module boundary is cut.
+    """
+    # Record the real bindings on the way in, while no patch of this module's
+    # is active yet -- reading them on the way out would record the leak.
+    _remember_matrix_bindings()
+    yield
+
+    dm_mod = sys.modules.get("src.display_manager")
+    if dm_mod is None:
+        return  # Module never imported it; nothing to reset.
+
+    # An instance left here is what the next module's DisplayManager() call
+    # gets back -- potentially one built against a MagicMock matrix.
+    dm_mod.DisplayManager._instance = None
+    dm_mod.DisplayManager._initialized = False
+
+    # Put a binding back if a patch outlived the module that started it.
+    # Restoring rather than failing: a leak reported against an innocent module
+    # later in the run is the diagnosis problem, not the fix for it.
+    for name, pristine in _PRISTINE_MATRIX_BINDINGS.items():
+        if isinstance(getattr(dm_mod, name, None), NonCallableMock):
+            setattr(dm_mod, name, pristine)
+
+
+def _remember_matrix_bindings():
+    dm_mod = sys.modules.get("src.display_manager")
+    if dm_mod is None:
+        return
+    for name in ("RGBMatrix", "RGBMatrixOptions"):
+        current = getattr(dm_mod, name, None)
+        if current is not None and not isinstance(current, NonCallableMock):
+            _PRISTINE_MATRIX_BINDINGS.setdefault(name, current)
+
+
 @pytest.fixture
 def mock_display_manager():
     """Create a mock DisplayManager for testing."""

test/test_display_dirty_tracking.py

@@ -44,6 +44,20 @@ def dm(tmp_path_factory):
     # this session; the individual tests that care still override it further.
     manager._snapshot_path = str(
         tmp_path_factory.mktemp("dirty_tracking") / "led_matrix_preview.png")
+    # _setup_matrix() swallows every construction failure and falls back to
+    # matrix=None, so a broken environment reaches the tests as fifteen
+    # identical "'NoneType' object has no attribute 'SwapOnVSync'" errors that
+    # name neither this fixture nor the real cause. Fail here instead, once,
+    # and say where to look.
+    if manager.matrix is None:
+        pytest.fail(
+            "DisplayManager fell back to matrix=None: RGBMatrix construction "
+            "raised (the 'Failed to initialize RGB Matrix' log line above "
+            "carries the reason). Known causes: the emulator adapter losing a "
+            "fixed TCP port to another process -- see pytest_configure in "
+            "test/conftest.py, which pins the port-free 'raw' adapter -- or a "
+            "patch('src.display_manager.RGBMatrix') leaked from an earlier "
+            "test module.")
     yield manager
     DisplayManager._instance = None
     DisplayManager._initialized = False

test/web_interface/integration/test_config_flows.py

@@ -83,19 +83,24 @@ class TestConfigFlowsIntegration(unittest.TestCase):
     
     def test_backup_rotation(self):
         """Test that backup rotation works correctly."""
-        max_backups = 3
+        # setUp built the manager with max_backups=5; ask it rather than
-        
+        # restating the number, which is how this drifted in the first place.
-        # Create multiple backups
+        max_backups = self.atomic_manager.max_backups
-        for i in range(5):
+
+        # Overshoot the limit so rotation actually has something to remove.
+        for i in range(max_backups + 3):
             config = {"test": f"value_{i}"}
             result = self.atomic_manager.save_config_atomic(config, create_backup=True)
             self.assertEqual(result.status, SaveResultStatus.SUCCESS)
-        
+
-        # List backups
+        # Rotation should have trimmed the excess, and kept exactly the limit.
+        #
+        # This used to assert against a hardcoded 3 while setUp configured 5,
+        # and still passed -- because every save in the loop landed in the same
+        # second and collapsed onto a single backup filename, so there was only
+        # ever one backup to count and rotation was never exercised at all.
         backups = self.atomic_manager.list_backups()
-        
+        self.assertEqual(max_backups, len(backups))
-        # Verify only max_backups are kept
-        self.assertLessEqual(len(backups), max_backups)
     
     def test_validation_failure_triggers_rollback(self):
         """Test that validation failure triggers automatic rollback."""
@@ -148,10 +153,21 @@ class TestConfigFlowsIntegration(unittest.TestCase):
             rollback_success = self.atomic_manager.rollback_config(backup_version=None)
         self.assertTrue(rollback_success)
         
-        # Verify rollback
+        # Verify rollback.
+        #
+        # result1's backup was taken *before* the first save, so it holds the
+        # config setUp wrote -- plugin1=30, plugin2=15 -- and neither change.
+        #
+        # This used to assert plugin1 == 45, a state no single backup ever held:
+        # 45 was only ever in result2's backup, and 15 only in result1's. It
+        # passed because both saves landed in the same second, so both backups
+        # were written to one filename and "result1's version" resolved to
+        # result2's content. Once the backup id became unique the rollback
+        # started returning the version actually asked for, and the assertion
+        # -- not the rollback -- was what had to change.
         rolled_back_config = self.config_manager.load_config()
-        self.assertEqual(rolled_back_config["plugin1"]["display_duration"], 45)
+        self.assertEqual(rolled_back_config["plugin1"]["display_duration"], 30)
-        self.assertEqual(rolled_back_config["plugin2"]["display_duration"], 15)  # Original value
+        self.assertEqual(rolled_back_config["plugin2"]["display_duration"], 15)
 
 
 if __name__ == '__main__':

test/web_interface/test_config_manager_atomic.py

@@ -6,8 +6,12 @@ import unittest
 import tempfile
 import shutil
 import json
+import threading
+from datetime import datetime, timedelta
 from pathlib import Path
+from unittest.mock import patch
 
+import src.config_manager_atomic as atomic_module
 from src.config_manager_atomic import AtomicConfigManager, SaveResultStatus
 
 
@@ -103,6 +107,217 @@ class TestAtomicConfigManager(unittest.TestCase):
         self.assertEqual(result.status, SaveResultStatus.SUCCESS)
 
 
+class TestBackupVersionsAreUnique(unittest.TestCase):
+    """
+    A backup's version is its identity: save_config_atomic() hands the path
+    back, rollback_config(backup_version=...) looks it up, and the paired
+    secrets backup is found by reusing the string. Two saves in the same second
+    used to produce the same filename, so the second overwrote the first and a
+    rollback to the earlier version restored the later content.
+    """
+
+    def setUp(self):
+        self.temp_dir = Path(tempfile.mkdtemp())
+        self.config_path = self.temp_dir / "config.json"
+        self.backup_dir = self.temp_dir / "backups"
+        self.config_path.write_text(json.dumps({"duration": 15}))
+        self.manager = AtomicConfigManager(
+            config_path=str(self.config_path),
+            backup_dir=str(self.backup_dir),
+            max_backups=10,
+        )
+
+    def tearDown(self):
+        shutil.rmtree(self.temp_dir)
+
+    def _version_of(self, backup_path):
+        return Path(backup_path).name.split('.backup.', 1)[-1]
+
+    def test_two_backups_in_the_same_second_are_two_files(self):
+        first = self.manager._create_backup()
+        self.config_path.write_text(json.dumps({"duration": 99}))
+        second = self.manager._create_backup()
+
+        self.assertNotEqual(first, second)
+        self.assertEqual(2, len(list(self.backup_dir.glob("config.json.backup.*"))))
+
+    def test_a_later_backup_does_not_overwrite_an_earlier_one(self):
+        first = self.manager._create_backup()
+        self.config_path.write_text(json.dumps({"duration": 99}))
+        self.manager._create_backup()
+
+        # The path the caller is still holding must hold what was backed up.
+        self.assertEqual({"duration": 15}, json.loads(Path(first).read_text()))
+
+    def test_rollback_to_an_earlier_version_restores_that_version(self):
+        result1 = self.manager.save_config_atomic({"duration": 45}, create_backup=True)
+        self.manager.save_config_atomic({"duration": 20}, create_backup=True)
+
+        # result1's backup was taken before that save, so it holds duration 15.
+        self.assertTrue(
+            self.manager.rollback_config(
+                backup_version=self._version_of(result1.backup_path)
+            )
+        )
+        self.assertEqual({"duration": 15}, json.loads(self.config_path.read_text()))
+
+    def test_the_reported_version_is_the_one_in_the_filename(self):
+        # rollback_config() matches on this string, so it has to be the name on
+        # disk and not a restamp of the file's mtime.
+        self.manager._create_backup()
+        self.config_path.write_text(json.dumps({"duration": 99}))
+        self.manager._create_backup()
+
+        backups = self.manager.list_backups()
+        self.assertEqual(2, len(backups))
+        for backup in backups:
+            self.assertEqual(self._version_of(backup.path), backup.version)
+        self.assertEqual(2, len({b.version for b in backups}))
+
+    def test_backups_are_ordered_newest_first_within_the_same_second(self):
+        older = self.manager._create_backup()
+        newer = self.manager._create_backup()
+
+        listed = [b.path for b in self.manager.list_backups()]
+        self.assertEqual([newer, older], listed)
+
+    def test_a_legacy_second_granularity_backup_is_still_restorable(self):
+        # Backups written before the version gained microseconds must stay
+        # usable, or an upgrade silently strips a rig's restore points.
+        legacy = self.backup_dir / "config.json.backup.20240101_120000"
+        legacy.write_text(json.dumps({"duration": 7}))
+
+        versions = {b.version for b in self.manager.list_backups()}
+        self.assertIn("20240101_120000", versions)
+
+        self.assertTrue(self.manager.rollback_config(backup_version="20240101_120000"))
+        self.assertEqual({"duration": 7}, json.loads(self.config_path.read_text()))
+
+    def test_an_unrecognized_backup_name_is_still_listed_and_restorable(self):
+        # Hand-copied or renamed files get ordered by mtime, but keep the
+        # version string they have on disk so they can still be named.
+        odd = self.backup_dir / "config.json.backup.hand-copied"
+        odd.write_text(json.dumps({"duration": 3}))
+
+        self.assertIn("hand-copied", {b.version for b in self.manager.list_backups()})
+        self.assertTrue(self.manager.rollback_config(backup_version="hand-copied"))
+        self.assertEqual({"duration": 3}, json.loads(self.config_path.read_text()))
+
+    def test_a_collision_suffix_never_costs_a_restore_point(self):
+        # Freeze the clock so every backup wants the identical filename.
+        frozen = datetime(2026, 1, 2, 3, 4, 5, 678901)
+
+        class FrozenDatetime(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return frozen
+
+        paths = []
+        with patch.object(atomic_module, 'datetime', FrozenDatetime):
+            for i in range(3):
+                self.config_path.write_text(json.dumps({"duration": i}))
+                paths.append(self.manager._create_backup())
+
+        self.assertEqual(3, len(set(paths)))
+        self.assertEqual(
+            [0, 1, 2],
+            [json.loads(Path(p).read_text())["duration"] for p in paths],
+        )
+
+    def test_collision_suffixed_backups_list_newest_first(self):
+        # All three share one frozen timestamp and differ only by their -N
+        # suffix. Unless N is folded into the sort key, they parse to the
+        # identical datetime and a stable sort leaves them in whatever order
+        # the filesystem glob happened to yield -- not necessarily creation
+        # order.
+        frozen = datetime(2026, 1, 2, 3, 4, 5, 678901)
+
+        class FrozenDatetime(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return frozen
+
+        created = []
+        with patch.object(atomic_module, 'datetime', FrozenDatetime):
+            for i in range(3):
+                self.config_path.write_text(json.dumps({"duration": i}))
+                created.append(self.manager._create_backup())
+
+        listed = [b.path for b in self.manager.list_backups()]
+        self.assertEqual(list(reversed(created)), listed)
+
+    def test_concurrent_backups_at_the_same_instant_never_collide(self):
+        # exists() followed by copy2() is two steps: two threads can both see
+        # a path as free before either creates it, and the second copy2()
+        # then silently destroys the first thread's restore point. Freezing
+        # the clock forces every thread to want the same filename.
+        frozen = datetime(2026, 3, 4, 5, 6, 7, 890123)
+
+        class FrozenDatetime(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return frozen
+
+        secrets_path = self.temp_dir / "secrets.json"
+        secrets_path.write_text(json.dumps({"token": "abc"}))
+        manager = AtomicConfigManager(
+            config_path=str(self.config_path),
+            secrets_path=str(secrets_path),
+            backup_dir=str(self.backup_dir),
+            max_backups=50,
+        )
+
+        n_threads = 8
+        barrier = threading.Barrier(n_threads)
+        results = [None] * n_threads
+
+        def worker(i):
+            barrier.wait()
+            results[i] = manager._create_backup()
+
+        with patch.object(atomic_module, 'datetime', FrozenDatetime):
+            threads = [threading.Thread(target=worker, args=(i,)) for i in range(n_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+
+        self.assertNotIn(None, results)
+        self.assertEqual(n_threads, len(set(results)))
+
+        config_backups = list(self.backup_dir.glob("config.json.backup.*"))
+        secrets_backups = list(self.backup_dir.glob("secrets.json.backup.*"))
+        self.assertEqual(n_threads, len(config_backups))
+        self.assertEqual(n_threads, len(secrets_backups))
+        for backup in config_backups:
+            self.assertEqual({"duration": 15}, json.loads(backup.read_text()))
+
+
+class TestParseBackupVersion(unittest.TestCase):
+    """
+    _parse_backup_version() only recognizes its own ``-N`` collision suffix
+    when N is numeric, and folds N into the returned timestamp so
+    same-instant collisions still sort deterministically.
+    """
+
+    def test_numeric_collision_suffix_is_folded_into_the_timestamp(self):
+        base = "20260102_030405_678901"
+        parsed_base = AtomicConfigManager._parse_backup_version(base)
+        parsed_collided = AtomicConfigManager._parse_backup_version(f"{base}-2")
+        self.assertEqual(parsed_base + timedelta(microseconds=2), parsed_collided)
+
+    def test_a_nonnumeric_suffix_is_not_mistaken_for_a_collision_marker(self):
+        # "-manual" is not a suffix this class ever writes. Stripping it
+        # anyway would either misparse the base or silently pick the wrong
+        # timestamp for a hand-named backup that happens to end in a
+        # dash-word.
+        self.assertIsNone(
+            AtomicConfigManager._parse_backup_version(
+                "20260102_030405_678901-manual"
+            )
+        )
+
+
 if __name__ == '__main__':
     unittest.main()