Update PKGBUILD

Bump to 0.4.2, add the glib2 dependency and drop the conflicts/replaces on xautolock-inhibit.

.gitignore

@@ -1 +1,5 @@
 CLAUDE.md
+__pycache__/
+.mypy_cache/
+.ruff_cache/
+.pytest_cache/

PKGBUILD

@@ -2,27 +2,38 @@
 
 # Maintainer: Daniele Fucini <dfucini [at] gmail [dot] com>
 
-pkgname=xautolock-inhibit
-pkgver=0.1.2
+pkgname=xidle-inhibit
+pkgver=0.4.2
 pkgrel=1
-pkgdesc='Honor org.freedesktop.ScreenSaver inhibit requests by disabling xautolock'
+pkgdesc='Honor org.freedesktop.ScreenSaver inhibit requests by blocking xautolock or xss-lock'
 arch=('any')
 url="https://git.shouldnt.work/fuxino/$pkgname"
 license=('GPL-3.0-or-later')
-depends=('libx11'
+depends=('glib2'
+         'libx11'
          'libxss'
          'python'
-         'python-gobject'
-         'xautolock')
+         'python-gobject')
+optdepends=('xautolock: xautolock backend'
+            'xss-lock: xss-lock backend')
 makedepends=('git')
+checkdepends=('dbus'
+              'python-pytest')
 source=("git+$url?signed#tag=$pkgver")
 validpgpkeys=('7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514')
-sha256sums=('6713c706e422cea0ae346fc7bd16611dd0e32487acf4d15ec3a4db325eaaa0c7')
+sha256sums=('e768123eee2cc9620c999d92c805794f8d45cb0f7ca3fdcfcf126ff83a13eaf6')
+
+check()
+{
+    cd "$srcdir"/"$pkgname"
+    pytest -p no:cacheprovider
+}
 
 package()
 {
     cd "$srcdir"/"$pkgname"
     install -Dm755 "$pkgname" "$pkgdir"/usr/bin/"$pkgname"
+    install -Dm755 xidle-dim "$pkgdir"/usr/bin/xidle-dim
     install -Dm644 "$pkgname".service "$pkgdir"/usr/lib/systemd/user/"$pkgname".service
     install -Dm644 README.md "$pkgdir"/usr/share/doc/"$pkgname"/README.md
 }

README.md

@@ -1,75 +1,122 @@
-# xautolock-inhibit
+# xidle-inhibit
 
-Makes screen locking with `xautolock` respect the "don't lock the screen" requests that apps send, on X11 sessions without a desktop environment (for example a plain window manager like xmonad).
+Makes screen locking with `xautolock` or `xss-lock` respect the "don't lock the screen" requests that apps send, on X11 sessions without a desktop environment (for example a plain window manager like xmonad).
 
-Browsers, Steam, gamemode and other apps ask the desktop not to lock or blank the screen while a video or game is running, through the D-Bus service `org.freedesktop.ScreenSaver`. Without a desktop environment nothing provides that service, so the requests are dropped and the screen locks in the middle of a video. xautolock-inhibit provides it:
+Browsers, Steam, gamemode and other apps ask the desktop not to lock or blank the screen while a video or game is running, through the D-Bus service `org.freedesktop.ScreenSaver`. Without a desktop environment nothing provides that service, so the requests are dropped and the screen locks in the middle of a video. xidle-inhibit provides it:
 
-- While at least one request is active, it runs `xautolock -disable` and pauses X screen blanking and DPMS.
-- When the last request is released, it waits a few seconds, then runs `xautolock -enable` and resumes blanking.
+- While at least one request is active, it pauses X screen blanking and DPMS, and with xautolock also runs `xautolock -disable`. xss-lock locks when the X screensaver starts, so pausing it is enough.
+- When the last request is released, it waits a few seconds, then resumes blanking (and runs `xautolock -enable`).
 
-xautolock stays your locker; xautolock-inhibit only switches it off and on.
+Your locker setup stays as it is; xidle-inhibit only holds it off.
+
+Formerly `xautolock-inhibit`, which supported only xautolock.
 
 ## How it works
 
-- Owns `org.freedesktop.ScreenSaver` on the session bus at both `/org/freedesktop/ScreenSaver` and `/ScreenSaver` (apps differ on the path), and implements `Inhibit` and `UnInhibit`.
+- Owns `org.freedesktop.ScreenSaver` on the session bus at both `/org/freedesktop/ScreenSaver` and `/ScreenSaver` (apps differ on the path), and implements `Inhibit`, `UnInhibit` and `Lock`.
 - Tracks requests per D-Bus client. If a client disconnects without releasing them (for example a browser crash), its requests are released.
-- Runs `xautolock` only when the number of active requests goes from 0 to 1 or back to 0. Re-enabling waits a few seconds (`--release-delay`), because browsers send short bursts of requests when a video starts.
+- Blocks and unblocks only when the number of active requests goes from 0 to 1 or back to 0. Unblocking waits a few seconds (`--release-delay`), because browsers send short bursts of requests when a video starts.
 - Pauses X blanking with `XScreenSaverSuspend`, the same mechanism mpv uses, instead of changing `xset` settings. The X server ends the pause by itself if the daemon exits or crashes.
 - Requests whose reason is on an ignore list (default: `Playing audio`) are accepted and logged, but don't block locking. Firefox-based browsers send one for audio playback, which would otherwise keep the screen unlocked while music plays with nobody at the computer.
-- On exit (including SIGTERM), it re-enables xautolock. The systemd unit also runs `xautolock -enable` when it stops, as a fallback.
+- Keeps the requests it holds in `$XDG_RUNTIME_DIR/xidle-inhibit.json`, so a restarted or crashed daemon takes them over. They are taken over only if the previous daemon stopped at most 30 seconds ago (`--restore-window`) on the same session bus, and only for clients still connected. A request released while no daemon was running can't be seen, so an old file isn't trusted.
+- With xautolock, it re-enables xautolock on exit (including SIGTERM). The systemd unit also runs `xidle-inhibit --cleanup` when it stops, which does the same, as a fallback.
+- `Lock` runs `xautolock -locknow`, or `loginctl lock-session` with xss-lock.
 - If another program (a desktop environment) already owns `org.freedesktop.ScreenSaver`, it exits with status 2 instead of taking over.
 
 ## Requirements
 
-- `xautolock`, already running in the session
+- `xautolock` or `xss-lock`, already running in the session
 - `python`, `python-gobject`
 - `libx11`, `libxss`
+- For `xidle-dim`: a compositor, such as picom
 
 ## Installation
 
