diff --git a/custom_components/keymaster/coordinator.py b/custom_components/keymaster/coordinator.py
index 7d4479b9..6fc62b20 100644
--- a/custom_components/keymaster/coordinator.py
+++ b/custom_components/keymaster/coordinator.py
@@ -4,8 +4,9 @@
 
 import asyncio
 import base64
-from collections.abc import Callable, Iterable, MutableMapping
+from collections.abc import AsyncIterator, Callable, Iterable, MutableMapping
 import contextlib
+from contextlib import asynccontextmanager
 from dataclasses import fields, is_dataclass
 from datetime import datetime as dt, time as dt_time, timedelta
 import functools
@@ -223,9 +224,9 @@ def __init__(self, hass: HomeAssistant) -> None:
         self._throttle = Throttle()
         self._last_unlock_code_slot: dict[str, int | None] = {}
         self._sync_status_counter: int = 0
-        self._quick_refresh: bool = False
-        self._cancel_quick_refresh: Callable | None = None
-        self._cancel_debounced_refresh: Callable | None = None
+        self._quick_refresh_entry_ids: set[str] = set()
+        self._cancel_quick_refresh: dict[str, Callable] = {}
+        self._cancel_debounced_refresh: dict[str, Callable] = {}
         self._pending_keypad_unlock_notifications: dict[str, Callable[[], None]] = {}
         self._state_change_autolock_started: set[str] = set()
         self._pending_provider_unlock_event: set[str] = set()
@@ -241,6 +242,9 @@ def __init__(self, hass: HomeAssistant) -> None:
         self._notify_handle: asyncio.Handle | None = None
         self._refresh_dirty_entry_ids: set[str] | None = None
         self._active_refresh_count = 0
+        self._sync_tx_depth: int = 0
+        self._sync_tx_dirty_ids: set[str] = set()
+        self._sync_tx_all_entry_ids: bool = False
         self._refresh_previous_update_success: bool | None = None
         self._externally_dirty_entry_ids: set[str] = set()
         self._refresh_keepalive_unsub: Callable[[], None] | None = None
@@ -302,11 +306,13 @@ async def async_shutdown(self) -> None:
                 self._refresh_keepalive_unsub()
                 self._refresh_keepalive_unsub = None
             if self._cancel_quick_refresh:
-                self._cancel_quick_refresh()
-                self._cancel_quick_refresh = None
+                for cancel in self._cancel_quick_refresh.values():
+                    cancel()
+                self._cancel_quick_refresh.clear()
             if self._cancel_debounced_refresh:
-                self._cancel_debounced_refresh()
-                self._cancel_debounced_refresh = None
+                for cancel in self._cancel_debounced_refresh.values():
+                    cancel()
+                self._cancel_debounced_refresh.clear()
             await super().async_shutdown()
             self._shutdown_complete = True
 
@@ -387,6 +393,27 @@ def async_remove_lock_coordinator(self, config_entry_id: str) -> None:
             self._refresh_keepalive_unsub()
             self._refresh_keepalive_unsub = None
 
