feat(android): on-device OS notification capture (NotificationListenerService)

Add an Android backend to os_notification_listener.py so notifications on the
experimental Android-TV build drive the existing NotificationColorStripSource
LED effects (flash/pulse/sweep, per-app colors + sounds) at app-name parity
with the Windows/Linux backends.

A Kotlin NotificationListenerService forwards the posting app's display label
across the Chaquopy JNI boundary into a new push-based _AndroidBackend +
module-level push_notification() receiver; the existing color-strip pipeline,
per-app colors/filters, and history endpoint are reused unchanged.

- Python: _AndroidBackend (probed first), push_notification() receiver,
  _LinuxBackend.probe() hardened with is_linux() to exclude Android (which
  also reports platform.system() == "Linux").
- Android: LedGrabNotificationListener NLS — serial single-thread executor,
  full crash isolation around Python.getInstance(), label-only forwarding
  (never notification title/body), ongoing/group-summary/self-package noise
  filtering. Manifest service exported + gated by
  BIND_NOTIFICATION_LISTENER_SERVICE (no new uses-permission).
- UX: prompt-once notification-access + manual "Grant notification access"
  button wired into the D-pad focus chain (computed from visible controls);
  en/ru/zh strings.
- Tests: 11 isolated unit tests — module-global + tmp_path history isolation,
  push routing contract, callback-exception swallowing, None app-name, and a
  desktop-regression lock on backend selection order.
- Docs: README OS-support Android column (notification + audio cells),
  ANDROID-REVIEW status flipped to Implemented.

Zero new Python deps; no build.gradle.kts / Chaquopy pip changes.

ANDROID-REVIEW/android-missing-functionality.md

@@ -45,7 +45,7 @@ Python receiver engine mirroring that pattern.**
 | LED transports (network/USB-serial/BLE) | ✅ | ✅ (USB via Android driver, BLE via Android bridge) | No |
 | System metrics | psutil | ✅ CPU/RAM/battery/thermal via `/proc`, `/sys` (`AndroidMetricsProvider`) | No |
 | **Audio capture** | WASAPI / Sounddevice | ❌ no PortAudio | **Yes** |
-| **Notification capture** | WinRT / D-Bus | ❌ listener only Win/Linux | **Yes** |
+| Notification capture | WinRT / D-Bus | ✅ NotificationListenerService → `push_notification()` | No (implemented) |
 | Webcam capture | OpenCV | ❌ no OpenCV wheel | Yes (niche) |
 | GPU monitoring | NVML | ❌ no NVIDIA GPU | Marginal |
 | Capture from *another* Android phone | scrcpy/ADB | ❌ | Skip (redundant) |
@@ -70,7 +70,7 @@ Python receiver engine mirroring that pattern.**
   media and the device's own audio. Root mode (no MediaProjection) → mic-only.
 - 📄 **See `android-audio-capture-plan.md`** for the full implementation plan.
 
-### 🔔 Notification capture — **FEASIBLE, HIGH VALUE** ⭐ (planned)
+### 🔔 Notification capture — **IMPLEMENTED** ✅ (shipped)
 
 - **Android is the *best* platform for this:** `NotificationListenerService` is the native,
   event-push mechanism (no polling).
@@ -82,8 +82,13 @@ Python receiver engine mirroring that pattern.**
 - **Permission:** user enables "Notification access" in Settings (`ACTION_NOTIFICATION_LISTENER_SETTINGS`);
   no runtime-permission popup.
 - **Effort:** moderate. **Value:** high.
-- 📄 **Plan approved & detailed** — see `C:\Users\Alexei\.claude\plans\deep-enchanting-muffin.md`
+- ✅ **Implemented** on branch `feature/android-notification-capture`: a push-based
-  (app-name parity; prompt-once permission UX).
+  `_AndroidBackend` + module-level `push_notification()` in `os_notification_listener.py`,
+  a Kotlin `LedGrabNotificationListener` (NLS), and prompt-once permission UX. App-name
+  parity — only the resolved app label crosses the JNI boundary, never the notification
+  title/body. ⚠️ App labels can differ across OSes (Windows `display_name` / Linux D-Bus
+  `app_name` / Android `getApplicationLabel`), so desktop-configured per-app colors/filters
+  may need re-matching on Android.
 
 ### 📷 Webcam capture — **FEASIBLE, LOW VALUE**
 
