leisure kill: never signal the desktop session, whatever now.json says

Defence in depth for a forged or buggy now.json while root: i3 and its
bar, X, systemd --user, dbus, audio, ssh/gpg agents, tmux/screen, every
terminal and shell, sudo, Claude (incl. its versioned native binary and
node), keepassxc, notification daemons, and Python running a locker,
leisure_watch or daily_limits are refused by exe/script. Each refusal is
logged once (journal + budget.jsonl leisure_kill_refused).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N

steam_backlog_enforcer/_leisure_kill.py

@@ -17,7 +17,9 @@ Each process gets SIGTERM once, then SIGKILL every tick from 30 s after *its*
 SIGTERM -- timed per process, so an app opened an hour into the block still
 gets its 30 s. A process is its (pid, start time), so a reused pid starts
 over. Root/system processes, lockers and the enforcer itself are never
-signalled. Every SIGTERM and the first SIGKILL go to ``budget.jsonl``.
+signalled, nor -- whatever ``now.json`` says -- the desktop session
+(:mod:`_leisure_never_kill`; each refusal logged once). Every SIGTERM and the
+first SIGKILL go to ``budget.jsonl``.
 """
 
 from __future__ import annotations
@@ -27,6 +29,7 @@ import os
 import signal
 from typing import TYPE_CHECKING, Final, Protocol
 
+from steam_backlog_enforcer._leisure_never_kill import never_kill
 from steam_backlog_enforcer._leisure_procs import (
     exe_name,
     is_browser,
@@ -45,6 +48,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 EVENT_LEISURE_KILL: Final = "leisure_kill"
+EVENT_KILL_REFUSED: Final = "leisure_kill_refused"
 SIGKILL_AFTER_SECONDS: Final = 30.0
 _KILLED_SOURCES: Final = frozenset({"focus", "media"})
 _FAILCLOSED: Final = "failclosed"
@@ -85,6 +89,7 @@ class LeisureKiller:
         self.log = log
         self._termed: dict[_Ident, float] = {}
         self._killed: set[_Ident] = set()
+        self._refused: set[_Ident] = set()
         self._last_error = ""
 
     def enforce(
@@ -152,16 +157,32 @@ class LeisureKiller:
                 if self._signal(ident[0], signal.SIGKILL, None if quiet else note):
                     self._killed.add(ident)
 
-    @staticmethod
-    def _identify(targets: dict[int, str]) -> dict[_Ident, str]:
+    def _identify(self, targets: dict[int, str]) -> dict[_Ident, str]:
         chain = own_chain()
         live: dict[_Ident, str] = {}
+        refused: set[_Ident] = set()
         for pid, key in targets.items():
             started = start_time(pid)
-            if started is not None and not is_protected(pid, chain):
+            if started is None or is_protected(pid, chain):
+                continue
+            guard = never_kill(pid)
+            if guard is None:
                 live[pid, started] = key
+                continue
+            refused.add((pid, started))
+            if (pid, started) not in self._refused:
+                self._refuse(pid, key, guard)
+        self._refused = refused
         return live
 
+    def _refuse(self, pid: int, key: str, name: str) -> None:
+        """Log, once per process, a kill the never-kill set stopped."""
+        logger.warning(
+            "Refused leisure kill of pid %d (%s): %s is never killed.", pid, key, name
+        )
+        if self.log is not None:
+            self.log.record(EVENT_KILL_REFUSED, pid=pid, key=key, name=name)
+
     def _signal(
         self, pid: int, sig: signal.Signals, note: dict[str, object] | None
     ) -> bool:

steam_backlog_enforcer/_leisure_never_kill.py

@@ -0,0 +1,121 @@
+"""What a leisure kill never signals, whatever ``now.json`` says.
+
+Defence in depth: ``now.json`` is signed but a determined forger (or a watcher
+bug) could name any pid, and the enforcer is root. The desktop session -- the
+window manager and its bar, X, the user's systemd and bus, audio, ssh/gpg
+agents, multiplexers, terminals and shells, sudo, Claude, the password manager,
+the notification daemon, and every locker or accountant -- survives that.
+
+Matched on the binary actually run (the ``exe`` link, never ``comm``, which any
+process may rename) or, for a Python interpreter, the script/module it runs.
+:mod:`_leisure_kill` asks :func:`never_kill` after :func:`is_protected`.
+"""
+
+from __future__ import annotations
+
+from pathlib import PurePath
+from typing import Final
+
+from steam_backlog_enforcer._leisure_procs import (
+    LOCKER_NAMES,
+    argv,
+    exe_name,
+    exe_path,
+    script_of,
+)
+
+_TERMINALS: Final = frozenset(
+    {
+        "alacritty",
+        "foot",
+        "footclient",
+        "ghostty",
+        "gnome-terminal",
+        "gnome-terminal-server",
+        "kitty",
+        "konsole",
+        "lxterminal",
+        "mate-terminal",
+        "qterminal",
+        "rxvt",
+        "sakura",
+        "st",
+        "terminator",
+        "terminology",
+        "tilix",
+        "urxvt",
+        "urxvtd",
+        "uxterm",
+        "wezterm",
+        "wezterm-gui",
+        "xfce4-terminal",
+        "xterm",
+    }
+)
+# Exact exe basenames (lower-case: ``exe_name`` lowers ``Xorg``).
+NEVER_KILL_NAMES: Final = _TERMINALS | frozenset(
+    {
+        "i3",
+        "i3bar",
+        "i3blocks",
+        "xorg",
+        "xwayland",
+        "systemd",
+        "wireplumber",
+        "pulseaudio",
+        "sshd",
+        "ssh-agent",
+        "gpg-agent",
+        "tmux",
+        "screen",
+        "zsh",
+        "bash",
+        "fish",
+        "sh",
+        "dash",
+        "sudo",
+        "keepassxc",
+        "dunst",
+        "xfce4-notifyd",
+        "claude",
+    }
+)
+NEVER_KILL_PREFIXES: Final = ("dbus-", "pipewire", "systemd-")
+# Python programs by script/module stem: Python terminals, and every locker,
+# accountant and the limits bar (LOCKER_NAMES already spares their children).
+NEVER_KILL_SCRIPTS: Final = LOCKER_NAMES | frozenset(
+    {"daily_limits", "guake", "terminator"}
+)
+_CLAUDE: Final = "claude"
+# Claude Code's native build: ~/.local/share/claude/versions/<version>.
+_CLAUDE_DIRS: Final = frozenset({"claude", "claude-code"})
+_JS_RUNTIMES: Final = frozenset({"node", "bun"})
+
+
+def _is_claude(pid: int, name: str) -> bool:
+    if _CLAUDE_DIRS & set(PurePath(exe_path(pid).lower()).parts):
+        return True
+    if name not in _JS_RUNTIMES:
+        return False
+    script = argv(pid)[1:2]
+    return bool(script) and bool(_CLAUDE_DIRS & set(PurePath(script[0]).parts))
+
+
+def never_kill(pid: int) -> str | None:
+    """Why *pid* is never signalled by a leisure kill, ``None`` if it may be.
+
+    Args:
+        pid: Candidate process (already past :func:`is_protected`).
+
+    Returns:
+        The protected name that matched (for the refusal log), or ``None``.
+    """
+    name = exe_name(pid)
+    if name in NEVER_KILL_NAMES or name.startswith(NEVER_KILL_PREFIXES):
+        return name
+    if _is_claude(pid, name):
+        return _CLAUDE
+    script = script_of(pid)
+    if script is not None and script[0] in NEVER_KILL_SCRIPTS:
+        return script[0]
+    return None

steam_backlog_enforcer/tests/test_leisure_never_kill.py

@@ -0,0 +1,177 @@
+"""The never-kill set: the desktop survives a forged or buggy now.json."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from leisure_watch.facts import Focus, Now, Reason
+import pytest
+
+from steam_backlog_enforcer._leisure_kill import LeisureKiller
+from steam_backlog_enforcer._leisure_never_kill import never_kill
+from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
+from steam_backlog_enforcer._shadow import ShadowVerdict
+from steam_backlog_enforcer.tests._isolate_leisure import leisure_proc
+from steam_backlog_enforcer.tests._leisure_fake_proc import FakeProc, add_proc
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+_PY = "/usr/bin/python3.14"
+_RULES = PlaytimeRules(
+    budget_seconds=100.0,
+    warn_at=(),
+    sigkill_after=30.0,
+    count_launchers=False,
+    enforcement=True,
+    demo=False,
+    leisure=True,
+)
+_SPENT = PlaytimeState(day_key="2026-11-02", seconds=100.0)
+
+
+@pytest.fixture
+def proc(tmp_path: Path) -> Path:
+    return leisure_proc(tmp_path)
+
+
+class TestNeverKill:
+    @pytest.mark.parametrize(
+        ("exe", "args", "why"),
+        [
+            ("/usr/bin/i3", ("i3",), "i3"),
+            ("/usr/bin/i3bar", ("i3bar",), "i3bar"),
+            ("/usr/bin/i3blocks", ("i3blocks",), "i3blocks"),
+            ("/usr/lib/Xorg", ("Xorg", ":0"), "xorg"),
+            ("/usr/bin/Xwayland", ("Xwayland",), "xwayland"),
+            ("/usr/lib/systemd/systemd", ("systemd", "--user"), "systemd"),
+            ("/usr/bin/dbus-broker", ("dbus-broker",), "dbus-broker"),
+            ("/usr/bin/dbus-daemon", ("dbus-daemon",), "dbus-daemon"),
+            ("/usr/bin/pipewire-pulse", ("pipewire-pulse",), "pipewire-pulse"),
+            ("/usr/bin/pipewire", ("pipewire",), "pipewire"),
+            ("/usr/bin/wireplumber", ("wireplumber",), "wireplumber"),
+            ("/usr/bin/pulseaudio", ("pulseaudio",), "pulseaudio"),
+            ("/usr/bin/sshd", ("sshd",), "sshd"),
+            ("/usr/bin/ssh-agent", ("ssh-agent",), "ssh-agent"),
+            ("/usr/bin/gpg-agent", ("gpg-agent",), "gpg-agent"),
+            ("/usr/bin/tmux", ("tmux",), "tmux"),
+            ("/usr/bin/screen", ("screen",), "screen"),
+            ("/usr/bin/alacritty", ("alacritty",), "alacritty"),
+            ("/usr/bin/kitty", ("kitty",), "kitty"),
+            ("/usr/bin/xterm", ("xterm",), "xterm"),
+            ("/usr/bin/foot", ("foot",), "foot"),
+            ("/usr/bin/zsh", ("-zsh",), "zsh"),
+            ("/usr/bin/bash", ("bash",), "bash"),
+            ("/usr/bin/fish", ("fish",), "fish"),
+            ("/usr/bin/sh", ("sh", "-c", "x"), "sh"),
+            ("/usr/bin/sudo", ("sudo", "x"), "sudo"),
+            ("/usr/bin/keepassxc", ("keepassxc",), "keepassxc"),
+            ("/usr/bin/dunst", ("dunst",), "dunst"),
+            ("/usr/lib/xfce4/notifyd/xfce4-notifyd", ("x",), "xfce4-notifyd"),
+            ("/usr/bin/vim (deleted)", ("vim",), None),
+            ("/usr/bin/mpv", ("mpv",), None),
+            ("/usr/bin/shellcheck", ("shellcheck",), None),
+            ("/usr/bin/screenkey", ("screenkey",), None),
+            ("/opt/helium-browser-bin/helium", ("helium",), None),
+        ],
+    )
+    def test_by_binary(
+        self, proc: Path, exe: str, args: tuple[str, ...], why: str | None
+    ) -> None:
+        add_proc(proc, FakeProc(10, exe=exe, argv=args))
+        assert never_kill(10) == why
+
+    @pytest.mark.parametrize(
+        ("args", "why"),
+        [
+            (("/usr/bin/python", "/usr/bin/terminator"), "terminator"),
+            (("python3", "-m", "leisure_watch", "run"), "leisure_watch"),
+            (
+                ("python3", "/home/kuhy/.local/bin/daily-limits", "--sync"),
+                "daily_limits",
+            ),
+            (("python3", "-m", "screen_locker._web"), "screen_locker"),
+            (("python3", "/x/leetcode_guard_tray.py"), "leetcode_guard_tray"),
+            (("python3", "-m", "wsg_grabber"), None),
+            (("python3", "-c", "import leisure_watch"), None),
+        ],
+    )
+    def test_by_python_script(
+        self, proc: Path, args: tuple[str, ...], why: str | None
+    ) -> None:
+        add_proc(proc, FakeProc(10, exe=_PY, argv=args))
+        assert never_kill(10) == why
+
+    @pytest.mark.parametrize(
+        ("exe", "args", "why"),
+        [
+            ("/home/kuhy/.local/share/claude/versions/2.1.302", ("x",), "claude"),
+            ("/x/claude_agent_sdk/_bundled/claude", ("claude",), "claude"),
+            (
+                "/usr/bin/node",
+                ("node", "/x/@anthropic-ai/claude-code/cli.js"),
+                "claude",
+            ),
+            ("/usr/bin/node", ("node", "/x/server.js"), None),
+            ("/usr/bin/node", ("node",), None),
+        ],
+    )
+    def test_claude(
+        self, proc: Path, exe: str, args: tuple[str, ...], why: str | None
+    ) -> None:
+        add_proc(proc, FakeProc(10, exe=exe, argv=args))
+        assert never_kill(10) == why
+
+
+class TestKillerRefuses:
+    def test_a_forged_fact_cannot_kill_the_desktop(
+        self, proc: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        add_proc(proc, FakeProc(10, exe="/usr/bin/i3", argv=("i3",)))
+        add_proc(proc, FakeProc(11, exe="/usr/bin/alacritty", argv=("alacritty",)))
+        add_proc(proc, FakeProc(12, exe="/usr/bin/mpv"))
+        fact = Now(
+            ts=0.0,
+            seq=1,
+            state="leisure",
+            paused_by=None,
+            reasons=(
+                Reason("failclosed", "failclosed:x", 10, "x"),
+                Reason("focus", "app:alacritty", 11, "x"),
+                Reason("media", "app:mpv", 12, "mpv"),
+            ),
+            focus=Focus("i3", "", None, idle=False),
+        )
+        verdict = ShadowVerdict(paused_by=None, keys=("k",), fact=fact)
+        records: list[tuple[str, dict[str, object]]] = []
+
+        class Log:
+            def record(self, event: str, **fields: object) -> None:
+                records.append((event, fields))
+
+        killer = LeisureKiller(Log())
+        with patch.object(os, "kill") as kill, caplog.at_level(logging.WARNING):
+            for stamp in (0.0, 3.0, 40.0):
+                killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=stamp)
+        assert {c.args[0] for c in kill.call_args_list} == {12}
+        refused = [f for e, f in records if e == "leisure_kill_refused"]
+        assert sorted((f["pid"], f["name"]) for f in refused) == [
+            (10, "i3"),
+            (11, "alacritty"),
+        ]
+        assert sum("never killed" in r.getMessage() for r in caplog.records) == 2
+
+    def test_a_refusal_without_a_log_only_warns(
+        self, proc: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        add_proc(proc, FakeProc(10, exe="/usr/bin/i3", argv=("i3",)))
+        reasons = (Reason("failclosed", "failclosed:x", 10, "x"),)
+        fact = Now(0.0, 1, "leisure", None, reasons, Focus("i3", "", None, idle=False))
+        verdict = ShadowVerdict(paused_by=None, keys=("k",), fact=fact)
+        with patch.object(os, "kill") as kill, caplog.at_level(logging.WARNING):
+            LeisureKiller(None).enforce(_SPENT, _RULES, verdict=verdict, stamp=0.0)
+        kill.assert_not_called()
+        assert "never killed" in caplog.text