Add cross-desktop activity detection backends

Drop idle time polling and remove the GNOME idle monitor from activity detection, keeping observer mode selection based on screen lock or power save state.\n\nAdd freedesktop.ScreenSaver with GNOME ScreenSaver fallback for cross-desktop lock detection, and add KDE Solid PowerManagement as a power save fallback.\n\nUpdate both systemd unit definitions to pass XDG_CURRENT_DESKTOP through the user service environment, and add async fallback-chain tests with pytest-asyncio coverage.

contrib/solstone-linux.service

···
 [Service]
 Type=simple
 ExecStart=/usr/bin/solstone-linux run
-PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR
+PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR XDG_CURRENT_DESKTOP
 Restart=on-failure
 RestartSec=10
 StartLimitIntervalSec=300

pyproject.toml

···
 [dependency-groups]
 dev = [
     "pytest",
+    "pytest-asyncio",
     "ruff",
 ]
 

src/solstone_linux/activity.py

···
 
 """Activity detection using DBus APIs.
 
-Extracted from solstone's observe/gnome/activity.py. The DBus services
-probed here are GNOME-specific; on other desktops (KDE, etc.) they may
-not be available. Every function degrades gracefully — returning a safe
-default — so the observer keeps running regardless of desktop environment.
-
-Changes from monorepo version:
-- Replaces `from observe.utils import assign_monitor_positions` with local module
+Detects screen lock and power save state across GNOME and KDE desktops
+using freedesktop, GNOME, and KDE DBus interfaces with ordered fallback
+chains. Every function degrades gracefully — returning a safe default —
+so the observer keeps running regardless of desktop environment.
 """
 
 import logging
···
     _HAS_GTK = False
 
 # DBus service constants
-IDLE_MONITOR_BUS = "org.gnome.Mutter.IdleMonitor"
-IDLE_MONITOR_PATH = "/org/gnome/Mutter/IdleMonitor/Core"
-IDLE_MONITOR_IFACE = "org.gnome.Mutter.IdleMonitor"
+FDO_SCREENSAVER_BUS = "org.freedesktop.ScreenSaver"
+FDO_SCREENSAVER_PATH = "/ScreenSaver"
+FDO_SCREENSAVER_IFACE = "org.freedesktop.ScreenSaver"
 
-SCREENSAVER_BUS = "org.gnome.ScreenSaver"
-SCREENSAVER_PATH = "/org/gnome/ScreenSaver"
-SCREENSAVER_IFACE = "org.gnome.ScreenSaver"
+GNOME_SCREENSAVER_BUS = "org.gnome.ScreenSaver"
+GNOME_SCREENSAVER_PATH = "/org/gnome/ScreenSaver"
+GNOME_SCREENSAVER_IFACE = "org.gnome.ScreenSaver"
 
 DISPLAY_CONFIG_BUS = "org.gnome.Mutter.DisplayConfig"
 DISPLAY_CONFIG_PATH = "/org/gnome/Mutter/DisplayConfig"
 DISPLAY_CONFIG_IFACE = "org.gnome.Mutter.DisplayConfig"
 
+KDE_POWER_BUS = "org.kde.Solid.PowerManagement"
+KDE_POWER_PATH = "/org/kde/Solid/PowerManagement"
+KDE_POWER_IFACE = "org.kde.Solid.PowerManagement"
 
-async def probe_activity_services(bus: MessageBus) -> dict[str, bool]:
-    """Check which activity DBus services are reachable.
 
-    Returns a dict of service name -> available. Used for startup logging
-    only — the observer runs regardless of what's available.
-    """
+async def probe_activity_services(bus: MessageBus) -> dict[str, bool]:
+    """Check which activity DBus services are reachable."""
     services = {
-        "idle_monitor": IDLE_MONITOR_BUS,
-        "screensaver": SCREENSAVER_BUS,
-        "display_config": DISPLAY_CONFIG_BUS,
+        "fdo_screensaver": FDO_SCREENSAVER_BUS,
+        "gnome_screensaver": GNOME_SCREENSAVER_BUS,
+        "gnome_display_config": DISPLAY_CONFIG_BUS,
+        "kde_power": KDE_POWER_BUS,
     }
     results = {}
     for name, bus_name in services.items():