@@ -128,7 +133,7 @@ Python receiver engine mirroring that pattern.**
 
 | Priority | Feature | Effort | Value | New Python deps | Status |
 | -------- | ------- | ------ | ----- | --------------- | ------ |
-| 1 | Notification capture | Moderate | High | None | **Plan approved** |
+| 1 | Notification capture | Moderate | High | None | **✅ Implemented** |
 | 2 | Audio capture | Moderate | High | None | **Plan written** (this folder) |
 | 3 | Automation: foreground-app condition | Moderate | Moderate | None | Idea |
 | 4 | Webcam capture (CameraX) | Moderate | Low | None | Idea |

README.md

@@ -105,16 +105,17 @@ LedGrab runs as a desktop / server application:
 
 ### Feature support by OS
 
-| Feature | Windows | Linux / macOS |
+| Feature | Windows | Linux / macOS | Android TV (experimental) |
-| ------- | ------- | ------------- |
+| ------- | ------- | ------------- | ------------------------- |
-| Screen capture | DXCam, BetterCam, WGC, MSS | MSS |
+| Screen capture | DXCam, BetterCam, WGC, MSS | MSS | MediaProjection; root `screenrecord` (rooted devices) |
-| Webcam capture | OpenCV (DirectShow) | OpenCV (V4L2) |
+| Webcam capture | OpenCV (DirectShow) | OpenCV (V4L2) | — (no OpenCV wheel) |
-| Audio capture | WASAPI, Sounddevice | Sounddevice (PulseAudio/PipeWire) |
+| Audio capture | WASAPI, Sounddevice | Sounddevice (PulseAudio/PipeWire) | AudioPlaybackCapture (API 29+) |
-| GPU monitoring | NVIDIA (nvidia-ml-py) | NVIDIA (nvidia-ml-py) |
+| GPU monitoring | NVIDIA (nvidia-ml-py) | NVIDIA (nvidia-ml-py) | — (CPU/RAM/battery/thermal via `/proc`) |
-| Capture from Android phone | scrcpy (ADB) | scrcpy (ADB) |
+| Capture from Android phone | scrcpy (ADB) | scrcpy (ADB) | — (captures its own screen instead) |
-| Notification capture | WinRT | dbus (Linux) |
+| Notification capture | WinRT | dbus (Linux) | NotificationListenerService |
-| Monitor names | Friendly names (WMI) | Generic ("Display 0") |
+| Monitor names | Friendly names (WMI) | Generic ("Display 0") | Single built-in display |
-| Automation: window/process conditions | Supported | Partial |
+| LED transports | Network, USB-serial, BLE | Network, USB-serial, BLE | Network, USB-serial (Android driver), BLE (Android bridge) |
+| Automation: window/process conditions | Supported | Partial | — |
 
 ## Requirements
 

android/app/src/main/AndroidManifest.xml

@@ -110,6 +110,23 @@
                 android:value="Root-mode screen capture for ambient LED sync. Uses /system/bin/screenrecord on rooted devices to avoid MediaProjection's persistent capture indicator overlay, which is required for the always-on ambient-lighting use case." />
         </service>
 
+        <!-- Notification capture — a NotificationListenerService bound by
+             system_server. exported="true" is REQUIRED here (the system binds
+             it cross-process) and intentionally diverges from CaptureService
+             (exported="false"); access is gated by the system-held
+             BIND_NOTIFICATION_LISTENER_SERVICE permission, so no new
+             <uses-permission> is needed. The user grants access via
+             Settings > Notification access (opened from MainActivity). -->
+        <service
+            android:name=".LedGrabNotificationListener"
+            android:label="@string/notification_listener_label"
+            android:exported="true"
+            android:permission="android.permission.BIND_NOTIFICATION_LISTENER_SERVICE">
+            <intent-filter>
+                <action android:name="android.service.notification.NotificationListenerService" />
+            </intent-filter>
+        </service>
+
         <!-- Autostart — fires on device boot (and package replace).
              On rooted devices, launches CaptureService directly so capture
              resumes without the user tapping Start. Unrooted devices are

android/app/src/main/java/com/ledgrab/android/LedGrabNotificationListener.kt