+    @asynccontextmanager
+    async def _parent_sync_transaction(self) -> AsyncIterator[None]:
+        """Suppress per-slot notifications and flush once on exit.
+
+        Re-entrant: increments a depth counter so nested transactions
+        share the same accumulator; only the outermost exit flushes.
+        Exception-safe: depth is decremented in a finally block.
+        """
+        self._sync_tx_depth += 1
+        try:
+            yield
+        finally:
+            self._sync_tx_depth -= 1
+            if self._sync_tx_depth == 0:
+                dirty = self._sync_tx_dirty_ids
+                all_ids = self._sync_tx_all_entry_ids
+                self._sync_tx_dirty_ids = set()
+                self._sync_tx_all_entry_ids = False
+                if dirty or all_ids:
+                    self.async_schedule_keymaster_notifications(dirty, all_entry_ids=all_ids)
+
     @callback
     def async_schedule_keymaster_notifications(
         self, entry_ids: Iterable[str], *, all_entry_ids: bool = False
@@ -396,6 +423,13 @@ def async_schedule_keymaster_notifications(
             return
 
         valid = {entry_id for entry_id in entry_ids if entry_id in self.kmlocks}
+
+        # Inside a parent-sync transaction: accumulate and defer
+        if self._sync_tx_depth > 0:
+            self._sync_tx_dirty_ids |= valid
+            self._sync_tx_all_entry_ids |= all_entry_ids
+            return
+
         if not valid and not all_entry_ids and not self._pending_notify_entry_ids:
             return
         self._pending_notify_entry_ids |= valid
@@ -1916,6 +1950,7 @@ async def _delete_lock(self, kmlock: KeymasterLock, _: dt) -> None:
         self._pending_provider_unlock_event.discard(kmlock.keymaster_config_entry_id)
         self._pending_provider_lock_event.discard(kmlock.keymaster_config_entry_id)
         self._cancel_pending_keypad_unlock_notification(kmlock)
+        self._cancel_entry_refresh_timers(kmlock.keymaster_config_entry_id)
         await self._rebuild_lock_relationships()
         await self._async_save_data()
         await self.async_refresh()
@@ -2048,7 +2083,7 @@ async def set_pin_on_lock(
 
         kmlock.code_slots[code_slot_num].synced = Synced.ADDING
         kmlock.code_slots[code_slot_num].sync_op_started_at = utcnow()
-        self._quick_refresh = True
+        self._quick_refresh_entry_ids.add(kmlock.keymaster_config_entry_id)
         # Defer notifying entities of sync status change
         self.async_schedule_keymaster_notifications([kmlock.keymaster_config_entry_id])
 
@@ -2130,7 +2165,7 @@ async def clear_pin_from_lock(
 
         kmlock.code_slots[code_slot_num].synced = Synced.DELETING
         kmlock.code_slots[code_slot_num].sync_op_started_at = utcnow()
-        self._quick_refresh = True
+        self._quick_refresh_entry_ids.add(kmlock.keymaster_config_entry_id)
         # Defer notifying entities of sync status change
         self.async_schedule_keymaster_notifications([kmlock.keymaster_config_entry_id])
 
@@ -2286,33 +2321,41 @@ async def _is_slot_active(kmslot: KeymasterCodeSlot) -> bool:
 
         return True
 
-    async def _trigger_quick_refresh(self, _: dt) -> None:
-        await self.async_request_refresh()
+    async def _trigger_quick_refresh_for_entry(self, entry_id: str, _: dt) -> None:
+        """Trigger a scoped quick-refresh for a single lock entry."""
+        self._cancel_quick_refresh.pop(entry_id, None)
+        if entry_id in self.kmlocks:
+            await self.async_refresh_lock(entry_id)
 
     async def async_request_debounced_refresh(self, entry_id: str | None = None) -> None:
         """Request a debounced coordinator refresh.
 
         Batches rapid entity updates into a single refresh by cancelling
-        any previously scheduled debounced refresh and scheduling a new
-        one after ENTITY_DEBOUNCE_SECONDS.
+        any previously scheduled debounced refresh for the same entry and
+        scheduling a new one after ENTITY_DEBOUNCE_SECONDS.
         """
         if entry_id is not None and entry_id in self.kmlocks:
             self._externally_dirty_entry_ids.add(entry_id)
 
-        if self._cancel_debounced_refresh:
-            self._cancel_debounced_refresh()
-            self._cancel_debounced_refresh = None
+        target = entry_id or "_global"
+
+        if target in self._cancel_debounced_refresh:
+            self._cancel_debounced_refresh[target]()
+            del self._cancel_debounced_refresh[target]
 
-        self._cancel_debounced_refresh = async_call_later(
+        self._cancel_debounced_refresh[target] = async_call_later(
             hass=self.hass,
             delay=ENTITY_DEBOUNCE_SECONDS,
-            action=self._trigger_debounced_refresh,
+            action=functools.partial(self._trigger_debounced_refresh_for_entry, target),
         )
 
-    async def _trigger_debounced_refresh(self, _: dt | None) -> None:
-        """Trigger a debounced refresh."""
-        self._cancel_debounced_refresh = None
-        await self.async_request_refresh()
+    async def _trigger_debounced_refresh_for_entry(self, entry_id: str, _: dt | None) -> None:
+        """Trigger a debounced refresh for a single entry."""
+        self._cancel_debounced_refresh.pop(entry_id, None)
+        if entry_id == "_global" or entry_id not in self.kmlocks:
+            await self.async_request_refresh()
+        else:
+            await self.async_refresh_lock(entry_id)
 
     async def update_slot_active_state(self, config_entry_id: str, code_slot_num: int) -> bool:
         """Update the active state for a code slot."""
@@ -2484,19 +2527,19 @@ async def _async_refresh_lock_data(
     ) -> set[str]:
         """Refresh one lock's data and return dirty entry IDs."""
         await self._initial_setup_done_event.wait()
-        self._quick_refresh = False
+        self._quick_refresh_entry_ids.discard(entry_id)
         if advance_sync_status:
             self._sync_status_counter += 1
-        await self._clear_pending_quick_refresh()
+        await self._clear_pending_quick_refresh(entry_id)
         dirty = (
             {entry_id}
             if entry_id in self.kmlocks and entry_id in self._externally_dirty_entry_ids
             else set()
         )
         self._externally_dirty_entry_ids.discard(entry_id)
-        if self._cancel_debounced_refresh and not self._externally_dirty_entry_ids:
-            self._cancel_debounced_refresh()
-            self._cancel_debounced_refresh = None
+        if entry_id in self._cancel_debounced_refresh:
+            self._cancel_debounced_refresh[entry_id]()
+            del self._cancel_debounced_refresh[entry_id]
         before = self._lock_snapshot(entry_id)
         await self._update_lock_data(entry_id)
         if before != self._lock_snapshot(entry_id):
@@ -2530,7 +2573,7 @@ async def _async_refresh_lock_data(
     async def async_refresh_all_locks(self) -> set[str]:
         """Refresh all locks for startup/maintenance; return dirty entry IDs."""
         await self._initial_setup_done_event.wait()
-        self._quick_refresh = False
+        self._quick_refresh_entry_ids.clear()
         self._sync_status_counter += 1
         dirty_entry_ids: set[str] = {
             entry_id for entry_id in self._externally_dirty_entry_ids if entry_id in self.kmlocks
@@ -2539,9 +2582,9 @@ async def async_refresh_all_locks(self) -> set[str]:
 
         # Clear any pending refresh callbacks
         await self._clear_pending_quick_refresh()
-        if self._cancel_debounced_refresh:
-            self._cancel_debounced_refresh()
-            self._cancel_debounced_refresh = None
+        for cancel in self._cancel_debounced_refresh.values():
+            cancel()
+        self._cancel_debounced_refresh.clear()
 
         for keymaster_config_entry_id in self.kmlocks:
             before = self._lock_snapshot(keymaster_config_entry_id)
@@ -2580,11 +2623,32 @@ async def _async_update_data(self) -> dict[str, Any]:
         self._record_refresh_dirty_entry_ids(await self.async_refresh_all_locks())
         return dict(self.kmlocks)
 
-    async def _clear_pending_quick_refresh(self) -> None:
-        """Clear any pending refresh callback."""
-        if self._cancel_quick_refresh:
-            self._cancel_quick_refresh()
-            self._cancel_quick_refresh = None
+    async def _clear_pending_quick_refresh(self, entry_id: str | None = None) -> None:
+        """Clear pending quick-refresh callback(s).
+
+        If entry_id is given, cancel only that entry's timer.
+        If None, cancel ALL pending quick-refresh timers.
+        """
+        if entry_id is not None:
+            cancel = self._cancel_quick_refresh.pop(entry_id, None)
+            if cancel:
+                cancel()
+        else:
+            for cancel in self._cancel_quick_refresh.values():
+                cancel()
+            self._cancel_quick_refresh.clear()
+
+    @callback
+    def _cancel_entry_refresh_timers(self, entry_id: str) -> None:
+        """Cancel quick-refresh and debounced-refresh timers for a single entry."""
+        cancel = self._cancel_quick_refresh.pop(entry_id, None)
+        if cancel:
+            cancel()
+        cancel = self._cancel_debounced_refresh.pop(entry_id, None)
+        if cancel:
+            cancel()
+        self._quick_refresh_entry_ids.discard(entry_id)
+        self._externally_dirty_entry_ids.discard(entry_id)
 
     async def _update_lock_data(self, keymaster_config_entry_id: str) -> None:
         """Update a single keymaster lock."""
@@ -2909,7 +2973,7 @@ async def _sync_pin(self, kmlock: KeymasterLock, code_slot_num: int, usercode: s
                 )
                 return
             slot.synced = Synced.OUT_OF_SYNC
-            self._quick_refresh = True
+            self._quick_refresh_entry_ids.add(kmlock.keymaster_config_entry_id)
             return
 
         # Don't import stale lock codes during grace period after a clear.
@@ -2980,11 +3044,12 @@ async def _sync_child_locks(self, keymaster_config_entry_id: str) -> set[str]:
         ):
             return dirty_entry_ids
 
-        for child_entry_id in kmlock.child_config_entry_ids:
-            before = self._lock_snapshot(child_entry_id)
-            await self._sync_child_lock(kmlock, child_entry_id)
-            if before != self._lock_snapshot(child_entry_id):
-                dirty_entry_ids.add(child_entry_id)
+        async with self._parent_sync_transaction():
+            for child_entry_id in kmlock.child_config_entry_ids:
+                before = self._lock_snapshot(child_entry_id)
+                await self._sync_child_lock(kmlock, child_entry_id)
+                if before != self._lock_snapshot(child_entry_id):
+                    dirty_entry_ids.add(child_entry_id)
         return dirty_entry_ids
 
     async def _sync_child_lock(self, kmlock: KeymasterLock, child_entry_id: str) -> None:
@@ -3097,7 +3162,7 @@ async def _update_child_code_slots(
                 or prev_active != child_slot.active
                 or child_needs_retry
             ):
-                self._quick_refresh = True
+                self._quick_refresh_entry_ids.add(child_kmlock.keymaster_config_entry_id)
                 if not kmslot.enabled or not kmslot.active or not kmslot.pin:
                     await self.clear_pin_from_lock(
                         config_entry_id=child_kmlock.keymaster_config_entry_id,
@@ -3116,14 +3181,19 @@ async def _update_child_code_slots(
                     child_kmlock.code_slots[code_slot_num].pin = kmslot.pin
 
     async def _schedule_quick_refresh_if_needed(self) -> None:
-        """Schedule quick refresh if required."""
-        if self._quick_refresh:
+        """Schedule per-entry quick refresh timers for pending entries."""
+        for entry_id in list(self._quick_refresh_entry_ids):
+            if entry_id in self._cancel_quick_refresh:
+                # Already has a pending timer — don't reschedule
+                continue
             _LOGGER.debug(
-                "[schedule_quick_refresh_if_needed] Scheduling refresh in %s seconds",
+                "[schedule_quick_refresh_if_needed] %s: Scheduling refresh in %s seconds",
+                entry_id,
                 QUICK_REFRESH_SECONDS,
             )
-            self._cancel_quick_refresh = async_call_later(
+            self._cancel_quick_refresh[entry_id] = async_call_later(
                 hass=self.hass,
                 delay=QUICK_REFRESH_SECONDS,
-                action=self._trigger_quick_refresh,
+                action=functools.partial(self._trigger_quick_refresh_for_entry, entry_id),
             )
+        self._quick_refresh_entry_ids.clear()
diff --git a/tests/test_coordinator.py b/tests/test_coordinator.py
index 784d0206..c7c210bb 100644
--- a/tests/test_coordinator.py
+++ b/tests/test_coordinator.py
@@ -3051,7 +3051,7 @@ def sync_coordinator(self, mock_hass):
             coordinator = KeymasterCoordinator(mock_hass)
             coordinator.hass = mock_hass
             coordinator.kmlocks = {}
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             coordinator.clear_pin_from_lock = AsyncMock()
             coordinator.set_pin_on_lock = AsyncMock()
             return coordinator
@@ -3146,7 +3146,7 @@ def sync_coordinator(self, mock_hass):
             coordinator.kmlocks = {}
             coordinator.set_pin_on_lock = AsyncMock()
             coordinator.clear_pin_from_lock = AsyncMock()
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             return coordinator
 
     @staticmethod
@@ -3340,16 +3340,16 @@ async def test_async_shutdown(hass: HomeAssistant) -> None:
     mock_cancel_quick = Mock()
     mock_cancel_debounced = Mock()
 
-    coordinator._cancel_quick_refresh = mock_cancel_quick
-    coordinator._cancel_debounced_refresh = mock_cancel_debounced
+    coordinator._cancel_quick_refresh = {"entry_1": mock_cancel_quick}
+    coordinator._cancel_debounced_refresh = {"entry_1": mock_cancel_debounced}
 
     await coordinator.async_shutdown()
 
     mock_cancel_quick.assert_called_once()
     mock_cancel_debounced.assert_called_once()
 
-    assert coordinator._cancel_quick_refresh is None
-    assert coordinator._cancel_debounced_refresh is None
+    assert coordinator._cancel_quick_refresh == {}
+    assert coordinator._cancel_debounced_refresh == {}
 
 
 async def test_failed_refresh_deferred_notify_preserves_failure_state(
diff --git a/tests/test_coordinator_code_import.py b/tests/test_coordinator_code_import.py
index 02130789..10eba47d 100644
--- a/tests/test_coordinator_code_import.py
+++ b/tests/test_coordinator_code_import.py
@@ -37,7 +37,7 @@ def mock_coordinator(mock_hass):
         coordinator = KeymasterCoordinator(mock_hass)
         coordinator.hass = mock_hass
         coordinator.kmlocks = {}
-        coordinator._quick_refresh = False
+        coordinator._quick_refresh_entry_ids = set()
         coordinator.set_pin_on_lock = AsyncMock()
         coordinator.clear_pin_from_lock = AsyncMock()
         coordinator.async_set_updated_data = Mock()
@@ -286,7 +286,7 @@ def real_coordinator(self, mock_hass):
             coordinator = KeymasterCoordinator(mock_hass)
             coordinator.hass = mock_hass
             coordinator.kmlocks = {}
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             coordinator._initial_setup_done_event = AsyncMock()
             coordinator._initial_setup_done_event.wait = AsyncMock()
             coordinator.async_set_updated_data = Mock()
diff --git a/tests/test_coordinator_lifecycle.py b/tests/test_coordinator_lifecycle.py
index e1690551..68b09ad3 100644
--- a/tests/test_coordinator_lifecycle.py
+++ b/tests/test_coordinator_lifecycle.py
@@ -409,8 +409,8 @@ async def test_shutdown_cleanup_runs_when_pending_save_flush_fails(hass) -> None
     notify_handle = MagicMock()
     coordinator._pending_save_entry_ids = {"entry_1"}
     coordinator._refresh_keepalive_unsub = keepalive_unsub
-    coordinator._cancel_quick_refresh = quick_refresh_unsub
-    coordinator._cancel_debounced_refresh = debounced_refresh_unsub
+    coordinator._cancel_quick_refresh = {"entry_1": quick_refresh_unsub}
+    coordinator._cancel_debounced_refresh = {"entry_1": debounced_refresh_unsub}
     coordinator._notify_handle = notify_handle
     coordinator._pending_notify_entry_ids = {"entry_1"}
     coordinator._lock_coordinators["entry_1"] = MagicMock()
@@ -426,8 +426,8 @@ async def test_shutdown_cleanup_runs_when_pending_save_flush_fails(hass) -> None
     debounced_refresh_unsub.assert_called_once()
     assert coordinator._notify_handle is None
     assert coordinator._refresh_keepalive_unsub is None
-    assert coordinator._cancel_quick_refresh is None
-    assert coordinator._cancel_debounced_refresh is None
+    assert coordinator._cancel_quick_refresh == {}
+    assert coordinator._cancel_debounced_refresh == {}
     assert coordinator._lock_coordinators == {}
     assert coordinator._shutdown_requested is True
     assert coordinator._shutdown_complete is True
@@ -1372,7 +1372,7 @@ async def test_async_refresh_lock_cancels_debounce_and_reports_sync_status_dirty
     coordinator.kmlocks["entry_1"] = lock
     coordinator._sync_status_counter = SYNC_STATUS_THRESHOLD
     cancel_debounced_refresh = MagicMock()
-    coordinator._cancel_debounced_refresh = cancel_debounced_refresh
+    coordinator._cancel_debounced_refresh = {"entry_1": cancel_debounced_refresh}
     coordinator._update_lock_data = AsyncMock()
     coordinator._sync_child_locks = AsyncMock(return_value=set())
     coordinator._async_save_data = AsyncMock()
@@ -1392,7 +1392,7 @@ async def update_door_and_lock_state(
 
     assert dirty == {"entry_1"}
     cancel_debounced_refresh.assert_called_once()
-    assert coordinator._cancel_debounced_refresh is None
+    assert "entry_1" not in coordinator._cancel_debounced_refresh
     assert coordinator._sync_status_counter == 0
     await coordinator.async_shutdown()
 
@@ -1440,14 +1440,14 @@ async def test_async_refresh_lock_health_transition_notifies_all_locks(hass) ->
     await coordinator.async_shutdown()
 
 
-async def test_scoped_refresh_keeps_shared_debounce_for_other_dirty_locks(hass) -> None:
-    """Test scoped refresh does not strand another lock's pending debounced refresh."""
+async def test_scoped_refresh_preserves_other_entry_debounce_timer(hass) -> None:
+    """Refreshing entry_1 must not cancel entry_2's per-entry debounce timer."""
     coordinator = KeymasterCoordinator(hass)
     coordinator._initial_setup_done_event.set()
     coordinator.kmlocks["entry_1"] = _make_lock("entry_1", "lock_1")
     coordinator.kmlocks["entry_2"] = _make_lock("entry_2", "lock_2")
     cancel_debounced_refresh = MagicMock()
-    coordinator._cancel_debounced_refresh = cancel_debounced_refresh
+    coordinator._cancel_debounced_refresh = {"entry_2": cancel_debounced_refresh}
     coordinator._externally_dirty_entry_ids = {"entry_2"}
     coordinator._update_lock_data = AsyncMock()
     coordinator._sync_child_locks = AsyncMock(return_value=set())
@@ -1458,7 +1458,7 @@ async def test_scoped_refresh_keeps_shared_debounce_for_other_dirty_locks(hass)
 
     assert dirty == set()
     assert coordinator._externally_dirty_entry_ids == {"entry_2"}
-    assert coordinator._cancel_debounced_refresh is cancel_debounced_refresh
+    assert "entry_2" in coordinator._cancel_debounced_refresh
     cancel_debounced_refresh.assert_not_called()
     await coordinator.async_shutdown()
 
diff --git a/tests/test_coordinator_sync.py b/tests/test_coordinator_sync.py
index 6410a8e9..38551b75 100644
--- a/tests/test_coordinator_sync.py
+++ b/tests/test_coordinator_sync.py
@@ -30,7 +30,7 @@ def mock_coordinator(mock_hass):
         coordinator = KeymasterCoordinator(mock_hass)
         coordinator.hass = mock_hass
         coordinator.kmlocks = {}
-        coordinator._quick_refresh = False
+        coordinator._quick_refresh_entry_ids = set()
         # Mock the PIN operations
         coordinator.set_pin_on_lock = AsyncMock()
         coordinator.clear_pin_from_lock = AsyncMock()
@@ -116,7 +116,7 @@ async def test_sync_parent_disabled_slot_clears_child(
             override=True,
         )
         assert child_slot.pin is None
-        assert mock_coordinator._quick_refresh is True
+        assert mock_coordinator._quick_refresh_entry_ids == {"child_id"}
 
     async def test_sync_parent_inactive_slot_clears_child(
         self, mock_coordinator, parent_lock, child_lock
@@ -885,7 +885,7 @@ def real_coordinator(self, mock_hass):
             coordinator = KeymasterCoordinator(mock_hass)
             coordinator.hass = mock_hass
             coordinator.kmlocks = {}
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             coordinator._initial_setup_done_event = AsyncMock()
             coordinator._initial_setup_done_event.wait = AsyncMock()
             coordinator.async_set_updated_data = Mock()
@@ -1041,7 +1041,7 @@ def real_coordinator(self, mock_hass):
             coordinator = KeymasterCoordinator(mock_hass)
             coordinator.hass = mock_hass
             coordinator.kmlocks = {}
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             coordinator._initial_setup_done_event = AsyncMock()
             coordinator._initial_setup_done_event.wait = AsyncMock()
             coordinator.async_set_updated_data = Mock()
@@ -1294,7 +1294,7 @@ def coordinator_for_update(self, mock_hass):
             coordinator = KeymasterCoordinator(mock_hass)
             coordinator.hass = mock_hass
             coordinator.kmlocks = {}
-            coordinator._quick_refresh = False
+            coordinator._quick_refresh_entry_ids = set()
             coordinator.set_pin_on_lock = AsyncMock(return_value=True)
             coordinator.clear_pin_from_lock = AsyncMock(return_value=True)
             coordinator._update_slot = AsyncMock()
@@ -1444,4 +1444,196 @@ async def test_child_sync_out_of_sync_recovery_full_cycle(
             pin="5678",
             override=True,
         )
-        assert mock_coordinator._quick_refresh is True
+        assert mock_coordinator._quick_refresh_entry_ids == {"child_id"}
+
+
+class TestParentSyncTransaction:
+    """Tests for _parent_sync_transaction coalescing notifications."""
+
+    @pytest.fixture
+    def real_coordinator(self, mock_hass):
+        """Coordinator with real async_schedule_keymaster_notifications and transaction."""
+        with patch.object(KeymasterCoordinator, "__init__", return_value=None):
+            coordinator = KeymasterCoordinator(mock_hass)
+            coordinator.hass = mock_hass
+            coordinator.kmlocks = {}
+            coordinator._quick_refresh_entry_ids = set()
+            coordinator._initial_setup_done_event = AsyncMock()
+            coordinator._initial_setup_done_event.wait = AsyncMock()
+            coordinator.async_set_updated_data = Mock()
+            # Real notification method (not mocked)
+            coordinator._deferred_notifications_shutting_down = False
+            coordinator._pending_notify_entry_ids = set()
+            coordinator._pending_notify_all_entry_ids = False
+            coordinator._pending_failed_refresh = False
+            coordinator._defer_refresh_listener_updates = False
+            coordinator._notify_handle = None
+            coordinator._lock_coordinators = {}
+            coordinator._sync_tx_depth = 0
+            coordinator._sync_tx_dirty_ids = set()
+            coordinator._sync_tx_all_entry_ids = False
+            coordinator.last_update_success = True
+            # Mock schedule to track calls that escape the transaction
+            coordinator._schedule_pending_keymaster_notifications = Mock()
+            return coordinator
+
+    @pytest.fixture
+    def parent_with_children(self, real_coordinator):
+        """Set up parent with two children in kmlocks."""
+        parent = KeymasterLock(
+            lock_name="Parent",
+            lock_entity_id="lock.parent",
+            keymaster_config_entry_id="parent_id",
+        )
+        parent.connected = True
+        parent.provider = Mock(spec=BaseLockProvider)
+        parent.child_config_entry_ids = ["child_a", "child_b"]
+        parent.code_slots = {
+            1: KeymasterCodeSlot(number=1, pin="1111", name="Slot1", active=True, enabled=True),
+            2: KeymasterCodeSlot(number=2, pin="2222", name="Slot2", active=True, enabled=True),
+        }
+
+        child_a = KeymasterLock(
+            lock_name="Child A",
+            lock_entity_id="lock.child_a",
+            keymaster_config_entry_id="child_a",
+        )
+        child_a.connected = True
+        child_a.provider = Mock(spec=BaseLockProvider)
+        child_a.provider.async_set_usercode = AsyncMock(return_value=True)
+        child_a.provider.async_clear_usercode = AsyncMock(return_value=True)
+        child_a.parent_config_entry_id = "parent_id"
+        child_a.code_slots = {
+            1: KeymasterCodeSlot(number=1, pin="0000", name="Slot1", active=True, enabled=True),
+            2: KeymasterCodeSlot(number=2, pin="0000", name="Slot2", active=True, enabled=True),
+        }
+
+        child_b = KeymasterLock(
+            lock_name="Child B",
+            lock_entity_id="lock.child_b",
+            keymaster_config_entry_id="child_b",
+        )
+        child_b.connected = True
+        child_b.provider = Mock(spec=BaseLockProvider)
+        child_b.provider.async_set_usercode = AsyncMock(return_value=True)
+        child_b.provider.async_clear_usercode = AsyncMock(return_value=True)
+        child_b.parent_config_entry_id = "parent_id"
+        child_b.code_slots = {
+            1: KeymasterCodeSlot(number=1, pin="0000", name="Slot1", active=True, enabled=True),
+            2: KeymasterCodeSlot(number=2, pin="0000", name="Slot2", active=True, enabled=True),
+        }
+
+        real_coordinator.kmlocks = {
+            "parent_id": parent,
+            "child_a": child_a,
+            "child_b": child_b,
+        }
+        return parent, child_a, child_b
+
+    async def test_multiple_slots_one_child_single_notification(
+        self, real_coordinator, parent_with_children
+    ):
+        """Parent sync touching multiple slots on one child produces ONE notification."""
+        parent, child_a, _child_b = parent_with_children
+        # Only sync child_a
+        parent.child_config_entry_ids = ["child_a"]
+
+        await real_coordinator._sync_child_locks("parent_id")
+
+        # The transaction should flush exactly once, producing one schedule call
+        assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 1
+        # The pending set should contain child_a (both slots triggered it)
+        assert "child_a" in real_coordinator._pending_notify_entry_ids
+
+    async def test_multiple_children_each_notified_once(
+        self, real_coordinator, parent_with_children
+    ):
+        """Parent with multiple children notifies parent + each child exactly once."""
+        _parent, _child_a, _child_b = parent_with_children
+
+        await real_coordinator._sync_child_locks("parent_id")
+
+        # Single flush at transaction exit
+        assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 1
+        # Both children should be in the pending set
+        assert "child_a" in real_coordinator._pending_notify_entry_ids
+        assert "child_b" in real_coordinator._pending_notify_entry_ids
+
+    async def test_override_parent_child_not_notified(self, real_coordinator, parent_with_children):
+        """Children with override_parent=True are neither mutated nor notified."""
+        parent, child_a, _child_b = parent_with_children
+        parent.child_config_entry_ids = ["child_a"]
+        # Set override_parent on child slots
+        child_a.code_slots[1].override_parent = True
+        child_a.code_slots[2].override_parent = True
+
+        await real_coordinator._sync_child_locks("parent_id")
+
+        # No slots were changed, so child_a is not dirty
+        assert "child_a" not in real_coordinator._pending_notify_entry_ids
+
+    async def test_exception_mid_transaction_does_not_leak_depth(
+        self, real_coordinator, parent_with_children
+    ):
+        """An exception during provider call does not leak transaction depth."""
+        parent, child_a, _child_b = parent_with_children
+        parent.child_config_entry_ids = ["child_a"]
+        # Make the provider raise on second slot
+        call_count = {"n": 0}
+
+        async def failing_set(*args, **kwargs):
+            call_count["n"] += 1
+            if call_count["n"] >= 2:
+                raise RuntimeError("Provider exploded")
+            return True
+
+        child_a.provider.async_set_usercode = AsyncMock(side_effect=failing_set)
+
+        with pytest.raises(RuntimeError, match="Provider exploded"):
+            await real_coordinator._sync_child_locks("parent_id")
+
+        # Depth must be back to 0
+        assert real_coordinator._sync_tx_depth == 0
+        # Accumulated IDs from before the exception should have been flushed
+        assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 1
+
+        # Subsequent call must schedule immediately (depth == 0)
+        real_coordinator._schedule_pending_keymaster_notifications.reset_mock()
+        real_coordinator.async_schedule_keymaster_notifications(["child_a"])
+        assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 1
+
+    async def test_nested_transaction_reentrant(self, real_coordinator, parent_with_children):
+        """Nested transactions share the accumulator; only outermost flushes."""
+        _parent, _child_a, _child_b = parent_with_children
+
+        async with real_coordinator._parent_sync_transaction():
+            real_coordinator.async_schedule_keymaster_notifications(["parent_id"])
+            assert real_coordinator._sync_tx_depth == 1
+            assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 0
+
+            async with real_coordinator._parent_sync_transaction():
+                real_coordinator.async_schedule_keymaster_notifications(["child_a"])
+                assert real_coordinator._sync_tx_depth == 2
+                assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 0
+
+            # Inner exited but depth > 0, no flush yet
+            assert real_coordinator._sync_tx_depth == 1
+            assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 0
+
+        # Outermost exits -> flush
+        assert real_coordinator._sync_tx_depth == 0
+        assert real_coordinator._schedule_pending_keymaster_notifications.call_count == 1
+        assert "parent_id" in real_coordinator._pending_notify_entry_ids
+        assert "child_a" in real_coordinator._pending_notify_entry_ids
+
+    async def test_all_entry_ids_preserved_through_transaction(
+        self, real_coordinator, parent_with_children
+    ):
+        """all_entry_ids=True requested during transaction is honoured at flush."""
+        _parent, _child_a, _child_b = parent_with_children
+
+        async with real_coordinator._parent_sync_transaction():
+            real_coordinator.async_schedule_keymaster_notifications(["child_a"], all_entry_ids=True)
+
+        # After flush, the all_entry_ids flag should have propagated
+        assert real_coordinator._pending_notify_all_entry_ids is True
diff --git a/tests/test_debounce.py b/tests/test_debounce.py
index f5cb0e97..36684d73 100644
--- a/tests/test_debounce.py
+++ b/tests/test_debounce.py
@@ -31,6 +31,7 @@
     KeymasterLock,
 )
 from custom_components.keymaster.number import KeymasterNumber, KeymasterNumberEntityDescription
+from custom_components.keymaster.providers._base import BaseLockProvider
 from custom_components.keymaster.switch import KeymasterSwitch, KeymasterSwitchEntityDescription
 from custom_components.keymaster.text import KeymasterText, KeymasterTextEntityDescription
 from custom_components.keymaster.time import KeymasterTime, KeymasterTimeEntityDescription
@@ -57,7 +58,7 @@ def mock_coordinator(mock_hass):
         coordinator = KeymasterCoordinator(mock_hass)
         coordinator.hass = mock_hass
         coordinator.kmlocks = {}
-        coordinator._quick_refresh = False
+        coordinator._quick_refresh_entry_ids = set()
         coordinator.set_pin_on_lock = AsyncMock()
         coordinator.clear_pin_from_lock = AsyncMock()
         return coordinator
@@ -95,7 +96,7 @@ async def test_mismatch_suppressed_during_grace_window(self, mock_coordinator):
 
         assert slot.pin == "5678"
         assert slot.synced == Synced.SYNCED
-        assert mock_coordinator._quick_refresh is False
+        assert "entry_1" not in mock_coordinator._quick_refresh_entry_ids
 
     async def test_mismatch_detected_after_grace_expires(self, mock_coordinator):
         """PIN mismatch after grace period should mark OUT_OF_SYNC."""
@@ -109,7 +110,7 @@ async def test_mismatch_detected_after_grace_expires(self, mock_coordinator):
         await mock_coordinator._sync_pin(lock, 1, "1234")
 
         assert slot.synced == Synced.OUT_OF_SYNC
-        assert mock_coordinator._quick_refresh is True
+        assert mock_coordinator._quick_refresh_entry_ids == {"entry_1"}
         assert slot.pin == "5678"
 
     async def test_mismatch_detected_when_no_grace_timestamp(self, mock_coordinator):
@@ -124,7 +125,7 @@ async def test_mismatch_detected_when_no_grace_timestamp(self, mock_coordinator)
         await mock_coordinator._sync_pin(lock, 1, "1234")
 
         assert slot.synced == Synced.OUT_OF_SYNC
-        assert mock_coordinator._quick_refresh is True
+        assert mock_coordinator._quick_refresh_entry_ids == {"entry_1"}
 
     async def test_empty_lock_response_during_grace_does_not_repush(self, mock_coordinator):
         """Lock reports empty during grace period — should NOT re-push PIN."""
@@ -165,7 +166,7 @@ async def test_matching_pin_updates_normally(self, mock_coordinator):
 
         assert slot.synced == Synced.SYNCED
         assert slot.pin == "5678"
-        assert mock_coordinator._quick_refresh is False
+        assert "entry_1" not in mock_coordinator._quick_refresh_entry_ids
 
     async def test_stale_code_after_clear_within_grace(self, mock_coordinator):
         """Lock reporting old PIN after clear within grace should not overwrite cleared state."""
@@ -252,32 +253,34 @@ class TestDebouncedRefresh:
     async def test_multiple_rapid_calls_result_in_single_refresh(self, hass: HomeAssistant):
         """Multiple rapid calls should schedule only one refresh."""
         coordinator = KeymasterCoordinator(hass)
-        coordinator._cancel_debounced_refresh = None
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
 
-        with patch.object(coordinator, "async_request_refresh", new=AsyncMock()) as mock_refresh:
-            await coordinator.async_request_debounced_refresh()
-            await coordinator.async_request_debounced_refresh()
-            await coordinator.async_request_debounced_refresh()
+        with patch.object(
+            coordinator, "async_refresh_lock", new=AsyncMock(return_value=set())
+        ) as mock_refresh:
+            await coordinator.async_request_debounced_refresh("entry_1")
+            await coordinator.async_request_debounced_refresh("entry_1")
+            await coordinator.async_request_debounced_refresh("entry_1")
 
-            assert coordinator._cancel_debounced_refresh is not None
+            assert "entry_1" in coordinator._cancel_debounced_refresh
 
             # Cancel the pending async_call_later timer before manually triggering
-            coordinator._cancel_debounced_refresh()
-            await coordinator._trigger_debounced_refresh(utcnow())
+            coordinator._cancel_debounced_refresh["entry_1"]()
+            await coordinator._trigger_debounced_refresh_for_entry("entry_1", utcnow())
 
-            mock_refresh.assert_called_once()
+            mock_refresh.assert_called_once_with("entry_1")
 
     async def test_cancels_previous_pending(self, hass: HomeAssistant):
         """Calling again should cancel the previous pending refresh."""
         coordinator = KeymasterCoordinator(hass)
-        coordinator._cancel_debounced_refresh = None
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
 
-        await coordinator.async_request_debounced_refresh()
-        first_cancel = coordinator._cancel_debounced_refresh
+        await coordinator.async_request_debounced_refresh("entry_1")
+        first_cancel = coordinator._cancel_debounced_refresh.get("entry_1")
         assert first_cancel is not None
 
-        await coordinator.async_request_debounced_refresh()
-        second_cancel = coordinator._cancel_debounced_refresh
+        await coordinator.async_request_debounced_refresh("entry_1")
+        second_cancel = coordinator._cancel_debounced_refresh.get("entry_1")
         assert second_cancel is not None
         assert first_cancel is not second_cancel
 
@@ -285,23 +288,23 @@ async def test_cancels_previous_pending(self, hass: HomeAssistant):
         second_cancel()
 
     async def test_trigger_debounced_refresh_clears_cancel(self, hass: HomeAssistant):
-        """_trigger_debounced_refresh should clear _cancel_debounced_refresh."""
+        """_trigger_debounced_refresh_for_entry should clear its entry."""
         coordinator = KeymasterCoordinator(hass)
-        coordinator._cancel_debounced_refresh = Mock()
+        coordinator._cancel_debounced_refresh["entry_1"] = Mock()
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
 
-        with patch.object(coordinator, "async_request_refresh", new=AsyncMock()):
-            await coordinator._trigger_debounced_refresh(utcnow())
+        with patch.object(coordinator, "async_refresh_lock", new=AsyncMock(return_value=set())):
+            await coordinator._trigger_debounced_refresh_for_entry("entry_1", utcnow())
 
-        assert coordinator._cancel_debounced_refresh is None
+        assert "entry_1" not in coordinator._cancel_debounced_refresh
 
     async def test_debounce_cancelled_on_full_refresh(self, hass: HomeAssistant):
         """_async_update_data should cancel any pending debounced refresh."""
         coordinator = KeymasterCoordinator(hass)
         cancel_mock = Mock()
-        coordinator._cancel_debounced_refresh = cancel_mock
+        coordinator._cancel_debounced_refresh = {"entry_1": cancel_mock}
         coordinator._initial_setup_done_event = Mock()
         coordinator._initial_setup_done_event.wait = AsyncMock()
-        coordinator._cancel_quick_refresh = None
 
         with (
             patch.object(coordinator, "_async_save_data", new=AsyncMock()),
@@ -313,7 +316,7 @@ async def test_debounce_cancelled_on_full_refresh(self, hass: HomeAssistant):
             await coordinator._async_update_data()
 
         cancel_mock.assert_called_once()
-        assert coordinator._cancel_debounced_refresh is None
+        assert coordinator._cancel_debounced_refresh == {}
 
 
 # ── Entity Handler Tests ────────────────────────────────────────────────────
@@ -594,3 +597,263 @@ async def test_kmlocks_to_dict_excludes_last_code_set_at(self, hass: HomeAssista
         assert result["number"] == 1
         assert result["enabled"] is True
         assert result["pin"] == "1234"
+
+
+# ── Per-Entry Scoped Refresh Tests ──────────────────────────────────────────
+
+
+class TestPerEntryQuickRefresh:
+    """Tests for per-entry quick refresh scheduling (issue #684)."""
+
+    @pytest.fixture
+    def real_coordinator(self, mock_hass):
+        """Coordinator with real _schedule_quick_refresh_if_needed."""
+        with patch.object(KeymasterCoordinator, "__init__", return_value=None):
+            coordinator = KeymasterCoordinator(mock_hass)
+            coordinator.hass = mock_hass
+            coordinator.kmlocks = {}
+            coordinator._quick_refresh_entry_ids = set()
+            coordinator._cancel_quick_refresh = {}
+            coordinator._cancel_debounced_refresh = {}
+            coordinator._externally_dirty_entry_ids = set()
+            coordinator._initial_setup_done_event = AsyncMock()
+            coordinator._initial_setup_done_event.wait = AsyncMock()
+            coordinator.async_set_updated_data = Mock()
+            coordinator.async_schedule_keymaster_notifications = Mock()
+            coordinator._sync_tx_depth = 0
+            coordinator._sync_tx_dirty_ids = set()
+            coordinator._sync_tx_all_entry_ids = False
+            return coordinator
+
+    async def test_set_pin_records_target_entry_id(self, real_coordinator):
+        """set_pin_on_lock adds the target lock's entry ID to pending set."""
+
+        provider = Mock(spec=BaseLockProvider)
+        provider.async_set_usercode = AsyncMock(return_value=True)
+        lock = KeymasterLock(
+            lock_name="Test",
+            lock_entity_id="lock.test",
+            keymaster_config_entry_id="entry_1",
+        )
+        lock.connected = True
+        lock.provider = provider
+        lock.code_slots = {
+            1: KeymasterCodeSlot(number=1, pin="1234", name="Slot", active=True, enabled=True),
+        }
+        real_coordinator.kmlocks["entry_1"] = lock
+
+        await real_coordinator.set_pin_on_lock("entry_1", 1, "5678", override=True)
+
+        assert real_coordinator._quick_refresh_entry_ids == {"entry_1"}
+
+    async def test_clear_pin_records_target_entry_id(self, real_coordinator):
+        """clear_pin_from_lock adds the target lock's entry ID to pending set."""
+
+        provider = Mock(spec=BaseLockProvider)
+        provider.async_clear_usercode = AsyncMock(return_value=True)
+        lock = KeymasterLock(
+            lock_name="Test",
+            lock_entity_id="lock.test",
+            keymaster_config_entry_id="entry_1",
+        )
+        lock.connected = True
+        lock.provider = provider
+        lock.code_slots = {
+            1: KeymasterCodeSlot(number=1, pin="1234", name="Slot", active=True, enabled=True),
+        }
+        real_coordinator.kmlocks["entry_1"] = lock
+
+        await real_coordinator.clear_pin_from_lock("entry_1", 1, override=True)
+
+        assert real_coordinator._quick_refresh_entry_ids == {"entry_1"}
+
+    async def test_two_locks_get_independent_quick_refresh_timers(self, hass: HomeAssistant):
+        """Edits on two locks produce two independent per-entry timers."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+        coordinator.kmlocks["entry_2"] = _make_lock("entry_2")
+        coordinator._quick_refresh_entry_ids = {"entry_1", "entry_2"}
+
+        await coordinator._schedule_quick_refresh_if_needed()
+
+        assert "entry_1" in coordinator._cancel_quick_refresh
+        assert "entry_2" in coordinator._cancel_quick_refresh
+        assert (
+            coordinator._cancel_quick_refresh["entry_1"]
+            is not (coordinator._cancel_quick_refresh["entry_2"])
+        )
+        # Clean up
+        coordinator._cancel_quick_refresh["entry_1"]()
+        coordinator._cancel_quick_refresh["entry_2"]()
+
+    async def test_cancelling_one_entry_does_not_disturb_another(self, hass: HomeAssistant):
+        """Cancelling/rescheduling for one entry must not disturb another."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+        coordinator.kmlocks["entry_2"] = _make_lock("entry_2")
+        coordinator._quick_refresh_entry_ids = {"entry_1", "entry_2"}
+
+        await coordinator._schedule_quick_refresh_if_needed()
+        entry_2_cancel = coordinator._cancel_quick_refresh["entry_2"]
+
+        # Clear only entry_1
+        await coordinator._clear_pending_quick_refresh("entry_1")
+
+        assert "entry_1" not in coordinator._cancel_quick_refresh
+        assert coordinator._cancel_quick_refresh["entry_2"] is entry_2_cancel
+        # Clean up
+        entry_2_cancel()
+
+    async def test_two_locks_get_independent_debounced_timers(self, hass: HomeAssistant):
+        """Debounced edits on two locks produce two independent per-entry timers."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+        coordinator.kmlocks["entry_2"] = _make_lock("entry_2")
+
+        await coordinator.async_request_debounced_refresh("entry_1")
+        await coordinator.async_request_debounced_refresh("entry_2")
+
+        assert "entry_1" in coordinator._cancel_debounced_refresh
+        assert "entry_2" in coordinator._cancel_debounced_refresh
+        assert (
+            coordinator._cancel_debounced_refresh["entry_1"]
+            is not (coordinator._cancel_debounced_refresh["entry_2"])
+        )
+        # Clean up
+        coordinator._cancel_debounced_refresh["entry_1"]()
+        coordinator._cancel_debounced_refresh["entry_2"]()
+
+    async def test_shutdown_cancels_all_per_entry_handles(self, hass: HomeAssistant):
+        """Shutdown cancels every per-entry handle and leaves dicts empty."""
+        coordinator = KeymasterCoordinator(hass)
+        cancel_quick_1 = Mock()
+        cancel_quick_2 = Mock()
+        cancel_debounce_1 = Mock()
+        coordinator._cancel_quick_refresh = {"e1": cancel_quick_1, "e2": cancel_quick_2}
+        coordinator._cancel_debounced_refresh = {"e1": cancel_debounce_1}
+
+        await coordinator.async_shutdown()
+
+        cancel_quick_1.assert_called_once()
+        cancel_quick_2.assert_called_once()
+        cancel_debounce_1.assert_called_once()
+        assert coordinator._cancel_quick_refresh == {}
+        assert coordinator._cancel_debounced_refresh == {}
+
+    async def test_delete_lock_cancels_pending_timers(self, hass: HomeAssistant):
+        """A lock deleted while holding a pending handle does not leave dangling timer."""
+        coordinator = KeymasterCoordinator(hass)
+        cancel_quick = Mock()
+        cancel_debounce = Mock()
+        coordinator._cancel_quick_refresh = {"entry_1": cancel_quick}
+        coordinator._cancel_debounced_refresh = {"entry_1": cancel_debounce}
+        coordinator._quick_refresh_entry_ids = {"entry_1"}
+        coordinator._externally_dirty_entry_ids = {"entry_1"}
+
+        coordinator._cancel_entry_refresh_timers("entry_1")
+
+        cancel_quick.assert_called_once()
+        cancel_debounce.assert_called_once()
+        assert "entry_1" not in coordinator._cancel_quick_refresh
+        assert "entry_1" not in coordinator._cancel_debounced_refresh
+        assert "entry_1" not in coordinator._quick_refresh_entry_ids
+        assert "entry_1" not in coordinator._externally_dirty_entry_ids
+
+    async def test_quick_refresh_routes_through_serialised_refresh_lock(self, hass: HomeAssistant):
+        """Quick refresh triggers async_refresh_lock which serialises via _debounced_refresh."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+
+        with patch.object(
+            coordinator, "async_refresh_lock", new=AsyncMock(return_value=set())
+        ) as mock_refresh:
+            await coordinator._trigger_quick_refresh_for_entry("entry_1", utcnow())
+            mock_refresh.assert_called_once_with("entry_1")
+
+    async def test_rapid_debounce_coalesces_for_one_lock(self, hass: HomeAssistant):
+        """Multiple rapid debounced calls for one lock coalesce into one refresh."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+
+        await coordinator.async_request_debounced_refresh("entry_1")
+        await coordinator.async_request_debounced_refresh("entry_1")
+        await coordinator.async_request_debounced_refresh("entry_1")
+
+        # Only one timer for entry_1
+        assert len(coordinator._cancel_debounced_refresh) == 1
+        assert "entry_1" in coordinator._cancel_debounced_refresh
+
+        # Trigger and verify only one call
+        with patch.object(
+            coordinator, "async_refresh_lock", new=AsyncMock(return_value=set())
+        ) as mock_refresh:
+            coordinator._cancel_debounced_refresh["entry_1"]()
+            await coordinator._trigger_debounced_refresh_for_entry("entry_1", utcnow())
+            mock_refresh.assert_called_once_with("entry_1")
+
+
+class TestPerEntryRefreshEdgeCases:
+    """Cover per-entry refresh fallback, cancel-all, and dedup guard paths."""
+
+    async def test_debounced_refresh_global_sentinel_calls_full_refresh(self, hass: HomeAssistant):
+        """_trigger_debounced_refresh_for_entry with '_global' calls async_request_refresh."""
+        coordinator = KeymasterCoordinator(hass)
+
+        with patch.object(coordinator, "async_request_refresh", new=AsyncMock()) as mock_refresh:
+            await coordinator._trigger_debounced_refresh_for_entry("_global", utcnow())
+            mock_refresh.assert_called_once()
+
+    async def test_debounced_refresh_removed_lock_calls_full_refresh(self, hass: HomeAssistant):
+        """If a lock is removed before its debounce timer fires, fall back to full refresh."""
+        coordinator = KeymasterCoordinator(hass)
+        # entry_1 was scheduled but then removed from kmlocks
+        coordinator.kmlocks.pop("entry_1", None)
+
+        with patch.object(coordinator, "async_request_refresh", new=AsyncMock()) as mock_refresh:
+            await coordinator._trigger_debounced_refresh_for_entry("entry_1", utcnow())
+            mock_refresh.assert_called_once()
+
+    async def test_clear_all_quick_refresh_cancels_every_handle(self, hass: HomeAssistant):
+        """_clear_pending_quick_refresh(None) cancels ALL per-entry handles."""
+        coordinator = KeymasterCoordinator(hass)
+        cancel_1 = Mock()
+        cancel_2 = Mock()
+        cancel_3 = Mock()
+        coordinator._cancel_quick_refresh = {
+            "entry_1": cancel_1,
+            "entry_2": cancel_2,
+            "entry_3": cancel_3,
+        }
+
+        await coordinator._clear_pending_quick_refresh()
+
+        cancel_1.assert_called_once()
+        cancel_2.assert_called_once()
+        cancel_3.assert_called_once()
+        assert coordinator._cancel_quick_refresh == {}
+
+    async def test_schedule_quick_refresh_does_not_reschedule_existing_timer(
+        self, hass: HomeAssistant
+    ):
+        """An entry with an existing pending handle is NOT rescheduled."""
+        coordinator = KeymasterCoordinator(hass)
+        coordinator.kmlocks["entry_1"] = _make_lock("entry_1")
+        coordinator.kmlocks["entry_2"] = _make_lock("entry_2")
+
+        # entry_1 already has a pending timer
+        existing_handle = Mock()
+        coordinator._cancel_quick_refresh = {"entry_1": existing_handle}
+        # Both entries are pending quick refresh
+        coordinator._quick_refresh_entry_ids = {"entry_1", "entry_2"}
+
+        await coordinator._schedule_quick_refresh_if_needed()
+
+        # entry_1's handle is unchanged (identity check — not rescheduled)
+        assert coordinator._cancel_quick_refresh["entry_1"] is existing_handle
+        # entry_2 got a NEW timer
+        assert "entry_2" in coordinator._cancel_quick_refresh
+        assert coordinator._cancel_quick_refresh["entry_2"] is not existing_handle
+        # pending set was cleared
+        assert coordinator._quick_refresh_entry_ids == set()
+        # Clean up
+        coordinator._cancel_quick_refresh["entry_2"]()