···
         except Exception:
             results[name] = False
 
-    available = [k for k, v in results.items() if v]
-    missing = [k for k, v in results.items() if not v]
-    if missing:
-        logger.warning(
-            "Activity signals unavailable: %s — observer will assume active",
-            ", ".join(missing),
-        )
-    if available:
-        logger.info("Activity signals available: %s", ", ".join(available))
-    if not available:
-        logger.warning(
-            "No activity signals available (non-GNOME desktop?) "
-            "— running in always-capture mode"
-        )
+    lock_backends = ["fdo_screensaver", "gnome_screensaver"]
+    power_backends = ["gnome_display_config", "kde_power"]
+
+    lock_available = [name for name in lock_backends if results.get(name)]
+    power_available = [name for name in power_backends if results.get(name)]
+
+    if lock_available:
+        logger.info("Screen lock backends: %s", ", ".join(lock_available))
+    else:
+        logger.warning("No screen lock backends available — will assume unlocked")
+
+    if power_available:
+        logger.info("Power save backends: %s", ", ".join(power_available))
+    else:
+        logger.warning("No power save backends available — will assume active display")
 
     results["gtk4"] = _HAS_GTK
     if not _HAS_GTK:
···
     return results
 
 
-async def get_idle_time_ms(bus: MessageBus) -> int:
-    """
-    Get the current idle time in milliseconds.
+async def is_screen_locked(bus: MessageBus) -> bool:
+    """Check if the screen is currently locked.
 
-    Args:
-        bus: Connected DBus session bus
-
-    Returns:
-        Idle time in milliseconds, or 0 if the service is unavailable
-        (0 = assume active, so the observer keeps capturing).
+    Tries freedesktop.ScreenSaver first (works on KDE, GNOME, and others),
+    then falls back to GNOME ScreenSaver.
     """
     try:
-        introspection = await bus.introspect(IDLE_MONITOR_BUS, IDLE_MONITOR_PATH)
-        proxy_obj = bus.get_proxy_object(
-            IDLE_MONITOR_BUS, IDLE_MONITOR_PATH, introspection
-        )
-        idle_monitor = proxy_obj.get_interface(IDLE_MONITOR_IFACE)
-        idle_time = await idle_monitor.call_get_idletime()
-        return idle_time
+        intro = await bus.introspect(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH)
+        obj = bus.get_proxy_object(FDO_SCREENSAVER_BUS, FDO_SCREENSAVER_PATH, intro)
+        iface = obj.get_interface(FDO_SCREENSAVER_IFACE)
+        return bool(await iface.call_get_active())
     except Exception:
-        return 0
-
+        pass
 
-async def is_screen_locked(bus: MessageBus) -> bool:
-    """
-    Check if the screen is currently locked using GNOME ScreenSaver.
-
-    Args:
-        bus: Connected DBus session bus
-
-    Returns:
-        True if screen is locked, False otherwise
-    """
     try:
-        intro = await bus.introspect(SCREENSAVER_BUS, SCREENSAVER_PATH)
-        obj = bus.get_proxy_object(SCREENSAVER_BUS, SCREENSAVER_PATH, intro)
-        iface = obj.get_interface(SCREENSAVER_IFACE)
+        intro = await bus.introspect(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH)
+        obj = bus.get_proxy_object(GNOME_SCREENSAVER_BUS, GNOME_SCREENSAVER_PATH, intro)
+        iface = obj.get_interface(GNOME_SCREENSAVER_IFACE)
         return bool(await iface.call_get_active())
     except Exception:
         return False
 
 
 async def is_power_save_active(bus: MessageBus) -> bool:
-    """
-    Check if display power save mode is active (screen blanked).
-
-    Args:
-        bus: Connected DBus session bus
+    """Check if display power save mode is active.
 
-    Returns:
-        True if power save is active, False otherwise
+    Tries GNOME Mutter DisplayConfig first (DPMS state), then falls back
+    to KDE Solid PowerManagement lid state.
     """
     try:
         intro = await bus.introspect(DISPLAY_CONFIG_BUS, DISPLAY_CONFIG_PATH)