@@ -0,0 +1,97 @@
+package com.ledgrab.android
+
+import android.app.Notification
+import android.service.notification.NotificationListenerService
+import android.service.notification.StatusBarNotification
+import android.util.Log
+import com.chaquo.python.Python
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.Executors
+
+/**
+ * Captures posted OS notifications and forwards the posting app's display
+ * label to the Python notification pipeline, where the existing
+ * `NotificationColorStripSource` fires its one-shot LED effect.
+ *
+ * Direction is Kotlin -> Python via the process-global Chaquopy instance
+ * (NOT a per-[CaptureService] [PythonBridge]): `system_server` binds this
+ * service independently of [CaptureService], so it resolves Python itself.
+ * The Python receiver (`os_notification_listener.push_notification`) is a
+ * no-op whenever the server/listener isn't running, so a notification
+ * arriving before — or after — a capture session is safely ignored.
+ */
+class LedGrabNotificationListener : NotificationListenerService() {
+
+    // Serial executor: the Python receiver does a (non-concurrency-safe) history
+    // disk write and may play a sound, so pushes must not overlap. Off the main
+    // looper to keep the system service responsive.
+    private val pushExecutor = Executors.newSingleThreadExecutor()
+
+    // packageName -> resolved human-readable label. Matches the app_name the
+    // Windows/Linux backends pass, so per-app colors/filters keep working.
+    // Naturally bounded by the number of notification-posting apps (tens) and
+    // cleared with the process — no eviction needed.
+    private val labelCache = ConcurrentHashMap<String, String>()
+
+    override fun onNotificationPosted(sbn: StatusBarNotification?) {
+        val notification = sbn ?: return
+
+        // The Python server (and thus the listener) only exists during a capture
+        // session. isRunning is a coarse early-out — the authoritative gate is the
+        // Python receiver's None-check — but it avoids needless JNI churn here.
+        if (!CaptureService.isRunning) return
+
+        // Filter notifications that should never drive an effect:
+        //  - ongoing (media transport, downloads): not user-facing "alerts"
+        //  - group summaries: duplicate their child notifications
+        //  - our own foreground-service notification: would self-trigger
+        if (notification.isOngoing) return
+        if ((notification.notification.flags and Notification.FLAG_GROUP_SUMMARY) != 0) return
+        if (notification.packageName == packageName) return
+
+        val label = resolveAppLabel(notification.packageName)
+
+        pushExecutor.execute {
+            try {
+                Python.getInstance()
+                    .getModule(PY_MODULE)
+                    .callAttr("push_notification", label)
+            } catch (t: Throwable) {
+                // Never crash a system-bound service. Python.getInstance() throws
+                // IllegalStateException if Python.start() hasn't run (e.g. the
+                // service was bound at boot before the app process initialized).
+                // Log at debug — the label is potentially sensitive on a shared TV.
+                Log.d(TAG, "push_notification failed: ${t.message}")
+            }
+        }
+    }
+
+    /** Resolve (and cache) a package's human-readable label; fall back to the package name. */
+    private fun resolveAppLabel(pkg: String): String {
+        labelCache[pkg]?.let { return it }
+        val resolved = runCatching {
+            val info = packageManager.getApplicationInfo(pkg, 0)
+            packageManager.getApplicationLabel(info).toString()
+        }.getOrDefault(pkg)
+        labelCache[pkg] = resolved
+        return resolved
+    }
+
+    override fun onListenerConnected() {
+        Log.i(TAG, "Notification listener connected")
+    }
+
+    override fun onListenerDisconnected() {
+        Log.i(TAG, "Notification listener disconnected")
+    }
+
+    override fun onDestroy() {
+        pushExecutor.shutdown()
+        super.onDestroy()
+    }
+
+    companion object {
+        private const val TAG = "LedGrabNotifListener"
+        private const val PY_MODULE = "ledgrab.core.processing.os_notification_listener"
+    }
+}

android/app/src/main/java/com/ledgrab/android/MainActivity.kt

@@ -24,6 +24,7 @@ import android.widget.ImageView
 import android.widget.LinearLayout
 import android.widget.ScrollView
 import android.widget.TextView
