leisure kill: sweep only after 30 s of continuous untrusted now.json

One stale moment (resume from sleep, a watcher restart) while spent no
longer kills every browser at once. A trusted or paused tick, a clock
step back, or a gap between ticks over 10 s (a suspend) restarts the
window; untrust is tracked while budget remains too.

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

steam_backlog_enforcer/_leisure_kill.py

@@ -8,9 +8,11 @@ Steam cutoff (:mod:`_playtime_cutoff`). Which pids:
   every ``failclosed`` reason's pid -- a browser that is not reporting (another
   browser, an unreported private window) has nothing redirecting it.
   ``game`` reasons belong to the Steam path.
-* **Untrusted** (stale, missing, badly signed, unreadable): every browser,
-  media player and wsg-grabber reviewer (:func:`_leisure_procs.sweep_targets`)
-  until the watcher is fresh again.
+* **Untrusted** (stale, missing, badly signed, unreadable) for 30 s in a row:
+  every browser, media player and wsg-grabber reviewer
+  (:func:`_leisure_procs.sweep_targets`) until the watcher is fresh again. One
+  stale moment (resume from sleep, a watcher restart) kills nothing: a trusted
+  or paused tick, or a gap between ticks (a suspend), restarts the window.
 * **Paused** (a locker grab is held): nothing; the lock is not a stale watcher.
 
 Each process gets SIGTERM once, then SIGKILL every tick from 30 s after *its*
@@ -25,6 +27,7 @@ first SIGKILL go to ``budget.jsonl``.
 from __future__ import annotations
 
 import logging
+import math
 import os
 import signal
 from typing import TYPE_CHECKING, Final, Protocol
@@ -50,6 +53,9 @@ logger = logging.getLogger(__name__)
 EVENT_LEISURE_KILL: Final = "leisure_kill"
 EVENT_KILL_REFUSED: Final = "leisure_kill_refused"
 SIGKILL_AFTER_SECONDS: Final = 30.0
+UNTRUSTED_SWEEP_AFTER: Final = 30.0
+# Ticks are 3 s apart; a longer gap (suspend, a stalled loop) is not "in a row".
+_MAX_TICK_GAP: Final = 10.0
 _KILLED_SOURCES: Final = frozenset({"focus", "media"})
 _FAILCLOSED: Final = "failclosed"
 
@@ -90,6 +96,8 @@ class LeisureKiller:
         self._termed: dict[_Ident, float] = {}
         self._killed: set[_Ident] = set()
         self._refused: set[_Ident] = set()
+        self._untrusted_since: float | None = None
+        self._last_stamp = -math.inf
         self._last_error = ""
 
     def enforce(
@@ -128,6 +136,7 @@ class LeisureKiller:
     ) -> None:
         if not rules.leisure or rules.demo or not rules.enforcement:
             return
+        untrusted_for = self._untrusted_for(verdict, stamp)
         if state.seconds < rules.budget_seconds:
             self._termed.clear()
             self._killed.clear()
@@ -135,6 +144,8 @@ class LeisureKiller:
         if verdict.paused_by is not None:
             return
         if verdict.fact is None:
+            if untrusted_for < UNTRUSTED_SWEEP_AFTER:
+                return
             targets, why = sweep_targets(), "now.json untrusted"
         else:
             targets, why = fact_targets(verdict.fact), "now.json"
@@ -157,6 +168,17 @@ class LeisureKiller:
                 if self._signal(ident[0], signal.SIGKILL, None if quiet else note):
                     self._killed.add(ident)
 
+    def _untrusted_for(self, verdict: ShadowVerdict, stamp: float) -> float:
+        """Seconds ``now.json`` has been untrusted in a row, as of this tick."""
+        gap = stamp - self._last_stamp
+        self._last_stamp = stamp
+        if verdict.fact is not None or verdict.paused_by is not None:
+            self._untrusted_since = None
+            return 0.0
+        if self._untrusted_since is None or not 0.0 <= gap <= _MAX_TICK_GAP:
+            self._untrusted_since = stamp
+        return stamp - self._untrusted_since
+
     def _identify(self, targets: dict[int, str]) -> dict[_Ident, str]:
         chain = own_chain()
         live: dict[_Ident, str] = {}

steam_backlog_enforcer/tests/test_leisure_kill.py

@@ -146,9 +146,9 @@ class TestEnforce:
         self, proc: Path, kill: MagicMock
     ) -> None:
         add_proc(proc, FakeProc(41, exe="/usr/bin/mpv", uid=0))