···
         mode_variant = await iface.call_get(DISPLAY_CONFIG_IFACE, "PowerSaveMode")
         mode = int(mode_variant.value)
         return mode != 0
+    except Exception:
+        pass
+
+    try:
+        intro = await bus.introspect(KDE_POWER_BUS, KDE_POWER_PATH)
+        obj = bus.get_proxy_object(KDE_POWER_BUS, KDE_POWER_PATH, intro)
+        iface = obj.get_interface(KDE_POWER_IFACE)
+        return bool(await iface.call_is_lid_closed())
     except Exception:
         return False
 

src/solstone_linux/cli.py

···
 [Service]
 Type=simple
 ExecStart={binary} run
-PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR
+PassEnvironment=DISPLAY WAYLAND_DISPLAY DBUS_SESSION_BUS_ADDRESS XDG_RUNTIME_DIR XDG_CURRENT_DESKTOP
 Restart=on-failure
 RestartSec=10
 StartLimitIntervalSec=300

src/solstone_linux/observer.py

···
 from dbus_next.constants import BusType
 
 from .activity import (
-    get_idle_time_ms,
     is_power_save_active,
     is_screen_locked,
     probe_activity_services,
···
 PLATFORM = platform.system().lower()
 
 # Constants
-IDLE_THRESHOLD_MS = 5 * 60 * 1000  # 5 minutes
 RMS_THRESHOLD = 0.01
 MIN_HITS_FOR_SAVE = 3
 CHUNK_DURATION = 5  # seconds
···
 
         # Activity status cache (updated each loop)
         self.cached_is_active = False
-        self.cached_idle_time_ms = 0
         self.cached_screen_locked = False
         self.cached_is_muted = False
         self.cached_power_save = False
···
         self.audio_recorder.start_recording()
         logger.info("Audio recording started")
 
-        # Connect to DBus for idle/lock detection
+        # Connect to DBus for activity detection
         self.bus = await MessageBus(bus_type=BusType.SESSION).connect()
         logger.info("DBus connection established")
 
···
 
     async def check_activity_status(self) -> str:
         """Check system activity status and determine capture mode."""
-        idle_time = await get_idle_time_ms(self.bus)
         screen_locked = await is_screen_locked(self.bus)
         power_save = await is_power_save_active(self.bus)
         sink_muted = await is_sink_muted()
 
         # Cache values for status events
-        self.cached_idle_time_ms = idle_time
         self.cached_screen_locked = screen_locked
         self.cached_is_muted = sink_muted
         self.cached_power_save = power_save
 
         # Determine screen activity
-        screen_idle = (idle_time > IDLE_THRESHOLD_MS) or screen_locked or power_save
+        screen_idle = screen_locked or power_save
         screen_active = not screen_idle
 
         # Determine mode
···
         # Activity info
         activity_info = {
             "active": self.cached_is_active,
-            "idle_time_ms": self.cached_idle_time_ms,
             "screen_locked": self.cached_screen_locked,
             "sink_muted": self.cached_is_muted,
             "power_save": self.cached_power_save,

tests/test_activity.py

···
+# SPDX-License-Identifier: AGPL-3.0-only
+# Copyright (c) 2026 sol pbc
+
+"""Tests for desktop activity detection fallbacks."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from solstone_linux import activity
+
+
+def _make_proxy(interface_name: str, iface: object) -> MagicMock:
+    proxy = MagicMock()
+    proxy.get_interface.side_effect = lambda name: (
+        iface if name == interface_name else None
+    )
+    return proxy
+
+
[e-mail]
+async def test_is_screen_locked_prefers_fdo_true():
+    bus = MagicMock()
+    fdo_iface = AsyncMock()
+    fdo_iface.call_get_active.return_value = True
+    bus.introspect = AsyncMock(return_value="fdo-intro")
+    bus.get_proxy_object.return_value = _make_proxy(
+        activity.FDO_SCREENSAVER_IFACE, fdo_iface
+    )
+
+    result = await activity.is_screen_locked(bus)
+
+    assert result is True
+    bus.introspect.assert_awaited_once_with(
+        activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH
+    )
+    fdo_iface.call_get_active.assert_awaited_once_with()
+
+
[e-mail]
+async def test_is_screen_locked_prefers_fdo_false():
+    bus = MagicMock()
+    fdo_iface = AsyncMock()
+    fdo_iface.call_get_active.return_value = False
+    bus.introspect = AsyncMock(return_value="fdo-intro")
+    bus.get_proxy_object.return_value = _make_proxy(
+        activity.FDO_SCREENSAVER_IFACE, fdo_iface
+    )
+
+    result = await activity.is_screen_locked(bus)
+
+    assert result is False
+    bus.introspect.assert_awaited_once_with(
+        activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH
+    )
+    fdo_iface.call_get_active.assert_awaited_once_with()
+
+
[e-mail]
+async def test_is_screen_locked_falls_back_to_gnome():
+    bus = MagicMock()
+    gnome_iface = AsyncMock()
+    gnome_iface.call_get_active.return_value = True
+    bus.introspect = AsyncMock(
+        side_effect=[Exception("fdo unavailable"), "gnome-intro"]
+    )
+
+    def get_proxy_object(bus_name: str, path: str, intro: str) -> MagicMock:
+        assert (bus_name, path, intro) == (
+            activity.GNOME_SCREENSAVER_BUS,
+            activity.GNOME_SCREENSAVER_PATH,
+            "gnome-intro",
+        )
+        return _make_proxy(activity.GNOME_SCREENSAVER_IFACE, gnome_iface)
+
+    bus.get_proxy_object.side_effect = get_proxy_object
+
+    result = await activity.is_screen_locked(bus)
+
+    assert result is True
+    assert bus.introspect.await_args_list == [
+        ((activity.FDO_SCREENSAVER_BUS, activity.FDO_SCREENSAVER_PATH),),
+        ((activity.GNOME_SCREENSAVER_BUS, activity.GNOME_SCREENSAVER_PATH),),
+    ]
+    gnome_iface.call_get_active.assert_awaited_once_with()
+
+
[e-mail]
+async def test_is_screen_locked_returns_false_when_all_backends_fail():
+    bus = MagicMock()
+    bus.introspect = AsyncMock(side_effect=Exception("unavailable"))
+
+    result = await activity.is_screen_locked(bus)
+
+    assert result is False
+    assert bus.introspect.await_count == 2
+
+
[e-mail]
+async def test_is_power_save_active_uses_gnome_display_config_when_nonzero():
+    bus = MagicMock()
+    props_iface = AsyncMock()
+    props_iface.call_get.return_value = SimpleNamespace(value=2)
+    bus.introspect = AsyncMock(return_value="display-intro")
+    bus.get_proxy_object.return_value = _make_proxy(
+        "org.freedesktop.DBus.Properties", props_iface
+    )
+
+    result = await activity.is_power_save_active(bus)
+
+    assert result is True
+    bus.introspect.assert_awaited_once_with(
+        activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH
+    )
+    props_iface.call_get.assert_awaited_once_with(
+        activity.DISPLAY_CONFIG_IFACE, "PowerSaveMode"
+    )
+
+
[e-mail]
+async def test_is_power_save_active_uses_gnome_display_config_when_zero():
+    bus = MagicMock()
+    props_iface = AsyncMock()
+    props_iface.call_get.return_value = SimpleNamespace(value=0)
+    bus.introspect = AsyncMock(return_value="display-intro")
+    bus.get_proxy_object.return_value = _make_proxy(
+        "org.freedesktop.DBus.Properties", props_iface
+    )
+
+    result = await activity.is_power_save_active(bus)
+
+    assert result is False
+    bus.introspect.assert_awaited_once_with(
+        activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH
+    )
+    props_iface.call_get.assert_awaited_once_with(
+        activity.DISPLAY_CONFIG_IFACE, "PowerSaveMode"
+    )
+
+
[e-mail]
+async def test_is_power_save_active_falls_back_to_kde():
+    bus = MagicMock()
+    kde_iface = AsyncMock()
+    kde_iface.call_is_lid_closed.return_value = True
+    bus.introspect = AsyncMock(
+        side_effect=[Exception("gnome unavailable"), "kde-intro"]
+    )
+
+    def get_proxy_object(bus_name: str, path: str, intro: str) -> MagicMock:
+        assert (bus_name, path, intro) == (
+            activity.KDE_POWER_BUS,
+            activity.KDE_POWER_PATH,
+            "kde-intro",
+        )
+        return _make_proxy(activity.KDE_POWER_IFACE, kde_iface)
+
+    bus.get_proxy_object.side_effect = get_proxy_object
+
+    result = await activity.is_power_save_active(bus)
+
+    assert result is True
+    assert bus.introspect.await_args_list == [
+        ((activity.DISPLAY_CONFIG_BUS, activity.DISPLAY_CONFIG_PATH),),
+        ((activity.KDE_POWER_BUS, activity.KDE_POWER_PATH),),
+    ]
+    kde_iface.call_is_lid_closed.assert_awaited_once_with()
+
+
[e-mail]
+async def test_is_power_save_active_returns_false_when_all_backends_fail():
+    bus = MagicMock()
+    bus.introspect = AsyncMock(side_effect=Exception("unavailable"))
+
+    result = await activity.is_power_save_active(bus)
+
+    assert result is False
+    assert bus.introspect.await_count == 2
+
+
[e-mail]
+async def test_probe_activity_services_all_available():
+    bus = MagicMock()
+    bus.introspect = AsyncMock(return_value="intro")
+
+    results = await activity.probe_activity_services(bus)
+
+    assert results["fdo_screensaver"] is True
+    assert results["gnome_screensaver"] is True
+    assert results["gnome_display_config"] is True
+    assert results["kde_power"] is True
+    assert results["gtk4"] is activity._HAS_GTK
+
+
[e-mail]
+async def test_probe_activity_services_all_unavailable():
+    bus = MagicMock()
+    bus.introspect = AsyncMock(side_effect=Exception("unavailable"))
+
+    results = await activity.probe_activity_services(bus)
+
+    assert results["fdo_screensaver"] is False
+    assert results["gnome_screensaver"] is False
+    assert results["gnome_display_config"] is False
+    assert results["kde_power"] is False
+    assert results["gtk4"] is activity._HAS_GTK
+
+
[e-mail]
+async def test_probe_activity_services_mixed_availability():
+    bus = MagicMock()
+
+    async def introspect(bus_name: str, path: str) -> str:
+        if bus_name in {activity.FDO_SCREENSAVER_BUS, activity.KDE_POWER_BUS}:
+            return "intro"
+        raise Exception(f"{bus_name} unavailable")
+
+    bus.introspect = AsyncMock(side_effect=introspect)
+
+    results = await activity.probe_activity_services(bus)
+
+    assert results["fdo_screensaver"] is True
+    assert results["gnome_screensaver"] is False
+    assert results["gnome_display_config"] is False
+    assert results["kde_power"] is True
+    assert results["gtk4"] is activity._HAS_GTK