-Build and install the package with the included PKGBUILD:
+Install [`xidle-inhibit`](https://aur.archlinux.org/packages/xidle-inhibit) from the AUR, for example with aurutils:
 
 ```
-makepkg -si
+aur sync xidle-inhibit
+pacman -S xidle-inhibit
 ```
 
-Then start it from your window manager's startup, after `xautolock`. For xmonad:
+The PKGBUILD builds the signed release tag, so import the signing key first:
+
+```
+gpg --recv-keys 7E12BC1FF3B6EDB2CD8053EB981A2B2A3BBF5514
+```
+
+Then start it from your window manager's startup, after the locker. For xmonad with xautolock:
 
 ```haskell
 spawnOnce "xautolock -time 10 -locker slock -detectsleep"
-spawnOnce "systemctl --user start xautolock-inhibit"
+spawnOnce "systemctl --user start xidle-inhibit"
 ```
 
+With xss-lock, the lock timeout is the X screensaver timeout (`xset s TIMEOUT CYCLE`). This dims the screen with `xidle-dim` at 9:50 and locks at 10:00:
+
+```haskell
+spawnOnce "xset s 590 10"
+spawnOnce "xss-lock -n xidle-dim -- slock"
+spawnOnce "systemctl --user start xidle-inhibit"
+```
+
+Without dimming, use `xset s 600 0` and `xss-lock -- slock`. Start xss-lock from the window manager, not from a terminal: terminals like kitty run commands outside the login session, and xss-lock then ignores `loginctl lock-session`.
+
 The unit has no `[Install]` section on purpose. Enabling it at login would also start it in a desktop environment, or before X is up; starting it from the window manager limits it to that session. The user service manager needs `DISPLAY` (and `XAUTHORITY`) in its environment, which `/etc/X11/xinit/xinitrc.d/50-systemd-user.sh` sets up for `startx` sessions.
 
+## Configuration
+
+The backend is set in `~/.config/xidle-inhibit/config` (`$XDG_CONFIG_HOME`):
+
+```
+backend = xss-lock
+```
+
+Values are `xautolock` and `xss-lock`. Without the file, xidle-inhibit uses whichever of the two is installed, and xautolock if both are. Restart the service after changing it.
+
 ## Usage
 
 ```
-xautolock-inhibit [-i REASON] [--count-all] [-d SECONDS]
+xidle-inhibit [-i REASON] [--count-all] [-d SECONDS] [--restore-window SECONDS] [--cleanup]
 ```
 
 - `-i`, `--ignore-reason REASON`: request reason that doesn't block locking. Can be repeated; replaces the default (`Playing audio`).
 - `--count-all`: every request blocks locking; no reasons are ignored.
-- `-d`, `--release-delay SECONDS`: wait this long after the last release before re-enabling xautolock (default 5).
+- `-d`, `--release-delay SECONDS`: wait this long after the last release before unblocking locking (default 5).
+- `--restore-window SECONDS`: take over requests from a previous daemon that stopped at most this long ago (default 30; 0 disables).
+- `--cleanup`: run `xautolock -enable` if xautolock is the backend, then exit. Used by the unit's `ExecStopPost`.
 
 To change the options used by the service:
 
 ```
-systemctl --user edit xautolock-inhibit.service
+systemctl --user edit xidle-inhibit.service
 ```
 
 ```ini
 [Service]
 ExecStart=
-ExecStart=/usr/bin/xautolock-inhibit --count-all
+ExecStart=/usr/bin/xidle-inhibit --count-all
 ```
 
-Every request, release and xautolock call is logged to the journal:
+Every request, release and locker command is logged to the journal:
 
 ```
-journalctl --user -u xautolock-inhibit
+journalctl --user -u xidle-inhibit
 ```
 
+### Dimming before the lock
+
+`xidle-dim` is an xss-lock notifier (`-n`). It fades a black window over the screen until xss-lock stops it, on user activity or when the locker starts. It needs a compositor such as picom and does nothing without one. Because it doesn't change gamma, it works alongside redshift.
+
+```
+xidle-dim [--opacity 0-1] [--fade SECONDS]
+```
+
+- `--opacity`: final darkness (default 0.6).
+- `--fade`: fade-in time (default 1.5).
+
+Options go inside the notifier command: `xss-lock -n "xidle-dim --opacity 0.8" -- slock`.
+
 ## App notes
 
 - **Sandboxed apps** may be blocked from reaching the service. For a firejail profile that filters D-Bus (`dbus-user filter`, as the Firefox-based profiles do), allow it in the app's `.local` file, e.g. `zen-browser.local`:
@@ -84,7 +131,8 @@ journalctl --user -u xautolock-inhibit
 
 ## Limitations
 
-- Restarting the daemon loses the requests it holds; apps don't send them again. Avoid restarting it while a video or game is running.
+- A release sent while the daemon is down is lost: if the daemon is back within `--restore-window`, that request stays active until its app exits. After a longer stop, all held requests are dropped and apps don't send them again.
+- With xss-lock, requests block only the idle timeout: suspending, or `loginctl lock-session` from elsewhere, still locks.
 - `SimulateUserActivity` is accepted but does nothing. Calls are logged once when a client starts sending them and once, with a count, after 5 minutes without any.
 - `GetActive` always returns `false`.
 

pyproject.toml

@@ -0,0 +1,14 @@
+[tool.ruff]
+line-length = 120
+extend-include = ["xidle-inhibit", "xidle-dim"]
+
+[tool.ruff.lint]
+select = ["E", "F", "W", "I", "B", "UP"]
+
+[tool.mypy]
+# Scripts both map to __main__, so check xidle-dim separately: mypy xidle-dim
+files = ["xidle-inhibit", "tests"]
+strict = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

tests/test_daemon.py

@@ -0,0 +1,432 @@
+"""End-to-end tests: run xidle-inhibit on a private D-Bus bus with fake lockers on PATH."""
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+from collections.abc import Callable, Iterator
+from pathlib import Path
+from typing import Any
+
+import pytest
+from gi.repository import Gio, GLib
+
+# XIDLE_INHIBIT tests another copy, e.g. /usr/bin/xidle-inhibit.
+SCRIPT = Path(os.environ.get("XIDLE_INHIBIT") or Path(__file__).resolve().parent.parent / "xidle-inhibit")
+BUS_NAME = "org.freedesktop.ScreenSaver"
+OBJECT_PATH = "/org/freedesktop/ScreenSaver"
+STATE_FILE = "xidle-inhibit.json"
+TIMEOUT = 5.0
+DELAY = ["--release-delay", "1"]
+SETTLE = 1.5  # longer than DELAY, to check that nothing happens
+
+ENABLE, DISABLE = "xautolock -enable", "xautolock -disable"
+
+# No <servicedir>: nothing is auto-started on this bus.
+BUS_CONFIG = """<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
+ "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+  <type>session</type>
+  <listen>unix:dir={dir}</listen>
+  <auth>EXTERNAL</auth>
+  <policy context="default">
+    <allow send_destination="*" eavesdrop="true"/>
+    <allow eavesdrop="true"/>
+    <allow own="*"/>
+  </policy>
+</busconfig>
+"""
+
+FAKE_TOOL = '#!/bin/sh\necho "${0##*/} $*" >> "$FAKE_LOG"\n'
+
+
+def wait_for(cond: Callable[[], bool], what: str, timeout: float = TIMEOUT) -> None:
+    deadline = time.monotonic() + timeout
+    while not cond():
+        if time.monotonic() > deadline:
+            raise AssertionError(f"timed out waiting for {what}")
+        time.sleep(0.05)
+
+
+class Bus:
+    def __init__(self, root: Path) -> None:
+        sock_dir = root / "sock"
+        sock_dir.mkdir(parents=True)
+        config = root / "bus.conf"
+        config.write_text(BUS_CONFIG.format(dir=sock_dir))
+        self.proc = subprocess.Popen(
+            ["dbus-daemon", f"--config-file={config}", "--nofork", "--print-address=1"],
+            stdout=subprocess.PIPE, text=True,
+        )
+        assert self.proc.stdout is not None
+        self.address = self.proc.stdout.readline().strip()
+        assert self.address, "dbus-daemon printed no address"
+
+    def connect(self) -> Gio.DBusConnection:
+        flags = Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION
+        return Gio.DBusConnection.new_for_address_sync(self.address, flags, None, None)
+
+    def stop(self) -> None:
+        self.proc.terminate()
+        self.proc.wait(TIMEOUT)
+
+
+class Client:
+    """One bus connection, so one D-Bus sender."""
+
+    def __init__(self, bus: Bus) -> None:
+        self.conn = bus.connect()
+        name = self.conn.get_unique_name()
+        assert name is not None
+        self.name = name
+
+    def call(
+        self, method: str, args: GLib.Variant | None = None, reply: str | None = None,
+        path: str = OBJECT_PATH, dest: str = BUS_NAME, iface: str = BUS_NAME,
+    ) -> Any:
+        result = self.conn.call_sync(
+            dest, path, iface, method, args, GLib.VariantType(reply) if reply else None,
+            Gio.DBusCallFlags.NO_AUTO_START, int(TIMEOUT * 1000), None,
+        )
+        return result.unpack()
+
+    def inhibit(self, reason: str = "Playing video", path: str = OBJECT_PATH) -> int:
+        return int(self.call("Inhibit", GLib.Variant("(ss)", ("test-app", reason)), "(u)", path)[0])
+
+    def uninhibit(self, cookie: int, path: str = OBJECT_PATH) -> None:
+        self.call("UnInhibit", GLib.Variant("(u)", (cookie,)), path=path)
+
+    def has_owner(self, name: str) -> bool:
+        return bool(self.call(
+            "NameHasOwner", GLib.Variant("(s)", (name,)), "(b)",
+            "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus",
+        )[0])
+
+    def close(self) -> None:
+        if not self.conn.is_closed():
+            self.conn.close_sync(None)
+
+
+class Sandbox:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self.bin = root / "bin"
+        self.runtime = root / "run"
+        self.config = root / "config"
+        for d in (self.bin, self.runtime, self.config / "xidle-inhibit"):
+            d.mkdir(parents=True)
+        for tool in ("xautolock", "xss-lock", "loginctl"):
+            path = self.bin / tool
+            path.write_text(FAKE_TOOL)
+            path.chmod(0o755)
+        self.fake_log = root / "fake.log"
+        self.daemon_log = root / "daemon.log"
+        self.state_file = self.runtime / STATE_FILE
+        self.buses: list[Bus] = []
+        self.clients: list[Client] = []
+        self.daemons: list[subprocess.Popen[bytes]] = []
+        self.bus = self.new_bus()
+        self.set_backend("xautolock")
+
+    def new_bus(self) -> Bus:
+        bus = Bus(self.root / f"bus{len(self.buses)}")
+        self.buses.append(bus)
+        return bus
+
+    def client(self, bus: Bus | None = None) -> Client:
+        client = Client(bus or self.bus)
+        self.clients.append(client)
+        return client
+
+    def set_backend(self, backend: str | None) -> None:
+        config = self.config / "xidle-inhibit" / "config"
+        if backend is None:
+            config.unlink(missing_ok=True)
+        else:
+            config.write_text(f"backend = {backend}\n")
+
+    def environ(self, bus: Bus) -> dict[str, str]:
+        env = dict(os.environ)
+        for var in ("DISPLAY", "WAYLAND_DISPLAY"):
+            env.pop(var, None)
+        env.update(
+            PATH=str(self.bin), DBUS_SESSION_BUS_ADDRESS=bus.address, FAKE_LOG=str(self.fake_log),
+            XDG_RUNTIME_DIR=str(self.runtime), XDG_CONFIG_HOME=str(self.config),
+        )
+        return env
+
+    def start(self, *args: str, bus: Bus | None = None, wait: bool = True) -> subprocess.Popen[bytes]:
+        """Start the daemon; with wait, return once it has taken the name and set the locker."""
+        bus = bus or self.bus
+        with open(self.daemon_log, "ab") as log:
+            proc = subprocess.Popen(
+                [sys.executable, str(SCRIPT), *args], env=self.environ(bus), stdout=log, stderr=subprocess.STDOUT,
+            )
+        self.daemons.append(proc)
+        if wait:
+            probe = Client(bus)
+            try:
+                # Method calls are handled after start() returns, so a reply means startup is done.
+                def answers() -> bool:
+                    assert proc.poll() is None, f"daemon exited:\n{self.log()}"
+                    try:
+                        probe.call("GetActive", reply="(b)")
+                        return True
+                    except GLib.Error:
+                        return False
+                wait_for(answers, "the daemon to own the name")
+            finally:
+                probe.close()
+        return proc
+
+    def stop(self, sig: int = signal.SIGTERM) -> int:
+        proc = self.daemons[-1]
+        proc.send_signal(sig)
+        return proc.wait(TIMEOUT)
+
+    def calls(self) -> list[str]:
+        try:
+            return self.fake_log.read_text().splitlines()
+        except FileNotFoundError:
+            return []
+
+    def wait_calls(self, expected: list[str]) -> None:
+        wait_for(lambda: self.calls() == expected, f"calls {expected}, got {self.calls()}")
+
+    def log(self) -> str:
+        return self.daemon_log.read_text()
+
+    def close(self) -> None:
+        for proc in self.daemons:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait()
+        for client in self.clients:
+            client.close()
+        for bus in self.buses:
+            bus.stop()
+
+
+@pytest.fixture
+def sb(tmp_path: Path) -> Iterator[Sandbox]:
+    sandbox = Sandbox(tmp_path)
+    yield sandbox
+    sandbox.close()
+
+
+# --- blocking and unblocking ---------------------------------------------------
+
+
+def test_blocks_once_and_unblocks_after_last_release(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    assert sb.calls() == [ENABLE]  # known state at startup
+    a = sb.client()
+    first, second = a.inhibit(), a.inhibit()
+    assert second == first + 1
+    assert sb.calls() == [ENABLE, DISABLE]
+    a.uninhibit(first)
+    time.sleep(SETTLE)
+    assert sb.calls() == [ENABLE, DISABLE]
+    a.uninhibit(second)
+    sb.wait_calls([ENABLE, DISABLE, ENABLE])
+
+
+def test_burst_within_release_delay_stays_blocked(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    a = sb.client()
+    a.uninhibit(a.inhibit())
+    a.inhibit()
+    time.sleep(SETTLE)
+    assert sb.calls() == [ENABLE, DISABLE]
+
+
+def test_both_object_paths_share_cookies(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    a = sb.client()
+    cookie = a.inhibit(path="/ScreenSaver")
+    assert sb.calls() == [ENABLE, DISABLE]
+    a.uninhibit(cookie, path=OBJECT_PATH)
+    sb.wait_calls([ENABLE, DISABLE, ENABLE])
+
+
+def test_ignored_reason_does_not_block(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    sb.client().inhibit(reason="Playing audio")
+    assert sb.calls() == [ENABLE]
+    assert "reason ignored" in sb.log()
+
+
+def test_count_all_blocks_on_any_reason(sb: Sandbox) -> None:
+    sb.start(*DELAY, "--count-all")
+    sb.client().inhibit(reason="Playing audio")
+    assert sb.calls() == [ENABLE, DISABLE]
+
+
+def test_uninhibit_from_another_sender_is_ignored(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    a, b = sb.client(), sb.client()
+    b.uninhibit(a.inhibit())
+    time.sleep(SETTLE)
+    assert sb.calls() == [ENABLE, DISABLE]
+    assert "unknown cookie, ignored" in sb.log()
+
+
+def test_sender_leaving_the_bus_releases_its_inhibits(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    a, b = sb.client(), sb.client()
+    a.inhibit()
+    a.inhibit()
+    b.inhibit(reason="Playing audio")  # another sender's cookie stays
+    a.close()
+    sb.wait_calls([ENABLE, DISABLE, ENABLE])
+    assert sb.log().count("left the bus") == 2
+
+
+def test_xss_lock_backend_runs_no_locker_commands(sb: Sandbox) -> None:
+    sb.set_backend("xss-lock")
+    sb.start(*DELAY)
+    a = sb.client()
+    a.uninhibit(a.inhibit())
+    time.sleep(SETTLE)
+    assert sb.calls() == []
+    assert sb.stop() == 0
+    assert sb.calls() == []
+
+
+@pytest.mark.parametrize(("backend", "command"), [
+    ("xautolock", "xautolock -locknow"),
+    ("xss-lock", "loginctl lock-session"),
+])
+def test_lock_and_stub_methods(sb: Sandbox, backend: str, command: str) -> None:
+    sb.set_backend(backend)
+    sb.start()
+    a = sb.client()
+    before = sb.calls()
+    a.call("Lock")
+    assert sb.calls() == [*before, command]
+    assert a.call("GetActive", reply="(b)") == (False,)
+    assert a.call("GetSessionIdleTime", reply="(u)") == (0,)
+
+
+def test_stop_unblocks(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    sb.client().inhibit()
+    assert sb.stop() == 0
+    assert sb.calls() == [ENABLE, DISABLE, ENABLE]
+
+
+# --- restoring after a restart -------------------------------------------------
+
+
+def test_restart_restores_inhibits(sb: Sandbox) -> None:
+    sb.start(*DELAY)
+    a = sb.client()
+    old = a.inhibit()
+    assert sb.stop() == 0
+    sb.start(*DELAY)
+    assert sb.calls() == [ENABLE, DISABLE, ENABLE, DISABLE]
+    new = a.inhibit()
+    assert new == old + 1  # numbering continues, so a late UnInhibit can't hit a new cookie
+    a.uninhibit(old)
+    a.uninhibit(new)
+    sb.wait_calls([ENABLE, DISABLE, ENABLE, DISABLE, ENABLE])
+
+
+def test_restore_skips_senders_that_left(sb: Sandbox) -> None:
+    sb.start()
+    a, probe = sb.client(), sb.client()
+    a.inhibit()
+    sb.stop()
+    a.close()
+    wait_for(lambda: not probe.has_owner(a.name), "the client to leave the bus")
+    sb.start()
+    assert sb.calls() == [ENABLE, DISABLE, ENABLE, ENABLE]
+    assert f"({a.name} left the bus)" in sb.log()
+
+
+def test_no_restore_outside_window(sb: Sandbox) -> None:
+    sb.start()
+    sb.client().inhibit()
+    sb.stop()
+    sb.start("--restore-window", "0")
+    assert sb.calls()[-1] == ENABLE
+    assert "over the 0s window" in sb.log()
+
+
+def test_no_restore_from_another_bus(sb: Sandbox) -> None:
+    sb.start()
+    sb.client().inhibit()
+    sb.stop()
+    other = sb.new_bus()
+    sb.start(bus=other)
+    assert sb.calls()[-1] == ENABLE
+    assert "saved on another session bus" in sb.log()
+    assert sb.client(other).inhibit() == 1  # cookie numbering is per bus
+
+
+def test_crash_restores_by_mtime(sb: Sandbox) -> None:
+    sb.start()
+    sb.client().inhibit()
+    sb.stop(signal.SIGKILL)
+    assert json.loads(sb.state_file.read_text())["stopped_at"] is None
+    sb.start()
+    assert sb.calls() == [ENABLE, DISABLE, DISABLE]
+
+
+def test_old_crash_is_not_restored(sb: Sandbox) -> None:
+    sb.start()
+    sb.client().inhibit()
+    sb.stop(signal.SIGKILL)
+    past = time.time() - 60
+    os.utime(sb.state_file, (past, past))
+    sb.start()
+    assert sb.calls()[-1] == ENABLE
+    assert "over the 30s window" in sb.log()
+
+
+def test_unreadable_state_file_is_ignored(sb: Sandbox) -> None:
+    sb.state_file.write_text("{not json")
+    sb.start()
+    assert sb.calls() == [ENABLE]
+    assert "Ignoring unreadable state file" in sb.log()
+    assert json.loads(sb.state_file.read_text())["cookies"] == []
+
+
+# --- startup and cleanup -------------------------------------------------------
+
+
+def test_exits_2_if_name_is_owned(sb: Sandbox) -> None:
+    other = sb.client()
+    do_not_queue = 4
+    reply = other.call(
+        "RequestName", GLib.Variant("(su)", (BUS_NAME, do_not_queue)), "(u)",
+        "/org/freedesktop/DBus", "org.freedesktop.DBus", "org.freedesktop.DBus",
+    )
+    assert reply == (1,)  # primary owner
+    proc = sb.start(wait=False)
+    assert proc.wait(TIMEOUT) == 2
+    assert sb.calls() == []
+    assert not sb.state_file.exists()
+
+
+@pytest.mark.parametrize(("backend", "xautolock_installed", "expected"), [
+    ("xautolock", True, [ENABLE]),
+    ("'xss-lock'", True, []),
+    (None, True, [ENABLE]),  # both installed: xautolock
+    (None, False, []),  # only xss-lock
+    ("bogus", False, []),
+])
+def test_cleanup(sb: Sandbox, backend: str | None, xautolock_installed: bool, expected: list[str]) -> None:
+    sb.set_backend(backend)
+    if not xautolock_installed:
+        (sb.bin / "xautolock").unlink()
+    result = subprocess.run(
+        [sys.executable, str(SCRIPT), "--cleanup"], env=sb.environ(sb.bus),
+        capture_output=True, text=True, timeout=TIMEOUT,
+    )
+    assert result.returncode == 0
+    assert sb.calls() == expected
+    if backend == "bogus":
+        assert "unknown backend 'bogus'" in result.stdout

xautolock-inhibit.service

@@ -1,10 +0,0 @@
-[Unit]
-Description=Turn org.freedesktop.ScreenSaver inhibit requests into xautolock control
-
-[Service]
-Type=dbus
-BusName=org.freedesktop.ScreenSaver
-ExecStart=/usr/bin/xautolock-inhibit
-ExecStopPost=-/usr/bin/xautolock -enable
-Restart=on-failure
-RestartPreventExitStatus=2

xidle-dim

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Dim the screen until killed, as an xss-lock notifier: xss-lock -n xidle-dim -- slock.
+
+Maps a black window over the whole screen and fades in its opacity through the compositor; gamma is left alone,
+so it doesn't clash with redshift. Without a compositor it does nothing, as the window would be solid black.
+"""
+
+import argparse
+import ctypes
+import signal
+import sys
+import time
+from types import FrameType
+
+CW_BACK_PIXEL = 1 << 1
+CW_OVERRIDE_REDIRECT = 1 << 9
+INPUT_OUTPUT = 1
+XA_CARDINAL = 6
+PROP_MODE_REPLACE = 0
+OPACITY_MAX = 0xFFFFFFFF
+
+
+class XSetWindowAttributes(ctypes.Structure):
+    _fields_ = [
+        ("background_pixmap", ctypes.c_ulong),
+        ("background_pixel", ctypes.c_ulong),
+        ("border_pixmap", ctypes.c_ulong),
+        ("border_pixel", ctypes.c_ulong),
+        ("bit_gravity", ctypes.c_int),
+        ("win_gravity", ctypes.c_int),
+        ("backing_store", ctypes.c_int),
+        ("backing_planes", ctypes.c_ulong),
+        ("backing_pixel", ctypes.c_ulong),
+        ("save_under", ctypes.c_int),
+        ("event_mask", ctypes.c_long),
+        ("do_not_propagate_mask", ctypes.c_long),
+        ("override_redirect", ctypes.c_int),
+        ("colormap", ctypes.c_ulong),
+        ("cursor", ctypes.c_ulong),
+    ]
+
+
+def load_x11() -> ctypes.CDLL:
+    x11 = ctypes.CDLL("libX11.so.6")
+    p, ul, i = ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int
+    x11.XOpenDisplay.argtypes = [ctypes.c_char_p]
+    x11.XOpenDisplay.restype = p
+    for name in ("XDefaultScreen",):
+        getattr(x11, name).argtypes = [p]
+    for name in ("XRootWindow", "XBlackPixel"):
+        getattr(x11, name).argtypes = [p, i]
+        getattr(x11, name).restype = ul
+    for name in ("XDisplayWidth", "XDisplayHeight"):
+        getattr(x11, name).argtypes = [p, i]
+    x11.XInternAtom.argtypes = [p, ctypes.c_char_p, i]
+    x11.XInternAtom.restype = ul
+    x11.XGetSelectionOwner.argtypes = [p, ul]
+    x11.XGetSelectionOwner.restype = ul
+    x11.XCreateWindow.argtypes = [p, ul, i, i, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint, i, ctypes.c_uint, p, ul,
+                                  ctypes.POINTER(XSetWindowAttributes)]
+    x11.XCreateWindow.restype = ul
+    x11.XChangeProperty.argtypes = [p, ul, ul, ul, i, i, p, i]
+    x11.XMapRaised.argtypes = [p, ul]
+    x11.XFlush.argtypes = [p]
+    return x11
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--opacity", type=float, default=0.6, help="final darkness, 0 to 1 (default: %(default)s)")
+    parser.add_argument("--fade", type=float, default=1.5, help="fade-in time in seconds (default: %(default)s)")
+    args = parser.parse_args()
+    if not 0 <= args.opacity <= 1 or args.fade < 0:
+        parser.error("--opacity must be between 0 and 1 and --fade not negative")
+    return args
+
+
+def exit_on_signal(_signum: int, _frame: FrameType | None) -> None:
+    sys.exit(0)  # the X server removes the window when the connection closes
+
+
+def main() -> None:
+    args = parse_args()
+    # xss-lock sends SIGHUP on user activity and SIGTERM when the locker starts.
+    for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT):
+        signal.signal(sig, exit_on_signal)
+
+    x11 = load_x11()
+    dpy = x11.XOpenDisplay(None)
+    if not dpy:
+        sys.exit("xidle-dim: cannot open display")
+    screen = x11.XDefaultScreen(dpy)
+    if not x11.XGetSelectionOwner(dpy, x11.XInternAtom(dpy, f"_NET_WM_CM_S{screen}".encode(), False)):
+        print("xidle-dim: no compositor running, not dimming", file=sys.stderr)
+        signal.pause()  # stay alive like a notifier should, until xss-lock kills us
+
+    attrs = XSetWindowAttributes(background_pixel=x11.XBlackPixel(dpy, screen), override_redirect=True)
+    win = x11.XCreateWindow(dpy, x11.XRootWindow(dpy, screen), 0, 0, x11.XDisplayWidth(dpy, screen),
+                            x11.XDisplayHeight(dpy, screen), 0, 0, INPUT_OUTPUT, None,
+                            CW_BACK_PIXEL | CW_OVERRIDE_REDIRECT, ctypes.byref(attrs))
+    opacity_atom = x11.XInternAtom(dpy, b"_NET_WM_WINDOW_OPACITY", False)
+
+    def set_opacity(level: float) -> None:
+        value = ctypes.c_ulong(int(level * OPACITY_MAX))  # format 32 properties are passed as longs
+        x11.XChangeProperty(dpy, win, opacity_atom, XA_CARDINAL, 32, PROP_MODE_REPLACE, ctypes.byref(value), 1)
+        x11.XFlush(dpy)
+
+    set_opacity(0)
+    x11.XMapRaised(dpy, win)
+    steps = max(1, int(args.fade * 30))
+    for n in range(1, steps + 1):
+        set_opacity(args.opacity * n / steps)
+        time.sleep(args.fade / steps)
+    while True:
+        signal.pause()
+
+
+if __name__ == "__main__":
+    main()

xidle-inhibit

@@ -1,18 +1,22 @@
 #!/usr/bin/env python3
-"""Own org.freedesktop.ScreenSaver and turn inhibit requests into xautolock -disable/-enable.
+"""Own org.freedesktop.ScreenSaver and block screen locking by xautolock or xss-lock while inhibits are held.
 
-While an inhibit is active, X screen blanking is also paused (like mpv does).
+While an inhibit is active, X screen blanking is paused (like mpv does), which also keeps xss-lock from
+locking; with xautolock, it is also switched off with xautolock -disable.
 """
 
 import argparse
 import ctypes
-import itertools
+import json
+import os
+import shutil
 import signal
 import subprocess
 import sys
 import time
 from dataclasses import dataclass
 from datetime import timedelta
+from typing import Any
 
 from gi.repository import Gio, GLib
 
@@ -26,6 +30,12 @@ BUS_NAME = "org.freedesktop.ScreenSaver"
 OBJECT_PATHS = ("/org/freedesktop/ScreenSaver", "/ScreenSaver")  # Vivaldi uses the first, Zen the second
 EXIT_NAME_TAKEN = 2
 ACTIVITY_QUIET = 300  # seconds without SimulateUserActivity that end a logged burst
+STATE_FILE = "xidle-inhibit.json"  # in $XDG_RUNTIME_DIR
+STATE_HEARTBEAT = 10  # seconds between state file touches while inhibits are held
+CONFIG_FILE = "xidle-inhibit/config"  # in $XDG_CONFIG_HOME
+BACKENDS = ("xautolock", "xss-lock")
+
+Cookie = tuple[str, str, str, bool]  # sender, application_name, reason, counts
 
 INTROSPECTION = """
 <node>
@@ -58,16 +68,43 @@ INTROSPECTION = """
 """
 
 
-def log(msg):
+def log(msg: str) -> None:
     print(msg, flush=True)
 
 
-def xautolock(arg):
+def run(*cmd: str) -> None:
     try:
-        subprocess.run(["xautolock", arg], check=True, timeout=10)
-        log(f"xautolock {arg}")
+        subprocess.run(cmd, check=True, timeout=10)
+        log(" ".join(cmd))
     except (OSError, subprocess.SubprocessError) as e:
-        log(f"xautolock {arg} failed: {e}")
+        log(f"{' '.join(cmd)} failed: {e}")
+
+
+def xautolock(arg: str) -> None:
+    run("xautolock", arg)
+
+
+def read_backend() -> str:
+    """Return the backend set in the config file, else the installed locker (xautolock if both are)."""
+    config_home = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config")
+    path = os.path.join(config_home, CONFIG_FILE)
+    try:
+        with open(path) as f:
+            for line in f:
+                key, sep, value = line.partition("=")
+                if not sep or key.strip() != "backend":
+                    continue
+                value = value.strip().strip("\"'")
+                if value in BACKENDS:
+                    return value
+                log(f"{path}: unknown backend '{value}' (expected {' or '.join(BACKENDS)}); detecting")
+    except FileNotFoundError:
+        pass
+    except OSError as e:
+        log(f"Cannot read {path}: {e}; detecting the backend")
+    if not shutil.which("xautolock") and shutil.which("xss-lock"):
+        return "xss-lock"
+    return "xautolock"
 
 
 class XBlanking:
@@ -77,8 +114,8 @@ class XBlanking:
     pause by itself when this client releases it or its connection closes (exit, crash).
     """
 
-    def __init__(self):
-        self.dpy = None
+    def __init__(self) -> None:
+        self.dpy: int | None = None
         try:
             self.x11 = ctypes.CDLL("libX11.so.6")
             self.xss = ctypes.CDLL("libXss.so.1")
@@ -102,7 +139,7 @@ class XBlanking:
             return
         self.dpy = dpy
 
-    def suspend(self, on):
+    def suspend(self, on: bool) -> None:
         if self.dpy:
             self.xss.XScreenSaverSuspend(self.dpy, int(on))
             self.x11.XFlush(self.dpy)
@@ -118,25 +155,33 @@ class ActivityBurst:
 
 
 class Inhibitor:
-    def __init__(self, ignored_reasons, release_delay, blanking):
+    def __init__(
+        self, backend: str, ignored_reasons: set[str], release_delay: int, blanking: XBlanking,
+        state_path: str | None, restore_window: int,
+    ) -> None:
+        self.xautolock = backend == "xautolock"  # else xss-lock: pausing X blanking is enough
         self.ignored_reasons = ignored_reasons
         self.blanking = blanking
         self.release_delay = release_delay
-        self.cookies = {}  # cookie -> (sender, application_name, reason, counts)
-        self.next_cookie = itertools.count(1)
-        self.proc_names = {}  # sender -> "comm[pid]", for logging only
+        self.state_path = state_path
+        self.restore_window = restore_window
+        self.state_ready = False  # set once this instance owns the name and may write the state file
+        self.heartbeat_timer = 0
+        self.cookies: dict[int, Cookie] = {}
+        self.last_cookie = 0
+        self.proc_names: dict[str, str] = {}  # sender -> "comm[pid]", for logging only
         self.activity: dict[str, ActivityBurst] = {}  # sender -> SimulateUserActivity burst
         self.disabled = False
         self.release_timer = 0
-        self.conn = None
+        self.conn: Gio.DBusConnection | None = None
 
     # --- lock state -------------------------------------------------------
 
-    def active_count(self):
+    def active_count(self) -> int:
         return sum(1 for *_, counts in self.cookies.values() if counts)
 
-    def update(self):
-        """Run xautolock only on 0->1 and 1->0 transitions; delay re-enabling."""
+    def update(self) -> None:
+        """Block or unblock only on 0->1 and 1->0 transitions; delay unblocking."""
         active = self.active_count()
         if active and self.release_timer:
             GLib.source_remove(self.release_timer)
@@ -147,19 +192,21 @@ class Inhibitor:
             # Browsers send quick Inhibit/UnInhibit bursts when a video starts.
             self.release_timer = GLib.timeout_add_seconds(self.release_delay, self.on_release_timeout)
 
-    def on_release_timeout(self):
+    def on_release_timeout(self) -> bool:
         self.release_timer = 0
         if not self.active_count() and self.disabled:
             self.unblock()
         return GLib.SOURCE_REMOVE
 
-    def block(self):
-        xautolock("-disable")
+    def block(self) -> None:
+        if self.xautolock:
+            xautolock("-disable")
         self.blanking.suspend(True)
         self.disabled = True
 
-    def unblock(self):
-        xautolock("-enable")
+    def unblock(self) -> None:
+        if self.xautolock:
+            xautolock("-enable")
         self.blanking.suspend(False)
         self.disabled = False
 
@@ -169,35 +216,140 @@ class Inhibitor:
         if self.release_timer:
             GLib.source_remove(self.release_timer)
             self.release_timer = 0
+        self.save(stopped=True)
         if self.disabled:
             self.unblock()
 
+    # --- state file -------------------------------------------------------
+
+    def start(self) -> None:
+        """Take over inhibits saved by a previous instance, then set the locker to match."""
+        self.restore()
+        self.state_ready = True
+        self.save()
+        if self.active_count():
+            self.block()
+        elif self.xautolock:
+            # Start from a known state, in case a previous instance died while disabled.
+            xautolock("-enable")
+
+    def restore(self) -> None:
+        if not self.state_path:
+            log("XDG_RUNTIME_DIR is not set; inhibits will not survive a restart")
+            return
+        assert self.conn is not None
+        path = self.state_path
+        try:
+            with open(path) as f:
+                state = json.load(f)
+            age = time.time() - (state["stopped_at"] or os.stat(path).st_mtime)
+            bus_id = state["bus_id"]
+            last_cookie = int(state["last_cookie"])
+            saved = [(int(c), s, a, r) for c, s, a, r in state["cookies"]]
+        except FileNotFoundError:
+            return
+        except (OSError, ValueError, KeyError, TypeError) as e:
+            log(f"Ignoring unreadable state file {path}: {e}")
+            return
+        if bus_id != self.conn.get_guid():
+            # Unique names like :1.42 are reused by a new bus, so they would mean other clients.
+            if saved:
+                log(f"Not restoring {len(saved)} inhibit(s): saved on another session bus")
+            return
+        # Keep numbering after old cookies, so a late UnInhibit can't hit a new one.
+        self.last_cookie = last_cookie
+        if not saved:
+            return
+        if age > self.restore_window:
+            # An UnInhibit sent while no instance ran is lost; don't trust an old file.
+            log(f"Not restoring {len(saved)} inhibit(s): saved {age:.0f}s ago, "
+                f"over the {self.restore_window}s window")
+            return
+        for cookie, sender, app, reason in saved:
+            if not self.has_owner(sender):
+                log(f"Not restoring {cookie}: '{app}' '{reason}' ({sender} left the bus)")
+                continue
+            counts = reason not in self.ignored_reasons
+            self.cookies[cookie] = (sender, app, reason, counts)
+            note = "" if counts else " (reason ignored, does not block locking)"
+            log(f"Restored {cookie}: '{app}' '{reason}' from {self.proc_name(sender)}{note}")
+
+    def save(self, stopped: bool = False) -> None:
+        """Write the held inhibits atomically; while any are held, touch the file regularly.
+
+        The touches keep the mtime recent, so after a crash the next instance can tell how
+        long ago it happened. A clean stop records the time in `stopped_at` instead.
+        """
+        if not self.state_ready or not self.state_path:
+            return
+        assert self.conn is not None
+        state = {
+            "bus_id": self.conn.get_guid(),
+            "last_cookie": self.last_cookie,
+            "stopped_at": time.time() if stopped else None,
+            "cookies": [[c, s, a, r] for c, (s, a, r, _) in self.cookies.items()],
+        }
+        tmp = f"{self.state_path}.tmp"
+        try:
+            with open(tmp, "w") as f:
+                json.dump(state, f)
+            os.replace(tmp, self.state_path)
+        except OSError as e:
+            log(f"Cannot write state file {self.state_path}: {e}")
+        if self.cookies and not stopped:
+            if not self.heartbeat_timer:
+                self.heartbeat_timer = GLib.timeout_add_seconds(STATE_HEARTBEAT, self.on_heartbeat)
+        elif self.heartbeat_timer:
+            GLib.source_remove(self.heartbeat_timer)
+            self.heartbeat_timer = 0
+
+    def on_heartbeat(self) -> bool:
+        assert self.state_path is not None
+        try:
+            os.utime(self.state_path)
+        except OSError as e:
+            log(f"Cannot touch state file {self.state_path}: {e}")
+        return GLib.SOURCE_CONTINUE
+
     # --- D-Bus ------------------------------------------------------------
 
-    def proc_name(self, sender):
+    def bus_call(self, method: str, name: str, reply_type: str) -> Any:
+        """Call an org.freedesktop.DBus method that takes a bus name; return its single result."""
+        assert self.conn is not None  # set in on_bus_acquired, before any method call
+        return self.conn.call_sync(
+            "org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus",
+            method, GLib.Variant("(s)", (name,)),
+            GLib.VariantType(reply_type), Gio.DBusCallFlags.NONE, 1000, None,
+        ).unpack()[0]
+
+    def has_owner(self, name: str) -> bool:
+        try:
+            return bool(self.bus_call("NameHasOwner", name, "(b)"))
+        except GLib.Error:
+            return False
+
+    def proc_name(self, sender: str) -> str:
         if sender not in self.proc_names:
             try:
-                pid = self.conn.call_sync(
-                    "org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus",
-                    "GetConnectionUnixProcessID", GLib.Variant("(s)", (sender,)),
-                    GLib.VariantType("(u)"), Gio.DBusCallFlags.NONE, 1000, None,
-                ).unpack()[0]
+                pid = self.bus_call("GetConnectionUnixProcessID", sender, "(u)")
                 with open(f"/proc/{pid}/comm") as f:
                     self.proc_names[sender] = f"{f.read().strip()}[{pid}]"
             except (GLib.Error, OSError):
                 self.proc_names[sender] = "?"
         return f"{sender} {self.proc_names[sender]}"
 
-    def inhibit(self, sender, app, reason):
-        cookie = next(self.next_cookie)
+    def inhibit(self, sender: str, app: str, reason: str) -> int:
+        self.last_cookie += 1
+        cookie = self.last_cookie
         counts = reason not in self.ignored_reasons
         self.cookies[cookie] = (sender, app, reason, counts)
         note = "" if counts else " (reason ignored, does not block locking)"
         log(f"Inhibit {cookie}: '{app}' '{reason}' from {self.proc_name(sender)}{note}")
+        self.save()
         self.update()
         return cookie
 
-    def uninhibit(self, sender, cookie):
+    def uninhibit(self, sender: str, cookie: int) -> None:
         entry = self.cookies.get(cookie)
         if entry is None or entry[0] != sender:
             log(f"UnInhibit {cookie} from {self.proc_name(sender)}: unknown cookie, ignored")
@@ -205,6 +357,7 @@ class Inhibitor:
         del self.cookies[cookie]
         _, app, reason, _ = entry
         log(f"UnInhibit {cookie}: '{app}' '{reason}'")
+        self.save()
         self.update()
 
     def simulate_activity(self, sender: str) -> None:
@@ -244,9 +397,13 @@ class Inhibitor:
         self.end_activity(sender)
         self.proc_names.pop(sender, None)
         if gone:
+            self.save()
             self.update()
 
-    def on_method_call(self, conn, sender, path, iface, method, params, invocation):
+    def on_method_call(
+        self, conn: Gio.DBusConnection, sender: str, path: str, iface: str, method: str,
+        params: GLib.Variant, invocation: Gio.DBusMethodInvocation,
+    ) -> None:
         if method == "Inhibit":
             app, reason = params.unpack()
             invocation.return_value(GLib.Variant("(u)", (self.inhibit(sender, app, reason),)))
@@ -258,19 +415,25 @@ class Inhibitor:
             invocation.return_value(None)
         elif method == "Lock":
             log(f"Lock from {self.proc_name(sender)}")
-            xautolock("-locknow")
+            if self.xautolock:
+                xautolock("-locknow")
+            else:
+                run("loginctl", "lock-session")
             invocation.return_value(None)
         elif method in ("GetActive", "SetActive"):
             invocation.return_value(GLib.Variant("(b)", (False,)))
         elif method in ("GetActiveTime", "GetSessionIdleTime"):
             invocation.return_value(GLib.Variant("(u)", (0,)))
 
-    def on_name_owner_changed(self, conn, sender, path, iface, signal_name, params):
+    def on_name_owner_changed(
+        self, conn: Gio.DBusConnection, sender: str | None, path: str, iface: str, signal_name: str,
+        params: GLib.Variant,
+    ) -> None:
         name, _old, new = params.unpack()
         if not new and name.startswith(":"):
             self.release_sender(name)
 
-    def on_bus_acquired(self, conn, name):
+    def on_bus_acquired(self, conn: Gio.DBusConnection, name: str) -> None:
         self.conn = conn
         iface = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION).interfaces[0]
         for path in OBJECT_PATHS:
@@ -281,7 +444,7 @@ class Inhibitor:
         )
 
 
-def main():
+def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
         "-i", "--ignore-reason", action="append", metavar="REASON",
@@ -293,28 +456,51 @@ def main():
     )
     parser.add_argument(
         "-d", "--release-delay", type=int, default=5, metavar="SECONDS",
-        help="wait this long after the last release before re-enabling xautolock (default: 5)",
+        help="wait this long after the last release before unblocking locking (default: 5)",
+    )
+    parser.add_argument(
+        "--restore-window", type=int, default=30, metavar="SECONDS",
+        help="take over inhibits from a previous instance that stopped at most this long ago "
+             "(default: 30; 0 disables)",
+    )
+    parser.add_argument(
+        "--cleanup", action="store_true",
+        help="re-enable xautolock if it is the backend, then exit (for the unit's ExecStopPost)",
     )
     args = parser.parse_args()
 
+    backend = read_backend()
+    if args.cleanup:
+        if backend == "xautolock":
+            xautolock("-enable")
+        return
+
+    ignored: set[str]
     if args.count_all:
         ignored = set()
     else:
         ignored = set(args.ignore_reason or ["Playing audio"])
-    inhibitor = Inhibitor(ignored, max(args.release_delay, 0), XBlanking())
+    runtime_dir = os.environ.get("XDG_RUNTIME_DIR")
+    state_path = os.path.join(runtime_dir, STATE_FILE) if runtime_dir else None
+    blanking = XBlanking()
+    if backend == "xss-lock" and not blanking.dpy:
+        log("Warning: with xss-lock, inhibits block locking only through X blanking, which is unavailable")
+    inhibitor = Inhibitor(
+        backend, ignored, max(args.release_delay, 0), blanking, state_path, max(args.restore_window, 0),
+    )
     loop = GLib.MainLoop()
     exit_code = 0
     acquired = False
 
-    def on_name_acquired(conn, name):
+    def on_name_acquired(conn: Gio.DBusConnection, name: str) -> None:
         nonlocal acquired
         acquired = True
-        log(f"Owning {name}; ignored reasons: {sorted(ignored) or 'none'}")
-        # Start from a known state, in case a previous instance died while disabled.
-        # Only after owning the name, so a second instance can't undo an active inhibit.
-        xautolock("-enable")
+        log(f"Owning {name}; backend: {backend}; ignored reasons: {sorted(ignored) or 'none'}")
+        # Only after owning the name, so a second instance can't undo an active inhibit
+        # or overwrite the state file.
+        inhibitor.start()
 
-    def on_name_lost(conn, name):
+    def on_name_lost(conn: Gio.DBusConnection | None, name: str) -> None:
         nonlocal exit_code
         if conn is None:
             log("Could not connect to the session bus")
@@ -327,7 +513,7 @@ def main():
             exit_code = EXIT_NAME_TAKEN
         loop.quit()
 
-    def on_signal():
+    def on_signal() -> bool:
         log("Signal received, exiting")
         loop.quit()
         return GLib.SOURCE_REMOVE

xidle-inhibit.service

@@ -0,0 +1,10 @@
+[Unit]
+Description=Block screen locking (xautolock or xss-lock) on org.freedesktop.ScreenSaver inhibit requests
+
+[Service]
+Type=dbus
+BusName=org.freedesktop.ScreenSaver
+ExecStart=/usr/bin/xidle-inhibit
+ExecStopPost=-/usr/bin/xidle-inhibit --cleanup
+Restart=on-failure
+RestartPreventExitStatus=2