fix(hypridle): keep displays awake during gamepad use (#480)

Assisted-by: DeepSeek Harness (gpt-6-astra)

home/linux/gui/base/hypridle/README.md

@@ -0,0 +1,63 @@
+# Idle activity protection
+
+Hypridle handles keyboard/pointer idle time, monitor power and locking. Its screen-off and lock
+listeners ignore application inhibitors because some applications keep them after playback stops.
+The `hypridle-activity` user service supplies an additional condition based on observed gamepad and
+media activity. It follows the same session target as hypridle and requires a Home Manager/system
+activation; editing these source files does not change the running desktop.
+
+## Activity and timing
+
+- Gamepads are identified by udev's `ID_INPUT_JOYSTICK` property and read through evdev without
+  grabbing the device. Discovery retries every five seconds, including after hotplug or permission
+  changes. Centered analog-stick noise, synchronization and release events do not count as activity.
+- Media is sampled every 15 seconds through PipeWire Pulse's output-monitor peak API and MPRIS. The
+  `pulsectl` library reads peak amplitudes, not recorded audio waveforms; no audio is saved.
+  Running, unmuted output sinks count only when their sampled peak exceeds 0.0001 (-80 dBFS), so a
+  browser keeping a silent stream open does not prevent idling. Sampling uses up to one second per
+  probe, rotating across outputs; the entire helper process has a two-second timeout. Only sink
+  monitors are sampled, never microphones. Playback must reach an output exposed by PipeWire Pulse.
+  An MPRIS player reporting `Playing` also counts, preserving silent-video playback protection.
+  Brief sounds between samples or sound below the threshold can be missed; virtual outputs may be
+  measured before the final hardware volume or mute.
+- Screen-off and locking each require their configured timeout since the last observed activity, in
+  addition to hypridle's keyboard/pointer timeout. Pausing playback or putting down the controller
+  therefore leaves an idle grace period. Sampling and hypridle's 30-second condition retry make
+  these approximate deadlines.
+
+The monitor writes atomic state under `$XDG_RUNTIME_DIR/hypridle-activity/` using monotonic time.
+Heartbeat updates remain every five seconds and do not count as activity. Gamepad events update
+activity immediately, independently of media sampling. Missing, malformed, future-dated or
+30-second-old heartbeat data does **not** block idle actions. This prevents a crashed monitor from
+disabling automatic locking indefinitely; detection failure can consequently allow screen-off during
+use. Starting the monitor does not manufacture activity. The monitor prevents automatic idle
+actions; it does not unlock the session or power on an already blanked display when a controller is
+pressed.
+
+## Verification
+
+Run the focused tests without hardware or third-party Python libraries:
+
+```console
+python3 -m unittest discover -s home/linux/gui/base/hypridle -p 'test_*.py' -v
+```
+
+After activating the configuration, inspect the service and its state:
+
+```console
+systemctl --user status hypridle-activity.service
+journalctl --user -u hypridle-activity.service -n 50 --no-pager
+hypridle-activity audio-peak
+hypridle-activity check --timeout 1200
+```
+
+The check exits **1** while activity blocks the requested timeout, and **0** when allowed. Use the
+host's screen-off or lock timeout as appropriate. Test a connected but untouched controller, actual
+controller input with the game muted, playback without keyboard input, and then no input/playback.
+Confirm that the state timestamp stops advancing in the last case. Keep normal keyboard access
+available for waking the display.
+
+If logs report permission denied for a controller, inspect its session/udev access rules. The
+service does not run as root and does not grant broad access to keyboard or mouse devices. An
+unchanging held control produces no new input events, and unusual controller layouts or drift
+outside the deadzone require hardware verification.

home/linux/gui/base/hypridle/activity_monitor.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""Track media and joystick activity for hypridle without creating inhibitors."""
+
+import argparse
+import json
+import logging
+import math
+import os
+from pathlib import Path
+import selectors
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+
+LOG = logging.getLogger("hypridle-activity")
+_FAILURES = {}
+
+
+def report_failure(key, message):
+    if _FAILURES.get(key) != message:
+        LOG.warning("%s", message)
+        _FAILURES[key] = message
+
+
+def recovered(key):
+    if _FAILURES.pop(key, None) is not None:
+        LOG.info("%s recovered", key)
+
+
+POLL_INTERVAL = 5.0  # Device discovery and heartbeat; input events remain immediate.
+MEDIA_INTERVAL = 15.0
+HEARTBEAT_TTL = 30.0
+COMMAND_TIMEOUT = 2.0
+AUDIO_THRESHOLD = 0.0001  # -80 dBFS: ignore digital silence and tiny background noise.
+PEAK_BUDGET = 1.0
+# Linux input-event-codes.h; keep check and tests independent of evdev/pyudev.
+EV_KEY, EV_ABS = 1, 3
+STICKS = {0, 1, 3, 4}  # ABS_X/Y/RX/RY
+TRIGGERS = {2, 5, 9, 10}  # ABS_Z/RZ/GAS/BRAKE
+HATS = set(range(16, 24))
+
+
+def state_path():
+    runtime = os.environ.get("XDG_RUNTIME_DIR")
+    if not runtime or not os.path.isabs(runtime):
+        raise ValueError("XDG_RUNTIME_DIR must be an absolute directory")
+    return Path(runtime) / "hypridle-activity" / "state.json"
+
+
+def timestamp(value):
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        return False
+    try:
+        return math.isfinite(value) and value >= 0
+    except OverflowError:
+        return False
+
+
+def blocked(state, timeout, now):
+    """Fail open for absent, stale, or invalid monitor state."""
+    if not isinstance(state, dict):
+        return False
+    heartbeat = state.get("heartbeat")
+    activity = state.get("last_activity")
+    if not timestamp(heartbeat) or heartbeat > now:
+        return False
+    if now - heartbeat >= HEARTBEAT_TTL:
+        return False
+    if activity is None:
+        return False
+    if not timestamp(activity) or activity > heartbeat:
+        return False
+    return now - activity < timeout
+
+
+def check(path, timeout, now=None):
+    try:
+        with path.open() as stream:
+            state = json.load(stream)
+    except (OSError, ValueError) as error:
+        LOG.warning("Cannot read activity state: %s", error)
+        return 0
+    return int(blocked(state, timeout, time.monotonic() if now is None else now))
+
+
+def write_state(path, last_activity, heartbeat):
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", dir=path.parent, prefix=".state-", delete=False
+        ) as stream:
+            temporary = stream.name
+            json.dump({"last_activity": last_activity, "heartbeat": heartbeat}, stream)
+            stream.write("\n")
+        os.replace(temporary, path)
+    finally:
+        if temporary is not None:
+            try:
+                os.unlink(temporary)
+            except FileNotFoundError:
+                pass
+
+
+def publish(path, activity, heartbeat):
+    try:
+        write_state(path, activity, heartbeat)
+        recovered("state")
+    except OSError as error:
+        report_failure("state", f"Cannot publish activity state: {error}")
+
+
+def audio_active(peak):
+    return timestamp(peak) and peak > AUDIO_THRESHOLD
+
+
+def output_peak(pulse):
+    """Sample output monitors only; a running sink can still contain silence."""
+    sinks = [
+        sink for sink in pulse.sink_list()
+        if sink.state == "running" and not sink.mute
+        and isinstance(sink.monitor_source, int)
+        and 0 <= sink.monitor_source < 0xFFFFFFFF
+    ]
+    if not sinks:
+        return 0.0
+    # Rotate the starting output so a large device list cannot starve later sinks.
+    now = time.monotonic()
+    start = int(now / MEDIA_INTERVAL) % len(sinks)
+    sinks = sinks[start:] + sinks[:start]
+    deadline = now + PEAK_BUDGET
+    peak = 0.0
+    for sink in sinks:
+        remaining = deadline - time.monotonic()
+        if remaining < 0.1:
+            break
+        value = pulse.get_peak_sample(sink.monitor_source, min(0.25, remaining))
+        if timestamp(value):
+            peak = max(peak, value)
+        if audio_active(peak):
+            break
+    return peak
+
+
+def probe_peak():
+    # Isolated by the caller's subprocess timeout, including connection/list calls.
+    import pulsectl
+
+    with pulsectl.Pulse("hypridle-activity", connect=False) as pulse:
+        pulse.connect(timeout=0.5)
+        return output_peak(pulse)
+
+
+def playing(output):
+    return any(line.strip() == "Playing" for line in output.splitlines())
+
+
+def media_active():
+    active = False
+    try:
+        result = subprocess.run(
+            [sys.executable, __file__, "audio-peak"],
+            capture_output=True, text=True, timeout=COMMAND_TIMEOUT, check=True,
+        )
+        active = audio_active(json.loads(result.stdout))
+        recovered("audio")
+    except (OSError, subprocess.SubprocessError, ValueError) as error:
+        report_failure("audio", f"Audio probe failed: {error}")
+    try:
+        result = subprocess.run(
+            ["playerctl", "-a", "status"],
+            capture_output=True, text=True, timeout=COMMAND_TIMEOUT, check=False,
+        )
+        # playerctl exits 1 when there are no players, an ordinary idle state.
+        if result.returncode in (0, 1):
+            recovered("players")
+            if result.returncode == 0:
+                active = active or playing(result.stdout)
+        else:
+            report_failure("players", f"Media player probe exited with {result.returncode}")
+    except (OSError, subprocess.SubprocessError) as error:
+        report_failure("players", f"Media player probe failed: {error}")
+    return active
+
+
+def gamepad_active(event, absinfo):
+    """Ignore synchronization, release, sensor, and centered-stick events."""
+    if event.type == EV_KEY:
+        return event.value in (1, 2)
+    if event.type != EV_ABS:
+        return False
+    if event.code in HATS:
+        return event.value != 0
+    if event.code not in STICKS | TRIGGERS:
+        return False
+    info = absinfo(event.code)
+    if info is None or info.max <= info.min:
+        return False
+    span = info.max - info.min
+    if event.code in STICKS:
+        center = (info.min + info.max) / 2
+        deadzone = max(info.flat, span * 0.15)
+        return abs(event.value - center) > deadzone
+    return event.value > info.min + max(info.flat, span * 0.05)
+
+
+class Joysticks:
+    def __init__(self, context, device_factory, selector):
+        self.context = context
+        self.device_factory = device_factory
+        self.selector = selector
+        self.devices = {}
+
+    def remove(self, path):
+        device = self.devices.pop(path)
+        self.selector.unregister(device.fd)
+        device.close()
+
+    def reconcile(self):
+        # Periodic enumeration also retries permissions and disconnected devices.
+        try:
+            paths = {
+                dev.device_node
+                for dev in self.context.list_devices(subsystem="input")
+                if dev.get("ID_INPUT_JOYSTICK") == "1"
+                and dev.device_node
+                and Path(dev.device_node).name.startswith("event")
+            }
+            recovered("discovery")
+        except Exception as error:
+            report_failure("discovery", f"Joystick discovery failed: {error}")
+            return
+        for path in set(self.devices) - paths:
+            self.remove(path)
+        for path in paths - self.devices.keys():
+            device = None
+            try:
+                device = self.device_factory(path)
+                # No grab(): Steam and games must continue receiving events.
+                self.selector.register(device.fd, selectors.EVENT_READ, path)
+                self.devices[path] = device
+                recovered(path)
+                LOG.info("Monitoring joystick %s", path)
+            except (OSError, ValueError) as error:
+                if device is not None:
+                    device.close()
+                report_failure(path, f"Cannot monitor joystick {path}: {error}")
+
+    def read(self, path):
+        device = self.devices.get(path)
+        if device is None:
+            return False
+        active = False
+        try:
+            # evdev.read() drains one bounded batch; the selector reports more.
+            for event in device.read():
+                active = gamepad_active(event, device.absinfo) or active
+        except BlockingIOError:
+            pass
+        except OSError as error:
+            LOG.info("Joystick unavailable %s: %s", path, error)
+            self.remove(path)
+        return active
+
+    def close(self):
+        for path in list(self.devices):
+            self.remove(path)
+
+
+def daemon(path):
+    # The check command must work without hardware libraries or device access.
+    import evdev
+    import pyudev
+
+    stopping = False
+
+    def stop(_signum, _frame):
+        nonlocal stopping
+        stopping = True
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    last_activity = None
+    last_event_publish = None
+    next_poll = 0.0
+    next_media = 0.0
+    with selectors.DefaultSelector() as selector:
+        joysticks = Joysticks(pyudev.Context(), evdev.InputDevice, selector)
+        try:
+            while not stopping:
+                now = time.monotonic()
+                if now >= next_poll:
+                    next_poll = now + POLL_INTERVAL
+                    joysticks.reconcile()
+                    if now >= next_media:
+                        next_media = now + MEDIA_INTERVAL
+                        if media_active():
+                            last_activity = time.monotonic()
+                    now = time.monotonic()
+                    publish(path, last_activity, now)
+                for key, _ in selector.select(max(0, next_poll - time.monotonic())):
+                    if joysticks.read(key.data):
+                        last_activity = time.monotonic()
+                        if last_event_publish is None or last_activity - last_event_publish >= 1:
+                            publish(path, last_activity, last_activity)
+                            last_event_publish = last_activity
+        finally:
+            joysticks.close()
+
+
+def positive_seconds(value):
+    number = float(value)
+    if not math.isfinite(number) or number <= 0:
+        raise argparse.ArgumentTypeError("timeout must be finite and positive")
+    return number
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest="command", required=True)
+    commands.add_parser("daemon")
+    commands.add_parser("audio-peak", help="print output peak amplitude without saving audio")
+    check_parser = commands.add_parser("check")
+    check_parser.add_argument("--timeout", type=positive_seconds, required=True)
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.INFO, format="%(name)s: %(levelname)s: %(message)s")
+    if args.command == "audio-peak":
+        print(json.dumps(probe_peak()))
+        return 0
+    try:
+        path = state_path()
+    except ValueError as error:
+        LOG.error("%s", error)
+        return 0 if args.command == "check" else 2
+    if args.command == "check":
+        return check(path, args.timeout)
+    daemon(path)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

home/linux/gui/base/hypridle/default.nix

@@ -1,10 +1,24 @@
 {
   config,
   lib,
+  pkgs,
   ...
 }:
 let
   cfg = config.modules.desktop.hypridle;
+  python = pkgs.python3.withPackages (ps: [
+    ps.evdev
+    ps.pulsectl
+    ps.pyudev
+  ]);
+  activityMonitor = pkgs.writeShellApplication {
+    name = "hypridle-activity";
+    runtimeInputs = [ pkgs.playerctl ];
+    text = ''
+      exec ${python}/bin/python3 ${./activity_monitor.py} "$@"
+    '';
+  };
+  activityCheck = timeout: "${lib.getExe activityMonitor} check --timeout ${toString timeout}";
 in
 {
   options.modules.desktop.hypridle = {
@@ -23,6 +37,27 @@ in
   };
 
   config = {
+    home.packages = [ activityMonitor ];
+    systemd.user.services.hypridle-activity = {
+      Unit = {
+        Description = "Gamepad and media activity monitor for hypridle";
+        After = [
+          config.services.hypridle.systemdTarget
+          "pipewire-pulse.service"
+        ];
+        PartOf = [ config.services.hypridle.systemdTarget ];
+      };
+      Service = {
+        ExecStart = "${lib.getExe activityMonitor} daemon";
+        Restart = "on-failure";
+        RestartSec = 3;
+        RuntimeDirectory = "hypridle-activity";
+        RuntimeDirectoryMode = "0700";
+        UMask = "0077";
+      };
+      Install.WantedBy = [ config.services.hypridle.systemdTarget ];
+    };
+
     # Hyprland/niri idle daemon. Declared in Nix (instead of a raw
     # hypridle.conf) so each host can tune the timeouts below.
     services.hypridle = {
@@ -48,11 +83,8 @@ in
           {
             timeout = cfg.screenOffTimeout;
             ignore_inhibit = true;
-            # Some apps keep idle-inhibit active even when they are only sitting
-            # on a page, which can prevent screen-off forever. Ignore those
-            # inhibitors for screen-off, but skip while an MPRIS player reports
-            # active playback.
-            condition_cmd = "! playerctl -a status 2>/dev/null | grep -q '^Playing$'";
+            # App inhibitors can outlive playback; use observed activity instead.
+            condition_cmd = activityCheck cfg.screenOffTimeout;
             condition_retry = 30;
             on-timeout = "niri msg action power-off-monitors";
             on-resume = "niri msg action power-on-monitors";
@@ -60,8 +92,7 @@ in
           {
             timeout = cfg.lockTimeout;
             ignore_inhibit = true;
-            # Skip while media is playing, same as screen-off above.
-            condition_cmd = "! playerctl -a status 2>/dev/null | grep -q '^Playing$'";
+            condition_cmd = activityCheck cfg.lockTimeout;
             condition_retry = 30;
             on-timeout = "noctalia msg session lock";
           }

home/linux/gui/base/hypridle/test_activity_monitor.py

@@ -0,0 +1,294 @@
+"""Regression tests for the hypridle activity policy and device lifecycle."""
+
+import argparse
+import json
+import os
+import sys
+import time
+from pathlib import Path
+import subprocess
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+import activity_monitor as monitor
+
+
+class StateTests(unittest.TestCase):
+    def test_activity_threshold_and_heartbeat_boundary(self):
+        state = {"heartbeat": 100, "last_activity": 80}
+        self.assertTrue(monitor.blocked(state, 21, 100))
+        self.assertFalse(monitor.blocked(state, 20, 100))
+        self.assertTrue(monitor.blocked(state, 100, 129.999))
+        self.assertFalse(monitor.blocked(state, 100, 130))
+
+    def test_invalid_and_future_state_never_blocks(self):
+        for state in [
+            None, [], {}, {"heartbeat": 100, "last_activity": None},
+            {"heartbeat": 101, "last_activity": 90},
+            {"heartbeat": 99, "last_activity": 100},
+            {"heartbeat": 100, "last_activity": -1},
+            {"heartbeat": True, "last_activity": 0},
+            {"heartbeat": 100, "last_activity": float("nan")},
+            {"heartbeat": float("inf"), "last_activity": 99},
+            {"heartbeat": "100", "last_activity": 99},
+        ]:
+            with self.subTest(state=state):
+                self.assertFalse(monitor.blocked(state, 1200, 100))
+
+    def test_state_publication_and_missing_corrupt_recovery(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "activity" / "state.json"
+            with self.assertLogs(monitor.LOG, level="WARNING"):
+                self.assertEqual(monitor.check(path, 20, 100), 0)
+            monitor.write_state(path, None, 100)
+            self.assertEqual(monitor.check(path, 20, 100), 0)
+            monitor.write_state(path, 90, 100)
+            self.assertEqual(monitor.check(path, 20, 100), 1)
+            self.assertEqual(list(path.parent.iterdir()), [path])
+            self.assertEqual(path.stat().st_mode & 0o777, 0o600)
+            path.write_text("{broken")
+            with self.assertLogs(monitor.LOG, level="WARNING"):
+                self.assertEqual(monitor.check(path, 20, 100), 0)
+            monitor.write_state(path, 100, 100)
+            self.assertEqual(json.loads(path.read_text())["heartbeat"], 100)
+
+    def test_cli_exit_statuses(self):
+        with tempfile.TemporaryDirectory() as runtime:
+            path = Path(runtime) / "hypridle-activity" / "state.json"
+            command = [sys.executable, "-B", monitor.__file__, "check", "--timeout", "1200"]
+            env = dict(os.environ, XDG_RUNTIME_DIR=runtime)
+            now = time.monotonic()
+            for activity, heartbeat, expected in [
+                (now, now, 1), (now - 1201, now, 0),
+                (now, now - 31, 0), (None, now, 0),
+            ]:
+                monitor.write_state(path, activity, heartbeat)
+                result = subprocess.run(command, env=env, capture_output=True, timeout=5)
+                self.assertEqual(result.returncode, expected, result.stderr)
+
+    def test_daemon_startup_has_no_fake_activity(self):
+        selector = Mock()
+        selector.select.side_effect = KeyboardInterrupt
+        manager = Mock()
+        manager.__enter__ = Mock(return_value=selector)
+        manager.__exit__ = Mock(return_value=False)
+        with tempfile.TemporaryDirectory() as runtime:
+            path = Path(runtime) / "state.json"
+            with (
+                patch.dict(sys.modules, {"evdev": Mock(), "pyudev": Mock()}),
+                patch.object(monitor.selectors, "DefaultSelector", return_value=manager),
+                patch.object(monitor, "Joysticks"),
+                patch.object(monitor, "media_active", return_value=False),
+                patch.object(monitor.signal, "signal"),
+                self.assertRaises(KeyboardInterrupt),
+            ):
+                monitor.daemon(path)
+            self.assertIsNone(json.loads(path.read_text())["last_activity"])
+
+    def test_first_gamepad_event_is_published_before_next_poll(self):
+        selector = Mock()
+        selector.select.side_effect = [[(SimpleNamespace(data="pad"), 1)], KeyboardInterrupt]
+        manager = Mock()
+        manager.__enter__ = Mock(return_value=selector)
+        manager.__exit__ = Mock(return_value=False)
+        with tempfile.TemporaryDirectory() as runtime:
+            path = Path(runtime) / "state.json"
+            with (
+                patch.dict(sys.modules, {"evdev": Mock(), "pyudev": Mock()}),
+                patch.object(monitor.selectors, "DefaultSelector", return_value=manager),
+                patch.object(monitor, "Joysticks") as joysticks,
+                patch.object(monitor, "media_active", return_value=False),
+                patch.object(monitor.time, "monotonic", return_value=100),
+                patch.object(monitor.signal, "signal"),
+                self.assertRaises(KeyboardInterrupt),
+            ):
+                joysticks.return_value.read.return_value = True
+                monitor.daemon(path)
+            self.assertEqual(json.loads(path.read_text())["last_activity"], 100)
+
+    def test_media_interval_preserves_heartbeat_and_input(self):
+        now = 0.0
+        media_times = []
+        selector = Mock()
+
+        def select(timeout):
+            nonlocal now
+            now += timeout
+            if now >= 20:
+                raise KeyboardInterrupt
+            return [(SimpleNamespace(data="pad"), 1)] if now == 5 else []
+
+        def media():
+            media_times.append(now)
+            return False
+
+        selector.select.side_effect = select
+        manager = Mock()
+        manager.__enter__ = Mock(return_value=selector)
+        manager.__exit__ = Mock(return_value=False)
+        with (
+            patch.dict(sys.modules, {"evdev": Mock(), "pyudev": Mock()}),
+            patch.object(monitor.selectors, "DefaultSelector", return_value=manager),
+            patch.object(monitor, "Joysticks") as joysticks,
+            patch.object(monitor, "media_active", side_effect=media),
+            patch.object(monitor, "publish") as publish,
+            patch.object(monitor.time, "monotonic", side_effect=lambda: now),
+            patch.object(monitor.signal, "signal"),
+            self.assertRaises(KeyboardInterrupt),
+        ):
+            joysticks.return_value.read.return_value = True
+            monitor.daemon(Path("unused"))
+        self.assertEqual(media_times, [0, 15])
+        self.assertEqual([call.args[2] for call in publish.call_args_list], [0, 5, 5, 10, 15])
+        self.assertEqual(publish.call_args_list[-1].args[1], 5)
+
+    def test_timeout_validation(self):
+        for value in ["0", "-1", "nan", "inf"]:
+            with self.subTest(value=value), self.assertRaises(argparse.ArgumentTypeError):
+                monitor.positive_seconds(value)
+        self.assertEqual(monitor.positive_seconds("20"), 20)
+
+
+class AudioTests(unittest.TestCase):
+    def setUp(self):
+        monitor._FAILURES.clear()
+
+    def test_error_log_deduplication_and_recovery(self):
+        with self.assertLogs(monitor.LOG, level="WARNING") as logs:
+            monitor.report_failure("audio", "unavailable")
+            monitor.report_failure("audio", "unavailable")
+        self.assertEqual(len(logs.output), 1)
+        monitor.recovered("audio")
+        with self.assertLogs(monitor.LOG, level="WARNING"):
+            monitor.report_failure("audio", "unavailable")
+
+    def test_silent_running_firefox_output_does_not_block(self):
+        pulse = Mock()
+        pulse.sink_list.return_value = [
+            SimpleNamespace(state="running", mute=False, monitor_source=42)
+        ]
+        pulse.get_peak_sample.return_value = 0.0
+        self.assertFalse(monitor.audio_active(monitor.output_peak(pulse)))
+        pulse.get_peak_sample.assert_called_once_with(42, 0.25)
+        pulse.get_peak_sample.return_value = 0.01
+        self.assertTrue(monitor.audio_active(monitor.output_peak(pulse)))
+
+    def test_peak_threshold_and_invalid_values(self):
+        self.assertTrue(monitor.audio_active(0.001))
+        for value in [0, 0.00001, monitor.AUDIO_THRESHOLD, -1, None, True,
+                      float("nan"), float("inf"), "0.1"]:
+            with self.subTest(value=value):
+                self.assertFalse(monitor.audio_active(value))
+
+    def test_only_valid_running_output_monitors_are_sampled(self):
+        pulse = Mock()
+        pulse.sink_list.return_value = [
+            SimpleNamespace(state="suspended", mute=False, monitor_source=10),
+            SimpleNamespace(state="running", mute=True, monitor_source=11),
+            SimpleNamespace(state="running", mute=False, monitor_source=0xFFFFFFFF),
+            SimpleNamespace(state="running", mute=False, monitor_source=None),
+        ]
+        self.assertEqual(monitor.output_peak(pulse), 0)
+        pulse.get_peak_sample.assert_not_called()
+        pulse.source_list.assert_not_called()
+
+    def test_multiple_outputs_and_total_sampling_budget(self):
+        pulse = Mock()
+        pulse.sink_list.return_value = [
+            SimpleNamespace(state="running", mute=False, monitor_source=i)
+            for i in range(6)
+        ]
+        pulse.get_peak_sample.return_value = 0
+        with patch.object(monitor.time, "monotonic", side_effect=[0, 0, 0.8, 1.1]):
+            self.assertEqual(monitor.output_peak(pulse), 0)
+        self.assertEqual(pulse.get_peak_sample.call_count, 2)
+        self.assertLessEqual(sum(c.args[1] for c in pulse.get_peak_sample.call_args_list), 1)
+
+    def test_mpris_requires_exact_playing_status(self):
+        self.assertTrue(monitor.playing("Paused\nPlaying\n"))
+        self.assertFalse(monitor.playing("Not Playing\nPaused\nStopped\n"))
+
+    @patch.object(monitor.subprocess, "run")
+    def test_timeout_does_not_hide_other_media_source(self, run):
+        run.side_effect = [
+            subprocess.TimeoutExpired("audio-peak", 2),
+            SimpleNamespace(returncode=0, stdout="Playing\n"),
+        ]
+        with self.assertLogs(monitor.LOG, level="WARNING"):
+            self.assertTrue(monitor.media_active())
+        for call in run.call_args_list:
+            self.assertEqual(call.kwargs["timeout"], monitor.COMMAND_TIMEOUT)
+
+    @patch.object(monitor.subprocess, "run")
+    def test_failed_probe_can_recover_next_poll(self, run):
+        run.side_effect = [
+            FileNotFoundError("python"), SimpleNamespace(returncode=1, stdout=""),
+            SimpleNamespace(stdout='0.01'),
+            SimpleNamespace(returncode=1, stdout=""),
+        ]
+        with self.assertLogs(monitor.LOG, level="WARNING"):
+            self.assertFalse(monitor.media_active())
+        self.assertTrue(monitor.media_active())
+
+
+class GamepadTests(unittest.TestCase):
+    def active(self, kind, code, value, minimum=-32768, maximum=32767, flat=0):
+        return monitor.gamepad_active(
+            SimpleNamespace(type=kind, code=code, value=value),
+            lambda _: SimpleNamespace(min=minimum, max=maximum, flat=flat),
+        )
+
+    def test_buttons_syn_and_unrelated_events(self):
+        self.assertTrue(self.active(1, 304, 1))
+        self.assertTrue(self.active(1, 304, 2))
+        self.assertFalse(self.active(1, 304, 0))
+        self.assertFalse(self.active(0, 0, 1))
+        self.assertFalse(self.active(4, 0, 1))
+        self.assertFalse(self.active(3, 40, 100))
+
+    def test_center_drift_and_unsigned_sticks(self):
+        self.assertFalse(self.active(3, 0, 1000))
+        self.assertTrue(self.active(3, 0, -20000))
+        self.assertTrue(self.active(3, 4, 20000))
+        self.assertFalse(self.active(3, 0, 128, 0, 255))
+        self.assertTrue(self.active(3, 0, 220, 0, 255))
+        self.assertFalse(self.active(3, 0, 15000, flat=20000))
+
+    def test_hat_trigger_and_invalid_axis(self):
+        self.assertFalse(self.active(3, 16, 0))
+        self.assertTrue(self.active(3, 16, -1))
+        self.assertFalse(self.active(3, 2, 2, 0, 255))
+        self.assertTrue(self.active(3, 2, 100, 0, 255))
+        self.assertFalse(self.active(3, 0, 1, 0, 0))
+
+    def test_reconcile_unplug_and_reconnect_without_grabbing(self):
+        context, selector, device = Mock(), Mock(), Mock()
+        node = Mock(device_node="/dev/input/event7")
+        node.get.return_value = "1"
+        context.list_devices.return_value = [node]
+        device.fd = 7
+        factory = Mock(return_value=device)
+        joysticks = monitor.Joysticks(context, factory, selector)
+        joysticks.reconcile()
+        device.grab.assert_not_called()
+        joysticks.reconcile()
+        factory.assert_called_once_with("/dev/input/event7")
+        device.read.return_value = [SimpleNamespace(type=0, code=0, value=0)]
+        self.assertFalse(joysticks.read(node.device_node))
+        device.read.return_value = [SimpleNamespace(type=1, code=304, value=1)]
+        self.assertTrue(joysticks.read(node.device_node))
+        device.read.side_effect = OSError("unplugged")
+        self.assertFalse(joysticks.read(node.device_node))
+        self.assertEqual(joysticks.devices, {})
+        device.close.assert_called_once()
+        joysticks.reconcile()
+        self.assertEqual(factory.call_count, 2)
+        context.list_devices.return_value = []
+        joysticks.reconcile()
+        self.assertEqual(joysticks.devices, {})
+
+
+if __name__ == "__main__":
+    unittest.main()

outputs/aarch64-linux/tests/home-manager/expected.nix

@@ -11,6 +11,8 @@ in
 lib.genAttrs hosts (_: {
   homeDirectory = "/home/${username}";
   hypridleScreenOffIgnoresInhibitors = true;
-  hypridleScreenOffSkipsPlayingMedia = true;
+  hypridleScreenOffTracksActivity = true;
+  hypridleLockTracksActivity = true;
+  hypridleActivitySession = true;
   hypridleLockIgnoresInhibitors = true;
 })

outputs/aarch64-linux/tests/home-manager/expr.nix

@@ -21,7 +21,16 @@ lib.genAttrs hosts (
   {
     homeDirectory = hm.home.homeDirectory;
     hypridleScreenOffIgnoresInhibitors = screenOff.ignore_inhibit or false;
-    hypridleScreenOffSkipsPlayingMedia = (screenOff.condition_cmd or "") != "";
+    hypridleScreenOffTracksActivity = lib.hasSuffix "check --timeout ${toString screenOff.timeout}" (
+      screenOff.condition_cmd or ""
+    );
+    hypridleLockTracksActivity = lib.hasSuffix "check --timeout ${toString lock.timeout}" (
+      lock.condition_cmd or ""
+    );
+    hypridleActivitySession =
+      hm.systemd.user.services.hypridle-activity.Install.WantedBy == [
+        hm.services.hypridle.systemdTarget
+      ];
     hypridleLockIgnoresInhibitors = lock.ignore_inhibit or false;
   }
 )