test(backend): prove concurrent shutdown flush is idempotent (#655)

Issue: #655
User-Visible: no

scripts/mutation-registry.mjs

@@ -13041,6 +13041,19 @@ const MUTANT_DEFINITIONS = [
       replace: '            self._save_task = asyncio.create_task(self._delayed_save())\n',
     }],
   },
+  {
+    id: 'virtual-light-concurrent-flushes-bypass-lock',
+    guard: 'node scripts/backend-test-guard.mjs '
+      + 'concurrent_flushes_after_failed_delayed_save_write_once '
+      + 'tests_backend/test_virtual_lights.py',
+    because: '#655 AC3 / review r1: stop and config transitions may flush the same dirty '
+      + 'post-failure state concurrently; the lock must keep the Store write idempotent',
+    patches: [{
+      file: 'custom_components/houseplan/virtual_lights.py',
+      find: '        async with self._flush_lock:\n',
+      replace: '        if True:  # mutant: concurrent flushes are not serialised\n',
+    }],
+  },
 ];
 
 const mutationCardSource = readFileSync(join(repoRoot, 'src/houseplan-card.ts'), 'utf8');

tests_backend/test_virtual_lights.py

@@ -126,3 +126,38 @@ def test_runtime_controller_coalesces_rapid_toggles_into_one_durable_write():
     assert second == {"marker_id": "lamp", "on": True, "rev": 2}
     assert store.writes == [{"rev": 2, "config_rev": 1, "off": []}]
     assert hass.created_tasks == 1, "the delayed writer must be tracked by HA"
+
+
+def test_concurrent_flushes_after_failed_delayed_save_write_once():
+    class BlockingStore(FakeStore):
+        def __init__(self):
+            super().__init__()
+            self.entered = 0
+            self.release = asyncio.Event()
+
+        async def async_save(self, data):
+            self.entered += 1
+            await self.release.wait()
+            await super().async_save(data)
+
+    async def exercise():
+        store = BlockingStore()
+        controller = VirtualLightController(FakeHass(), store)
+        # This is the exact state left by a failed delayed save: the newest
+        # payload is dirty, while its completed task has already detached.
+        controller._state = {"rev": 1, "config_rev": 1, "off": ["lamp"]}
+        controller._dirty = True
+        controller._save_task = None
+
+        first = asyncio.create_task(controller.async_flush())
+        second = asyncio.create_task(controller.async_flush())
+        for _ in range(3):
+            await asyncio.sleep(0)
+        entered_before_release = store.entered
+        store.release.set()
+        await asyncio.gather(first, second)
+        return entered_before_release, store.writes
+
+    entered, writes = _run(exercise())
+    assert entered == 1, "only one concurrent flush may reach durable storage"
+    assert writes == [{"rev": 1, "config_rev": 1, "off": ["lamp"]}]