fix: confirm notification delivery before deduplication

src/openadapt_tray/app.py

@@ -185,7 +185,6 @@ def _show_state_notification(self, state: AppState) -> None:
         """
         if state.state == self._last_notified_tray_state:
             return
-        self._last_notified_tray_state = state.state
 
         messages = {
             TrayState.RECORDING: (
@@ -200,13 +199,18 @@ def _show_state_notification(self, state: AppState) -> None:
             TrayState.ERROR: ("Error", state.error_message or "An error occurred"),
         }
 
-        if state.state in messages:
-            title, body = messages[state.state]
-            self.notifications.show(
-                title,
-                body,
-                duration_ms=self.config.notification_duration_ms,
-            )
+        if state.state not in messages:
+            self._last_notified_tray_state = state.state
+            return
+
+        title, body = messages[state.state]
+        delivered = self.notifications.show(
+            title,
+            body,
+            duration_ms=self.config.notification_duration_ms,
+        )
+        if delivered:
+            self._last_notified_tray_state = state.state
 
     def _toggle_recording(self) -> None:
         """Toggle recording state."""

src/openadapt_tray/notifications.py

@@ -11,8 +11,11 @@
 from collections.abc import Callable
 from pathlib import Path
 
+_DELIVERY_CONFIRMATION_TIMEOUT_SECONDS = 5
+
 try:
     from desktop_notifier import Button, DesktopNotifier, ReplyField, Urgency
+
     DESKTOP_NOTIFIER_AVAILABLE = True
 except ImportError:
     DESKTOP_NOTIFIER_AVAILABLE = False
@@ -71,21 +74,23 @@ def _detect_backend(self) -> str:
         # On macOS, check if running from app bundle
         if sys.platform == "darwin":
             # Check both APP_BUNDLE env var and actual bundle structure
-            is_app_bundle = (
-                os.environ.get('APP_BUNDLE') or
-                'Contents/MacOS' in str(Path(__file__).resolve())
+            is_app_bundle = os.environ.get("APP_BUNDLE") or "Contents/MacOS" in str(
+                Path(__file__).resolve()
             )
             if not is_app_bundle:
                 # Not in app bundle - try desktop-notifier anyway, fall back if it fails
                 try:
                     from desktop_notifier.macos import CocoaNotificationCenter
+
                     # Try to initialize to see if it works
                     CocoaNotificationCenter()
                     print("desktop-notifier initialized successfully")
                     return "desktop-notifier"
                 except Exception as e:
                     # If desktop-notifier can't initialize, use AppleScript
-                    print(f"desktop-notifier not available ({e}), using AppleScript for notifications")
+                    print(
+                        f"desktop-notifier not available ({e}), using AppleScript for notifications"
+                    )
                     return "macos"
 
         # Use desktop-notifier on all platforms
@@ -188,11 +193,24 @@ def _show_desktop_notifier(
         if reply_field:
             reply_field_object = ReplyField(title=reply_field, button_title="Send")
 
-        # Show notification asynchronously
+        # Show the notification and confirm that the backend accepted it. A queued
+        # coroutine is not delivery: callers use this result to decide whether a
+        # failed notification must be retried.
         try:
             if self._loop.is_running():
-                # If loop is already running, schedule the coroutine
-                asyncio.run_coroutine_threadsafe(
+                try:
+                    running_loop = asyncio.get_running_loop()
+                except RuntimeError:
+                    running_loop = None
+
+                if running_loop is self._loop:
+                    print(
+                        "desktop-notifier sync API cannot confirm delivery from "
+                        "its running event-loop thread"
+                    )
+                    return False
+
+                future = asyncio.run_coroutine_threadsafe(
                     self._notifier.send(
                         title=title,
                         message=body,
@@ -202,10 +220,14 @@ def _show_desktop_notifier(
                         reply_field=reply_field_object,
                         on_clicked=on_clicked,
                     ),
-                    self._loop
+                    self._loop,
                 )
+                try:
+                    future.result(timeout=_DELIVERY_CONFIRMATION_TIMEOUT_SECONDS)
+                except Exception:
+                    future.cancel()
+                    raise
             else:
-                # Run in the event loop
                 self._loop.run_until_complete(
                     self._notifier.send(
                         title=title,

tests/test_app.py

@@ -12,9 +12,10 @@ def _make_test_app():
     """Build an app without platform or tray side effects."""
     from openadapt_tray.app import TrayApplication
 
-    with patch("openadapt_tray.app.get_platform_handler") as platform, patch(
-        "openadapt_tray.app.pystray"
-    ) as pystray:
+    with (
+        patch("openadapt_tray.app.get_platform_handler") as platform,
+        patch("openadapt_tray.app.pystray") as pystray,
+    ):
         platform.return_value = MagicMock()
         pystray.Icon.return_value = MagicMock()
         return TrayApplication(config=TrayConfig())
@@ -225,6 +226,34 @@ def test_sync_change_does_not_claim_recording_stopped(self):
 
         mock_show.assert_not_called()
 
+    def test_failed_notification_is_retried(self):
+        app = _make_test_app()
+        with patch.object(app.notifications, "show", return_value=False) as show:
+            app.state.transition(TrayState.RECORDING, current_capture="test")
+            app._show_state_notification(app.state.current)
+
+        assert show.call_count == 2
+        assert app._last_notified_tray_state == TrayState.IDLE
+
+    def test_delivered_notification_is_not_repeated(self):
+        app = _make_test_app()
+        with patch.object(app.notifications, "show", return_value=True) as show:
+            app.state.transition(TrayState.RECORDING, current_capture="test")
+            app._show_state_notification(app.state.current)
+
+        show.assert_called_once()
+        assert app._last_notified_tray_state == TrayState.RECORDING
+
+    def test_unmessaged_transition_resets_notification_deduplication(self):
+        app = _make_test_app()
+        with patch.object(app.notifications, "show", return_value=False) as show:
+            app.state.transition(TrayState.RECORDING, current_capture="test")
+            app.state.transition(TrayState.RECORDING_STOPPING)
+            app.state.transition(TrayState.IDLE)
+
+        assert show.call_count == 2
+        assert show.call_args.args[0] == "Recording Stopped"
+
 
 class TestQuit:
     """Tests for application quit functionality."""
@@ -328,7 +357,7 @@ def test_a_clean_load_says_nothing(self, mock_platform, mock_pystray):
 
 
 class TestStartRecordingWhenNoDialogCanBeShown:
-    """"The prompt never appeared" used to be indistinguishable from "cancelled"."""
+    """ "The prompt never appeared" used to be indistinguishable from "cancelled"."""
 
     def _app(self, mock_platform, mock_pystray):
         from openadapt_tray.app import TrayApplication
@@ -389,9 +418,7 @@ def test_click_that_opened_nothing_tells_the_user(
 
     @patch("openadapt_tray.app.pystray")
     @patch("openadapt_tray.app.get_platform_handler")
-    def test_click_that_opened_something_stays_quiet(
-        self, mock_platform, mock_pystray
-    ):
+    def test_click_that_opened_something_stays_quiet(self, mock_platform, mock_pystray):
         from openadapt_tray.app import TrayApplication
 
         mock_platform.return_value = MagicMock()

tests/test_notifications.py

@@ -63,3 +63,63 @@ def test_returncode_drives_result(self):
             assert manager._show_linux("t", "b", None) is True
             run.return_value = subprocess.CompletedProcess([], 1, b"", b"")
             assert manager._show_linux("t", "b", None) is False
+
+
+class TestDesktopNotifierDelivery:
+    def _manager(self):
+        manager = _manager()
+        manager._notifier = MagicMock()
+        manager._notifier.send.return_value = MagicMock()
+        manager._loop = MagicMock()
+        manager._loop.is_running.return_value = True
+        return manager
+
+    @patch("openadapt_tray.notifications.asyncio.run_coroutine_threadsafe")
+    @patch("openadapt_tray.notifications.asyncio.get_running_loop")
+    @patch("openadapt_tray.notifications.Urgency", create=True)
+    def test_running_loop_backend_failure_is_not_delivery(
+        self, _urgency, get_running_loop, run_coroutine_threadsafe
+    ):
+        manager = self._manager()
+        get_running_loop.side_effect = RuntimeError("no loop in caller thread")
+        future = run_coroutine_threadsafe.return_value
+        future.result.side_effect = RuntimeError("notification daemon rejected request")
+
+        assert (
+            manager._show_desktop_notifier("t", "b", None, None, "normal", None, None)
+            is False
+        )
+
+        future.cancel.assert_called_once()
+
+    @patch("openadapt_tray.notifications.asyncio.run_coroutine_threadsafe")
+    @patch("openadapt_tray.notifications.asyncio.get_running_loop")
+    @patch("openadapt_tray.notifications.Urgency", create=True)
+    def test_running_loop_waits_for_backend_confirmation(
+        self, _urgency, get_running_loop, run_coroutine_threadsafe
+    ):
+        manager = self._manager()
+        get_running_loop.side_effect = RuntimeError("no loop in caller thread")
+
+        assert (
+            manager._show_desktop_notifier("t", "b", None, None, "normal", None, None)
+            is True
+        )
+
+        run_coroutine_threadsafe.return_value.result.assert_called_once_with(timeout=5)
+
+    @patch("openadapt_tray.notifications.asyncio.run_coroutine_threadsafe")
+    @patch("openadapt_tray.notifications.asyncio.get_running_loop")
+    @patch("openadapt_tray.notifications.Urgency", create=True)
+    def test_sync_api_refuses_unconfirmed_delivery_on_its_event_loop(
+        self, _urgency, get_running_loop, run_coroutine_threadsafe
+    ):
+        manager = self._manager()
+        get_running_loop.return_value = manager._loop
+
+        assert (
+            manager._show_desktop_notifier("t", "b", None, None, "normal", None, None)
+            is False
+        )
+
+        run_coroutine_threadsafe.assert_not_called()