+import androidx.core.app.NotificationManagerCompat
 import androidx.core.content.ContextCompat
 import androidx.core.splashscreen.SplashScreen.Companion.installSplashScreen
 import com.google.zxing.BarcodeFormat
@@ -55,6 +56,8 @@ class MainActivity : Activity() {
         private const val REQUEST_POST_NOTIFICATIONS = 1002
         private const val REQUEST_RECORD_AUDIO = 1003
         private const val QR_SIZE_PX = 560
+        private const val NOTIF_PREFS = "ledgrab_notif"
+        private const val KEY_NOTIF_ACCESS_PROMPTED = "notif_access_prompted"
     }
 
     // Stopped-state views (always inflated).
@@ -64,6 +67,7 @@ class MainActivity : Activity() {
     private lateinit var versionText: TextView
     private lateinit var autostartCheck: CheckBox
     private lateinit var autostartPrefs: AutostartPrefs
+    private lateinit var grantNotificationButton: Button
 
     // Running-state views (lazy-inflated via ViewStub).
     private lateinit var runningPanelStub: ViewStub
@@ -107,6 +111,7 @@ class MainActivity : Activity() {
         toggleButton = findViewById(R.id.toggle_button)
         versionText = findViewById(R.id.version_text)
         autostartCheck = findViewById(R.id.autostart_check)
+        grantNotificationButton = findViewById(R.id.grant_notification_button)
 
         val versionName = packageManager.getPackageInfo(packageName, 0).versionName
         versionText.text = getString(R.string.version_prefix, versionName ?: "?")
@@ -127,8 +132,10 @@ class MainActivity : Activity() {
             autostartCheck.visibility = View.GONE
         }
 
+        grantNotificationButton.setOnClickListener { openNotificationListenerSettings() }
         toggleButton.setOnClickListener { startCapture() }
 
+        updateNotificationAccessUi()
         updateUI()
     }
 
@@ -149,12 +156,16 @@ class MainActivity : Activity() {
 
     override fun onResume() {
         super.onResume()
+        if (!::stoppedPanel.isInitialized) return
         // Restart the pulse if we returned to the foreground while the
-        // service is still running. The running panel's view may have
+        // service is still running. The running panel's view may have been
-        // been recreated; ensureRunningPanelInflated already keys off
+        // recreated; ensureRunningPanelInflated already keys off the field
-        // the field reference.
+        // reference. When stopped, refresh the notification-access button —
-        if (CaptureService.isRunning && ::stoppedPanel.isInitialized) {
+        // the user may have just granted/revoked access in Settings.
+        if (CaptureService.isRunning) {
             updateUI()
+        } else {
+            updateNotificationAccessUi()
         }
     }
 
@@ -197,6 +208,7 @@ class MainActivity : Activity() {
 
     private fun startRootCaptureService() {
         ensureNotificationPermission()
+        ensureNotificationListenerAccess()
         ContextCompat.startForegroundService(this, CaptureService.createRootIntent(this))
         updateUI()
     }
@@ -216,6 +228,7 @@ class MainActivity : Activity() {
 
     private fun startCaptureService(resultCode: Int, resultData: Intent) {
         ensureNotificationPermission()
+        ensureNotificationListenerAccess()
         ensureAudioPermission()
         val intent = CaptureService.createIntent(this, resultCode, resultData)
         ContextCompat.startForegroundService(this, intent)
@@ -493,4 +506,60 @@ class MainActivity : Activity() {
             )
         }
     }
+
+    /** Whether the user has granted notification-listener access to this app. */
+    private fun isNotificationAccessGranted(): Boolean =
+        NotificationManagerCompat.getEnabledListenerPackages(this).contains(packageName)
+
+    /** Open the system Notification-access screen (manual affordance / re-grant). */
+    private fun openNotificationListenerSettings() {
+        runCatching {
+            startActivity(Intent(Settings.ACTION_NOTIFICATION_LISTENER_SETTINGS))
+        }.onFailure { Log.w(TAG, "Notification-access settings unavailable: ${it.message}") }
+    }
+
+    /**
+     * Prompt-once-then-remember: the first time capture starts without
+     * notification-listener access, open the settings screen so the user can
+     * grant it — then never nag again (the manual "Grant notification access"
+     * button stays available). Fire-and-forget like [ensureNotificationPermission].
+     */
+    private fun ensureNotificationListenerAccess() {
+        if (isNotificationAccessGranted()) return
+        val prefs = getSharedPreferences(NOTIF_PREFS, MODE_PRIVATE)
+        if (prefs.getBoolean(KEY_NOTIF_ACCESS_PROMPTED, false)) return
+        prefs.edit().putBoolean(KEY_NOTIF_ACCESS_PROMPTED, true).apply()
+        openNotificationListenerSettings()
+    }
+
+    /**
+     * Show the "Grant notification access" button only while access is missing,
+     * then re-wire the D-pad focus chain. Called on create and on resume
+     * (access can change in Settings while we're backgrounded).
+     */
+    private fun updateNotificationAccessUi() {
+        if (!::grantNotificationButton.isInitialized) return
+        grantNotificationButton.visibility =
+            if (isNotificationAccessGranted()) View.GONE else View.VISIBLE
+        wireStoppedFocusChain()
+    }
+
+    /**
+     * Link the visible stopped-panel controls into a single up/down D-pad chain.
+     * Both optional controls (the grant-access button and the root-only autostart
+     * checkbox) may be GONE, so the chain is computed from whatever is visible —
+     * a static nextFocus pointing at a GONE view would strand the focus on a TV
+     * remote.
+     */
+    private fun wireStoppedFocusChain() {
+        val chain = listOfNotNull(
+            toggleButton,
+            grantNotificationButton.takeIf { it.visibility == View.VISIBLE },
+            autostartCheck.takeIf { it.visibility == View.VISIBLE },
+        )
+        chain.forEachIndexed { i, view ->
+            view.nextFocusUpId = (chain.getOrNull(i - 1) ?: view).id
+            view.nextFocusDownId = (chain.getOrNull(i + 1) ?: view).id
+        }
+    }
 }

android/app/src/main/res/layout/activity_main.xml

@@ -66,6 +66,21 @@
             android:focusableInTouchMode="true"
             android:nextFocusDown="@+id/autostart_check" />
 
+        <!-- Shown only while notification-listener access is missing. The D-pad
+             focus chain is wired at runtime (wireStoppedFocusChain) because this
+             button and the autostart checkbox are both conditionally visible. -->
+        <Button
+            android:id="@+id/grant_notification_button"
+            style="@style/Widget.LedGrab.Button.Secondary"
+            android:layout_width="320dp"
+            android:layout_height="56dp"
+            android:layout_marginTop="20dp"
+            android:text="@string/btn_grant_notification_access"
+            android:textSize="18sp"
+            android:focusable="true"
+            android:focusableInTouchMode="true"
+            android:visibility="gone" />
+
         <CheckBox
             android:id="@+id/autostart_check"
             android:layout_width="wrap_content"

android/app/src/main/res/values-ru/strings.xml

@@ -25,4 +25,6 @@
     <string name="notification_channel_description">Отображается, пока LedGrab захватывает экран.</string>
     <string name="notification_title">LedGrab работает</string>
     <string name="notification_text">Веб-интерфейс: %1$s</string>
+    <string name="notification_listener_label">Захват уведомлений LedGrab</string>
+    <string name="btn_grant_notification_access">Разрешить доступ к уведомлениям</string>
 </resources>

android/app/src/main/res/values-zh/strings.xml

@@ -25,4 +25,6 @@
     <string name="notification_channel_description">LedGrab 捕获屏幕时显示。</string>
     <string name="notification_title">LedGrab 运行中</string>
     <string name="notification_text">Web界面:%1$s</string>
+    <string name="notification_listener_label">LedGrab 通知捕获</string>
+    <string name="btn_grant_notification_access">授予通知访问权限</string>
 </resources>

android/app/src/main/res/values/strings.xml

@@ -25,4 +25,6 @@
     <string name="notification_channel_description">Shows while LedGrab is capturing the screen.</string>
     <string name="notification_title">LedGrab Running</string>
     <string name="notification_text">Web UI: %1$s</string>
+    <string name="notification_listener_label">LedGrab notification capture</string>
+    <string name="btn_grant_notification_access">Grant notification access</string>
 </resources>

server/src/ledgrab/core/processing/os_notification_listener.py

@@ -8,6 +8,8 @@ Supported platforms:
 - **Windows**: polls toast notifications via winrt UserNotificationListener
   (falls back to winsdk if winrt packages are not installed)
 - **Linux**: monitors org.freedesktop.Notifications via D-Bus (dbus-next)
+- **Android**: receives notifications pushed from a Kotlin NotificationListenerService
+  via Chaquopy (push-based; see push_notification() and _AndroidBackend)
 """
 
 import asyncio
@@ -17,9 +19,10 @@ import platform
 import threading
 import time
 from pathlib import Path
-from typing import Dict, List, Optional, Set
+from typing import Callable, Dict, List, Optional, Set
 
 from ledgrab.utils import get_logger
+from ledgrab.utils.platform import is_linux
 
 logger = get_logger(__name__)
 
@@ -30,15 +33,71 @@ _HISTORY_MAX = 50
 # Module-level singleton for dependency access
 _instance: Optional["OsNotificationListener"] = None
 
+# Push target for the Android backend — set by _AndroidBackend.start(), read by
+# push_notification(). None when the Android backend isn't running (desktop / server down).
+_android_target: Callable[[str | None], None] | None = None
+
 
 def get_os_notification_listener() -> Optional["OsNotificationListener"]:
     """Return the global OsNotificationListener instance (or None)."""
     return _instance
 
 
+def push_notification(app_name: str | None) -> None:
+    """Receive an Android notification pushed from Kotlin via Chaquopy.
+
+    Called by the LedGrabNotificationListener service through
+    ``Python.getInstance().getModule(...).callAttr("push_notification", label)``.
+    Routes the posting app's display label into the active listener's
+    ``_on_new_notification`` handler. No-op when the Android backend isn't running,
+    so a notification arriving before the server is ready (or on desktop) is safely
+    ignored.
+    """
+    # Snapshot into a local first: stop() may null _android_target concurrently, but an
+    # in-flight push then still completes against the prior callback. Do NOT collapse this
+    # into `if _android_target is not None: _android_target(...)` — that reintroduces a
+    # TOCTOU None-deref race.
+    cb = _android_target
+    if cb is None:
+        return
+    try:
+        cb(app_name)
+    except Exception as exc:  # never let a JNI-side call crash the bound service
+        logger.warning("push_notification callback error: %s", exc)
+
+
 # ── Platform backends ──────────────────────────────────────────────────
 
 
+class _AndroidBackend:
+    """Push-based backend — notifications arrive from Kotlin via push_notification().
+
+    Unlike the Windows/Linux backends (which poll or eavesdrop on a thread), Android
+    notifications are delivered by a Kotlin NotificationListenerService across the
+    Chaquopy JNI boundary into the module-level push_notification() receiver, so
+    start()/stop() simply register/clear the receiver target.
+    """
+
+    def __init__(self, on_notification):
+        self._on_notification = on_notification
+
+    @staticmethod
+    def probe() -> bool:
+        """Return True when running on Android (Chaquopy)."""
+        from ledgrab.utils.platform import is_android
+
+        return is_android()
+
+    def start(self) -> None:
+        global _android_target
+        _android_target = self._on_notification
+        logger.info("OS notification listener: Android backend active")
+
+    def stop(self) -> None:
+        global _android_target
+        _android_target = None
+
+
 def _import_winrt_notifications():
     """Try to import WinRT notification APIs: winrt first, then winsdk fallback.
 
@@ -193,7 +252,9 @@ class _LinuxBackend:
     @staticmethod
     def probe() -> bool:
         """Return True if this backend can run on the current system."""
-        if platform.system() != "Linux":
+        # is_linux() excludes Android, which also reports platform.system() == "Linux"
+        # but has no D-Bus session — defense-in-depth beyond probe ordering.
+        if not is_linux():
             return False
         try:
             import dbus_next  # noqa: F401
@@ -312,8 +373,9 @@ class OsNotificationListener:
         global _instance
         _instance = self
 
-        # Try platform backends in order
+        # Try platform backends in order (Android first — it reports platform.system()
-        for backend_cls in (_WindowsBackend, _LinuxBackend):
+        # == "Linux", so probing it ahead of _LinuxBackend is the robust ordering).
+        for backend_cls in (_AndroidBackend, _WindowsBackend, _LinuxBackend):
             if backend_cls.probe():
                 self._backend = backend_cls(on_notification=self._on_new_notification)
                 self._backend.start()

server/tests/core/processing/test_os_notification_listener_android.py

@@ -0,0 +1,237 @@
+"""Tests for the Android push-based notification backend.
+
+These run on desktop CI (no Android device needed): ``is_android`` is
+monkeypatched and the app label is pushed directly into the module-level
+``push_notification`` receiver, exactly as the Kotlin
+``NotificationListenerService`` would across the Chaquopy bridge.
+
+Isolation (critical): the listener keeps process-global state
+(``_android_target``, ``_instance``) and persists history to a hardcoded
+``data/notification_history.json``. Every test resets those globals and
+repoints ``_HISTORY_FILE`` to ``tmp_path`` so the suite never leaks state
+between tests or clobbers the real repo data file.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+import ledgrab.core.processing.os_notification_listener as nl
+from ledgrab.storage.color_strip_source import NotificationColorStripSource
+
+PLATFORM_MOD = "ledgrab.utils.platform"
+
+
+# ---------------------------------------------------------------------------
+# Test doubles
+# ---------------------------------------------------------------------------
+
+
+class _FakeStream:
+    """Stub NotificationColorStripStream — records fire() calls."""
+
+    def __init__(self, accept: bool = True):
+        self._accept = accept
+        self.fired_with: list = []
+
+    def fire(self, app_name=None) -> bool:
+        self.fired_with.append(app_name)
+        return self._accept
+
+
+class _FakeStore:
+    def __init__(self, sources):
+        self._sources = sources
+
+    def get_all_sources(self):
+        return list(self._sources)
+
+
+class _FakeStreamManager:
+    def __init__(self, streams):
+        self._streams = streams
+
+    def get_streams_by_source_id(self, source_id):
+        return list(self._streams)
+
+
+def _notif_source(
+    *, source_id: str = "css_test", os_listener: bool = True
+) -> NotificationColorStripSource:
+    now = datetime.now(timezone.utc)
+    return NotificationColorStripSource.create_from_kwargs(
+        id=source_id,
+        name="Test Notification Source",
+        source_type="notification",
+        created_at=now,
+        updated_at=now,
+        os_listener=os_listener,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Fixtures — module-global + disk isolation
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def nl_mod(monkeypatch, tmp_path):
+    """Reset module globals and repoint the history file to tmp_path.
+
+    ``monkeypatch.setattr`` auto-restores originals on teardown, so even though
+    ``start()``/``stop()`` rebind ``_android_target`` and ``_instance`` during a
+    test, the globals are returned to their pre-test values afterward — no
+    cross-test leakage and no write to the real repo ``data/`` dir.
+    """
+    monkeypatch.setattr(nl, "_android_target", None)
+    monkeypatch.setattr(nl, "_instance", None)
+    monkeypatch.setattr(nl, "_HISTORY_FILE", tmp_path / "notification_history.json")
+    return nl
+
+
+# ---------------------------------------------------------------------------
+# _AndroidBackend.probe()
+# ---------------------------------------------------------------------------
+
+
+def test_probe_true_on_android(nl_mod, monkeypatch):
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+    assert nl_mod._AndroidBackend.probe() is True
+
+
+def test_probe_false_on_desktop(nl_mod, monkeypatch):
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: False)
+    assert nl_mod._AndroidBackend.probe() is False
+
+
+# ---------------------------------------------------------------------------
+# push_notification() routing contract
+# ---------------------------------------------------------------------------
+
+
+def test_push_is_noop_before_start(nl_mod):
+    # _android_target is None → no callback, no exception.
+    nl_mod.push_notification("Telegram")  # must not raise
+
+
+def test_push_routes_after_start_and_stops_after_stop(nl_mod):
+    received: list = []
+    backend = nl_mod._AndroidBackend(on_notification=received.append)
+
+    backend.start()
+    nl_mod.push_notification("Telegram")
+    assert received == ["Telegram"]
+
+    backend.stop()
+    nl_mod.push_notification("Signal")  # no-op after stop
+    assert received == ["Telegram"]
+
+
+def test_push_swallows_callback_exception(nl_mod):
+    def boom(_app):
+        raise RuntimeError("callback exploded")
+
+    nl_mod._AndroidBackend(on_notification=boom).start()
+    # JNI entry point must never propagate — would crash the bound service.
+    nl_mod.push_notification("X")
+
+
+# ---------------------------------------------------------------------------
+# Integration — start() selects Android, push fires the stream + records history
+# ---------------------------------------------------------------------------
+
+
+def test_android_selected_push_fires_stream_and_records_history(nl_mod, monkeypatch, tmp_path):
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+    stream = _FakeStream(accept=True)
+    listener = nl_mod.OsNotificationListener(
+        _FakeStore([_notif_source(os_listener=True)]),
+        _FakeStreamManager([stream]),
+    )
+
+    listener.start()
+    assert listener.available is True  # flips True on backend selection, not on push
+
+    nl_mod.push_notification("Telegram")
+
+    assert stream.fired_with == ["Telegram"]
+    assert listener.recent_history[0]["app"] == "Telegram"
+    assert listener.recent_history[0]["fired"] == 1
+    # history written under tmp_path — never the repo data/ dir
+    assert nl_mod._HISTORY_FILE.exists()
+    assert nl_mod._HISTORY_FILE.parent == tmp_path
+
+    listener.stop()
+
+
+def test_push_with_none_app_name_is_recorded(nl_mod, monkeypatch):
+    # The Windows (_extract_app_name) and Linux D-Bus paths can yield None;
+    # the Android path falls back to the package name, but None must still be
+    # handled end-to-end without raising.
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+    stream = _FakeStream(accept=True)
+    listener = nl_mod.OsNotificationListener(
+        _FakeStore([_notif_source(os_listener=True)]),
+        _FakeStreamManager([stream]),
+    )
+
+    listener.start()
+    nl_mod.push_notification(None)
+
+    assert stream.fired_with == [None]
+    assert listener.recent_history[0]["app"] is None
+    listener.stop()
+
+
+def test_get_os_notification_listener_tracks_started_instance(nl_mod, monkeypatch):
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+    assert nl_mod.get_os_notification_listener() is None
+
+    listener = nl_mod.OsNotificationListener(_FakeStore([]), _FakeStreamManager([]))
+    listener.start()
+    assert nl_mod.get_os_notification_listener() is listener
+    listener.stop()
+
+
+def test_source_with_os_listener_off_does_not_fire(nl_mod, monkeypatch):
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: True)
+    stream = _FakeStream()
+    listener = nl_mod.OsNotificationListener(
+        _FakeStore([_notif_source(os_listener=False)]),
+        _FakeStreamManager([stream]),
+    )
+
+    listener.start()
+    nl_mod.push_notification("Telegram")
+
+    assert stream.fired_with == []  # os_listener=False → skipped
+    listener.stop()
+
+
+# ---------------------------------------------------------------------------
+# Desktop regression — the probe-order change must not alter desktop selection
+# ---------------------------------------------------------------------------
+
+
+def test_android_probe_false_on_real_desktop(nl_mod, monkeypatch):
+    # With is_android() False, the new first-in-tuple backend must not be selectable.
+    monkeypatch.setattr(f"{PLATFORM_MOD}.is_android", lambda: False)
+    assert nl_mod._AndroidBackend.probe() is False
+
+
+def test_desktop_selection_unchanged_windows_wins(nl_mod, monkeypatch):
+    # Deterministically control probes and stub start() so no real polling thread spawns.
+    # Order under test is (_AndroidBackend, _WindowsBackend, _LinuxBackend): Android skipped,
+    # Windows is the first True → it must be the selected backend, exactly as before.
+    monkeypatch.setattr(nl_mod._AndroidBackend, "probe", staticmethod(lambda: False))
+    monkeypatch.setattr(nl_mod._WindowsBackend, "probe", staticmethod(lambda: True))
+    monkeypatch.setattr(nl_mod._LinuxBackend, "probe", staticmethod(lambda: False))
+    started: list = []
+    monkeypatch.setattr(nl_mod._WindowsBackend, "start", lambda self: started.append("win"))
+
+    listener = nl_mod.OsNotificationListener(_FakeStore([]), _FakeStreamManager([]))
+    listener.start()
+
+    assert listener.available is True
+    assert isinstance(listener._backend, nl_mod._WindowsBackend)
+    assert started == ["win"]