-        LeisureKiller(StubLog()).enforce(
-            _SPENT, _RULES, verdict=_verdict(None), stamp=1.0
-        )
+        killer = LeisureKiller(StubLog())
+        for stamp in range(0, 31, 3):  # untrusted for the whole 30 s window
+            killer.enforce(_SPENT, _RULES, verdict=_verdict(None), stamp=stamp)
         assert sorted(c.args[0] for c in kill.call_args_list) == [10, 20, 30]
 
     def test_paused_is_not_stale(self, proc: Path, kill: MagicMock) -> None:
@@ -229,8 +229,8 @@ class TestEnforce:
     ) -> None:
         killer = LeisureKiller(None)
         with patch.object(_leisure_kill, "sweep_targets", side_effect=RecursionError):
-            for _ in range(2):
-                killer.enforce(_SPENT, _RULES, verdict=_verdict(None), stamp=0.0)
+            for stamp in range(0, 34, 3):  # sweeps (and fails) at 30 and 33
+                killer.enforce(_SPENT, _RULES, verdict=_verdict(None), stamp=stamp)
         assert sum("Leisure kill failed" in r.message for r in caplog.records) == 1
 
     def test_a_failed_sigkill_is_retried_and_logged_once_delivered(

steam_backlog_enforcer/tests/test_leisure_kill_debounce.py

@@ -0,0 +1,110 @@
+"""An untrusted watcher is swept only after 30 s of continuous untrust."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from leisure_watch.facts import Focus, Now
+import pytest
+
+from steam_backlog_enforcer._leisure_kill import UNTRUSTED_SWEEP_AFTER, LeisureKiller
+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 collections.abc import Iterable, Iterator
+    from pathlib import Path
+    from unittest.mock import MagicMock
+
+_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)
+_FREE = PlaytimeState(day_key="2026-11-02", seconds=1.0)
+_TRUSTED = ShadowVerdict(
+    paused_by=None,
+    keys=(),
+    fact=Now(0.0, 1, "free", None, (), Focus("i3", "", None, idle=False)),
+)
+_STALE = ShadowVerdict(paused_by=None, keys=("k",), fact=None)
+_PAUSED = ShadowVerdict(paused_by="i3lock", keys=(), fact=None)
+
+
+@pytest.fixture
+def kill(tmp_path: Path) -> Iterator[MagicMock]:
+    add_proc(leisure_proc(tmp_path), FakeProc(10, exe="/usr/bin/mpv"))
+    with patch.object(os, "kill") as mock:
+        yield mock
+
+
+def _ticks(
+    killer: LeisureKiller,
+    stamps: Iterable[float],
+    verdict: ShadowVerdict = _STALE,
+    state: PlaytimeState = _SPENT,
+) -> None:
+    for stamp in stamps:
+        killer.enforce(state, _RULES, verdict=verdict, stamp=float(stamp))
+
+
+def test_the_window_is_30_s() -> None:
+    assert UNTRUSTED_SWEEP_AFTER == 30.0
+
+
+def test_one_stale_moment_kills_nothing(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(0, 28, 3))
+    kill.assert_not_called()
+    _ticks(killer, [30])
+    assert [c.args[0] for c in kill.call_args_list] == [10]
+
+
+def test_a_trusted_tick_restarts_the_window(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(0, 25, 3))
+    _ticks(killer, [27], _TRUSTED)
+    _ticks(killer, range(30, 58, 3))
+    kill.assert_not_called()
+    _ticks(killer, [60])
+    kill.assert_called_once()
+
+
+def test_a_pause_restarts_the_window(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(0, 25, 3))
+    _ticks(killer, [27], _PAUSED)
+    _ticks(killer, range(30, 55, 3))
+    kill.assert_not_called()
+
+
+def test_resume_from_sleep_restarts_the_window(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(0, 28, 3))
+    _ticks(killer, range(3600, 3628, 3))
+    kill.assert_not_called()
+    _ticks(killer, [3630])
+    kill.assert_called_once()
+
+
+def test_untrust_counts_while_the_budget_lasts(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(0, 31, 3), state=_FREE)
+    _ticks(killer, [33])
+    kill.assert_called_once()
+
+
+def test_a_clock_step_back_restarts_the_window(kill: MagicMock) -> None:
+    killer = LeisureKiller(None)
+    _ticks(killer, range(100, 128, 3))
+    _ticks(killer, range(0, 28, 3))
+    kill.assert_not_called()