From 26a2aabbc7b4a0ce604ffa034eaaf3c8764774e9 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 19 Oct 2026 16:40:49 +0200 Subject: [PATCH] leisure Phase 2 (dormant): bill, warn and kill on the leisure budget from go-live Dormant until earned_time.leisure_active(gaming day) -- the switch file written by `leisure-watch go-live`. From that day: - enforced `seconds` are billed by the shadow rule (lock pause, fail-closed, once per wall clock); budget from plain resolve(); warnings and the cutoff say "Leisure". - at 0, beside the unchanged Steam path, _leisure_kill SIGTERMs (SIGKILL 30 s later, per process) every focus/media pid in now.json except browsers, every failclosed browser pid, and -- with now.json untrusted -- every browser, media player and wsg-grabber reviewer. Root/system, lockers and the enforcer's own chain are never signalled. Kills go to budget.jsonl. - tests: an autouse fixture redirects both switch readers, so no test reads /etc/leisure-watch/leisure_from; existing tests unchanged. Re-pins earned-time to v0.15.0 (leisure_active). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N --- .pre-commit-config.yaml | 4 +- pyproject.toml | 2 +- requirements.txt | 2 +- steam_backlog_enforcer/_leisure_kill.py | 175 +++++++++++++ steam_backlog_enforcer/_leisure_mode.py | 114 ++++++++ steam_backlog_enforcer/_leisure_procs.py | 205 +++++++++++++++ steam_backlog_enforcer/_playtime.py | 26 +- steam_backlog_enforcer/_playtime_budget.py | 35 ++- steam_backlog_enforcer/_playtime_cutoff.py | 28 +- steam_backlog_enforcer/_playtime_session.py | 21 +- steam_backlog_enforcer/_playtime_state.py | 3 + steam_backlog_enforcer/_shadow.py | 9 +- .../tests/_isolate_leisure.py | 52 ++++ .../tests/_leisure_fake_proc.py | 48 ++++ steam_backlog_enforcer/tests/conftest.py | 2 + .../tests/test_leisure_kill.py | 247 ++++++++++++++++++ .../tests/test_leisure_mode.py | 160 ++++++++++++ .../tests/test_leisure_procs.py | 174 ++++++++++++ .../tests/test_leisure_tick.py | 183 +++++++++++++ 19 files changed, 1460 insertions(+), 30 deletions(-) create mode 100644 steam_backlog_enforcer/_leisure_kill.py create mode 100644 steam_backlog_enforcer/_leisure_mode.py create mode 100644 steam_backlog_enforcer/_leisure_procs.py create mode 100644 steam_backlog_enforcer/tests/_isolate_leisure.py create mode 100644 steam_backlog_enforcer/tests/_leisure_fake_proc.py create mode 100644 steam_backlog_enforcer/tests/test_leisure_kill.py create mode 100644 steam_backlog_enforcer/tests/test_leisure_mode.py create mode 100644 steam_backlog_enforcer/tests/test_leisure_procs.py create mode 100644 steam_backlog_enforcer/tests/test_leisure_tick.py diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6c8799a..efc1666 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -110,7 +110,7 @@ repos: - --disable-error-code=list-item additional_dependencies: - types-requests - - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time + - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.15.0#subdirectory=earned_time - leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.2 # =========================================================================== @@ -131,7 +131,7 @@ repos: # (steam_backlog_enforcer._mcp_server). additional_dependencies: - aiohttp==3.14.5 - - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time + - earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.15.0#subdirectory=earned_time - leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.2 - howlongtobeatpy==1.0.24 - mcp==2.3.0 diff --git a/pyproject.toml b/pyproject.toml index 4ee6099..01f9253 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -16,7 +16,7 @@ dependencies = [ "websockets==17.2", # The earner registry (gaming budget terms); stdlib-only, so the root # daemon imports it from the desktop user's site-packages. - "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time", + "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.15.0#subdirectory=earned_time", # The shared /proc game scan and the signed leisure facts (shadow budget, # contract C4); stdlib-only, imported by root from the user site like # earned-time. diff --git a/requirements.txt b/requirements.txt index a81fa01..9b6c833 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,7 +5,7 @@ aiohttp==3.14.5 bandit==1.9.4 codespell==2.4.3 coverage==7.16.2 -earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time +earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.15.0#subdirectory=earned_time howlongtobeatpy==1.0.24 leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.2 mcp==2.3.0 diff --git a/steam_backlog_enforcer/_leisure_kill.py b/steam_backlog_enforcer/_leisure_kill.py new file mode 100644 index 0000000..50efc37 --- /dev/null +++ b/steam_backlog_enforcer/_leisure_kill.py @@ -0,0 +1,175 @@ +"""Phase 2 at 0: end every leisure app the watcher names (Steam is separate). + +Runs every 3 s tick while the leisure budget is spent, beside the unchanged +Steam cutoff (:mod:`_playtime_cutoff`). Which pids: + +* **Trusted** ``now.json``: every ``focus``/``media`` reason's pid unless it is + a browser (the extension redirects those to the "leisure spent" page), and + 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. +* **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* +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``. +""" + +from __future__ import annotations + +import logging +import os +import signal +from typing import TYPE_CHECKING, Final, Protocol + +from steam_backlog_enforcer._leisure_procs import ( + exe_name, + is_browser, + is_protected, + own_chain, + start_time, + sweep_targets, +) + +if TYPE_CHECKING: + from leisure_watch.facts import Now + + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + from steam_backlog_enforcer._shadow import ShadowVerdict + +logger = logging.getLogger(__name__) + +EVENT_LEISURE_KILL: Final = "leisure_kill" +SIGKILL_AFTER_SECONDS: Final = 30.0 +_KILLED_SOURCES: Final = frozenset({"focus", "media"}) +_FAILCLOSED: Final = "failclosed" + +_Ident = tuple[int, int] + + +class KillLog(Protocol): + """Where kills are recorded (``BudgetLog``).""" + + def record(self, event: str, **fields: object) -> None: + """Append one record.""" + + +def fact_targets(fact: Now) -> dict[int, str]: + """Pids a trusted *fact* says must end, to the reason key that named them.""" + found: dict[int, str] = {} + for reason in fact.reasons: + if reason.pid is None: + continue + failclosed = reason.source == _FAILCLOSED + if failclosed or ( + reason.source in _KILLED_SOURCES and not is_browser(reason.pid) + ): + found.setdefault(reason.pid, reason.key) + return found + + +class LeisureKiller: + """The per-daemon kill state: when each process got its SIGTERM.""" + + def __init__(self, log: KillLog | None = None) -> None: + """Start with nothing signalled. + + Args: + log: The budget audit log; ``None`` records nowhere. + """ + self.log = log + self._termed: dict[_Ident, float] = {} + self._killed: set[_Ident] = set() + self._last_error = "" + + def enforce( + self, + state: PlaytimeState, + rules: PlaytimeRules, + *, + verdict: ShadowVerdict, + stamp: float, + ) -> None: + """Signal this tick's leisure targets if the leisure budget is spent. + + Never raises: the enforce loop has no per-tick guard. + + Args: + state: Accounting state after this tick's policy. + rules: Policy for this tick (``leisure`` set by the billing step). + verdict: This tick's shadow verdict, carrying the trusted fact. + stamp: Epoch seconds of this tick. + """ + try: + self._enforce(state, rules, verdict=verdict, stamp=stamp) + except Exception as exc: # the loop must outlive any kill bug + message = f"{type(exc).__name__}: {exc}" + if message != self._last_error: + self._last_error = message + logger.warning("Leisure kill failed (%s).", message, exc_info=True) + + def _enforce( + self, + state: PlaytimeState, + rules: PlaytimeRules, + *, + verdict: ShadowVerdict, + stamp: float, + ) -> None: + if not rules.leisure or rules.demo or not rules.enforcement: + return + if state.seconds < rules.budget_seconds: + self._termed.clear() + self._killed.clear() + return + if verdict.paused_by is not None: + return + if verdict.fact is None: + targets, why = sweep_targets(), "now.json untrusted" + else: + targets, why = fact_targets(verdict.fact), "now.json" + live = self._identify(targets) + self._termed = {i: t for i, t in self._termed.items() if i in live} + self._killed &= set(live) + base = { + "why": why, + "day_key": state.day_key, + "seconds": round(state.seconds, 3), + } + for ident, key in sorted(live.items()): + note = {"key": key, "name": exe_name(ident[0]), **base} + first = self._termed.get(ident) + if first is None: + self._termed[ident] = stamp + self._signal(ident[0], signal.SIGTERM, note) + elif stamp - first >= SIGKILL_AFTER_SECONDS: + quiet = ident in self._killed + 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]: + chain = own_chain() + live: dict[_Ident, str] = {} + for pid, key in targets.items(): + started = start_time(pid) + if started is not None and not is_protected(pid, chain): + live[pid, started] = key + return live + + def _signal( + self, pid: int, sig: signal.Signals, note: dict[str, object] | None + ) -> bool: + """Send *sig*; record it unless *note* is ``None``. True if delivered.""" + try: + os.kill(pid, sig) + except ProcessLookupError, PermissionError: + return False + if note is not None and self.log is not None: + self.log.record(EVENT_LEISURE_KILL, pid=pid, signal=sig.name, **note) + return True diff --git a/steam_backlog_enforcer/_leisure_mode.py b/steam_backlog_enforcer/_leisure_mode.py new file mode 100644 index 0000000..465580a --- /dev/null +++ b/steam_backlog_enforcer/_leisure_mode.py @@ -0,0 +1,114 @@ +"""Phase 2 of the leisure rollout: is it on for this gaming day, and its billing. + +Dormant until ``leisure-watch go-live`` writes the switch file. The only +question asked of it is :func:`earned_time.leisure_active` for the *billed* +gaming day (06:00 boundary, ``state.day_key`` after the roll-over), so the +billing mode always matches the day being billed; no code here reads the file. + +When on, the enforced ``seconds`` are billed by the shadow rule (contract C4, +:mod:`_shadow`) instead of games-only: a held locker grab or i3lock pauses +everything, games included; anything the watcher calls leisure -- or a watcher +that cannot be trusted -- bills; overlaps bill the wall clock once. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import date, datetime +from typing import TYPE_CHECKING, Final + +import earned_time +from leisure_watch.procscan import attributed_key + +from steam_backlog_enforcer._playtime_budget import accumulate, bill + +if TYPE_CHECKING: + from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState + from steam_backlog_enforcer._shadow import ShadowVerdict + +# Passed explicitly on every call: ``leisure_active``'s default is bound at +# import, so this module constant is what a test redirects. +LEISURE_FROM_PATH: Final = earned_time.LEISURE_FROM_PATH + + +def leisure_on(day_key: str) -> bool: + """Whether the gaming day *day_key* (``YYYY-MM-DD``) is on the leisure budget. + + Args: + day_key: The gaming day being billed. + + Returns: + ``earned_time.leisure_active`` for that day; ``False`` for a key that is + not a date (the games-only behaviour, which is what ran before). + """ + try: + day = date.fromisoformat(day_key) + except ValueError: + return False + return bool(earned_time.leisure_active(day, LEISURE_FROM_PATH)) + + +@dataclass(frozen=True) +class Tick: + """What one enforce-loop tick observed, for billing.""" + + now: datetime + """Timezone-aware local timestamp for this tick.""" + interval: float + """Nominal seconds between ticks.""" + owners: dict[int, str] + """Qualifying game PIDs to attribution keys.""" + verdict: ShadowVerdict + """This tick's shadow verdict.""" + + +def leisure_billed(state: PlaytimeState, tick: Tick) -> PlaytimeState: + """Bill this tick's enforced time by the shadow rule. + + Only a game (``attributed_key``) is ever credited to ``per_game``; other + leisure shows as unattributed, its split lives in ``shadow_per_source``. + + Args: + state: Accounting state (shadow already accumulated for this tick). + tick: This tick's observations. + + Returns: + The billed state. + """ + billed = tick.verdict.paused_by is None and bool(tick.verdict.keys) + return bill( + state, + now=tick.now, + billed=billed, + interval=tick.interval, + credited_key=attributed_key(tick.owners) if billed else "", + ) + + +def bill_tick( + state: PlaytimeState, rules: PlaytimeRules, tick: Tick +) -> tuple[PlaytimeState, PlaytimeRules]: + """Bill this tick's enforced time: games-only, or (Phase 2) by the shadow rule. + + Dormant (no switch, a day before it, or a demo): exactly the old + ``accumulate`` call, and *rules* returned as they came. + + Args: + state: Accounting state (shadow already accumulated for this tick). + rules: Policy for this tick. + tick: This tick's observations. + + Returns: + The billed state, and the tick's rules -- with ``leisure=True`` when + live, so the warnings, the cutoff and the kills act for leisure. + """ + if rules.demo or not leisure_on(state.day_key): + billed = accumulate( + state, + now=tick.now, + qualifying=set(tick.owners), + interval=tick.interval, + credited_key=attributed_key(tick.owners), + ) + return billed, rules + return leisure_billed(state, tick), replace(rules, leisure=True) diff --git a/steam_backlog_enforcer/_leisure_procs.py b/steam_backlog_enforcer/_leisure_procs.py new file mode 100644 index 0000000..3d28079 --- /dev/null +++ b/steam_backlog_enforcer/_leisure_procs.py @@ -0,0 +1,205 @@ +"""Who a leisure kill may touch: browsers, players, lockers, the desktop user. + +Read straight from ``/proc`` (the daemon is root), never from ``comm`` alone: +Firefox content processes rename it ("Isolated Web Co") and Chromium children +run as ``/proc/self/exe``, so a browser is named by its ``exe`` basename. + +Nothing here signals anything; :mod:`_leisure_kill` does, after +:func:`is_protected` says no to root/system processes, the enforcer's own +chain, and every locker and its children. +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Final + +from leisure_watch.facts import DESKTOP_UID + +_PROC: Final = Path("/proc") + +# THE browser list (Phase 2): exempt from focus/media kills because the +# extension redirects instead, and swept when the watcher cannot be trusted. +# Matched as the exe basename, or that name plus a "-"/"_" suffix +# (vivaldi-bin, firefox-bin, helium_crashpad_handler). +BROWSER_FAMILIES: Final = ( + "helium", + "chromium", + "chrome", + "google-chrome", + "brave", + "vivaldi", + "opera", + "msedge", + "microsoft-edge", + "firefox", + "librewolf", + "zen", + "thorium", + "floorp", + "waterfox", + "mullvad-browser", +) +MEDIA_PLAYERS: Final = ("mpv", "vlc", "totem", "freetube") +_WSG_NAMES: Final = frozenset({"wsg_grabber", "wsg-grabber"}) +# Headless `wsg-grabber scrape` never counts as leisure (Q7): never swept. +_WSG_HEADLESS: Final = "scrape" +# A process whose argv names one of these -- or any ancestor's does -- is a +# locker (or the accountant itself) and is never signalled. +LOCKER_NAMES: Final = frozenset( + { + "i3lock", + "xss_lock", + "gatelock", + "screen_locker", + "leetcode_guard", + "leetcode_guard_tray", + "book_guard", + "diet_guard", + "anki_guard", + "home_guard", + "meal_gate", + "workout_locker", + "steam_backlog_enforcer", + "leisure_watch", + } +) +_STAT_PPID: Final = 1 +_STAT_START: Final = 19 +_MAX_DEPTH: Final = 32 + + +def _read(pid: int, name: str) -> str | None: + try: + return (_PROC / str(pid) / name).read_text(encoding="utf-8", errors="replace") + except OSError: + return None + + +def exe_name(pid: int) -> str: + """Lower-case basename of the pid's binary (``comm`` if exe is unreadable).""" + try: + return (_PROC / str(pid) / "exe").readlink().name.lower() + except OSError: + return (_read(pid, "comm") or "").strip().lower() + + +def argv(pid: int) -> tuple[str, ...]: + """The pid's command line, ``()`` if gone.""" + raw = _read(pid, "cmdline") or "" + return tuple(arg for arg in raw.split("\0") if arg) + + +def real_uid(pid: int) -> int | None: + """The real uid from ``status``, ``None`` if unreadable.""" + for line in (_read(pid, "status") or "").splitlines(): + if line.startswith("Uid:"): + fields = line.split() + return int(fields[1]) if fields[1:] and fields[1].isdigit() else None + return None + + +def _stat_field(pid: int, index: int) -> int | None: + # comm may hold spaces and parentheses: split after the last ") ". + rest = (_read(pid, "stat") or "").rpartition(") ")[2].split() + if len(rest) <= index or not rest[index].isdigit(): + return None + return int(rest[index]) + + +def parent(pid: int) -> int | None: + """The parent pid, ``None`` if unreadable.""" + return _stat_field(pid, _STAT_PPID) + + +def start_time(pid: int) -> int | None: + """Start time in clock ticks: with the pid, the process's identity.""" + return _stat_field(pid, _STAT_START) + + +def _family(name: str, families: tuple[str, ...]) -> bool: + return any( + name == fam or name.startswith((f"{fam}-", f"{fam}_")) for fam in families + ) + + +def is_browser(pid: int) -> bool: + """Whether *pid* runs a browser binary (any process of it).""" + return _family(exe_name(pid), BROWSER_FAMILIES) + + +def _arg_names(pid: int) -> set[str]: + # "/x/leetcode_guard_tray.py" -> leetcode_guard_tray; "-m screen_locker._web" + # -> screen_locker; "book-guard" -> book_guard. + names = set() + for arg in argv(pid): + stem = Path(arg).name.lower().removesuffix(".py").partition(".")[0] + names.add(stem.replace("-", "_")) + return names + + +def _ancestry(pid: int) -> list[int]: + chain: list[int] = [] + current: int | None = pid + while current is not None and current > 0 and current not in chain: + if len(chain) >= _MAX_DEPTH: + break + chain.append(current) + current = parent(current) + return chain + + +def own_chain() -> set[int]: + """This daemon and its ancestors: never signalled.""" + return set(_ancestry(os.getpid())) + + +def is_protected(pid: int, chain: set[int]) -> bool: + """Whether *pid* must never be signalled by a leisure kill. + + Args: + pid: Candidate process. + chain: :func:`own_chain`, computed once per tick. + + Returns: + True for anything not the desktop user's (root, system, other users, + unreadable or gone), the enforcer's own chain, and lockers with + everything under them. + """ + if pid in chain or real_uid(pid) != DESKTOP_UID: + return True + return any(_arg_names(p) & LOCKER_NAMES for p in _ancestry(pid) if p > 1) + + +def _wsg_reviewer(pid: int) -> bool: + args = argv(pid) + for i, arg in enumerate(args): + if Path(arg).name in _WSG_NAMES: + return args[i + 1 : i + 2] != (_WSG_HEADLESS,) + return False + + +def sweep_targets() -> dict[int, str]: + """Every browser, media player and wsg-grabber reviewer on the machine. + + The kill set while the watcher cannot be trusted (stale, missing or badly + signed ``now.json``). + + Returns: + Pid to a ``sweep:...`` key naming why it was picked. + """ + try: + pids = [int(e.name) for e in _PROC.iterdir() if e.name.isdigit()] + except OSError: + return {} + found: dict[int, str] = {} + for pid in sorted(pids): + name = exe_name(pid) + if _family(name, BROWSER_FAMILIES): + found[pid] = f"sweep:browser:{name}" + elif _family(name, MEDIA_PLAYERS): + found[pid] = f"sweep:media:{name}" + elif _wsg_reviewer(pid): + found[pid] = "sweep:wsg-grabber" + return found diff --git a/steam_backlog_enforcer/_playtime.py b/steam_backlog_enforcer/_playtime.py index 97bf809..71a929a 100644 --- a/steam_backlog_enforcer/_playtime.py +++ b/steam_backlog_enforcer/_playtime.py @@ -23,9 +23,10 @@ suspends, missed ticks, daemon restarts and NTP steps all behave. ``python3``. The scan (``leisure_watch.procscan``, shared with the leisure watcher) therefore reads argv as well as ``comm``. -**Shadow leisure time rides the same tick** (:mod:`_shadow`, contract C4): it -is measured from the same stamp and clamp, before ``accumulate``, and never -touches an enforced field. Under a total block it is not billed. +**Shadow leisure time rides the same tick** (:mod:`_shadow`, contract C4), +from the same stamp and clamp, before ``accumulate``; under a total block it +is not billed. From the go-live day (:mod:`_leisure_mode`, Phase 2) its rule +bills the enforced count too, and at 0 :mod:`_leisure_kill` ends leisure apps. """ from __future__ import annotations @@ -36,15 +37,13 @@ import logging import time from typing import TYPE_CHECKING -from leisure_watch.procscan import attributed_key, qualifying_pids +from leisure_watch.procscan import qualifying_pids from steam_backlog_enforcer import config as config_mod from steam_backlog_enforcer._gaming_days import gaming_day_key +from steam_backlog_enforcer._leisure_mode import Tick, bill_tick from steam_backlog_enforcer._playtime_block import mounted_targets -from steam_backlog_enforcer._playtime_budget import ( - accumulate, - roll_over, -) +from steam_backlog_enforcer._playtime_budget import roll_over from steam_backlog_enforcer._playtime_cutoff import ( _begin_cutoff, _sustain_block, @@ -125,14 +124,11 @@ def playtime_tick( state = shadow_accumulate( state, stamp=now.timestamp(), interval=interval, verdict=shadow ) - state = accumulate( - state, - now=now, - qualifying=pids, - interval=interval, - credited_key=attributed_key(owners), - ) + tick = Tick(now=now, interval=interval, owners=owners, verdict=shadow) + state, rules = bill_tick(state, rules, tick) state = _policy(state, rules, now=now) + if rules.leisure: + session.leisure.enforce(state, rules, verdict=shadow, stamp=now.timestamp()) session.journal.observe(pids, state, rules=rules, now_monotonic=time.monotonic()) session.shadow.observe(shadow, state) save_state(state, demo=demo) diff --git a/steam_backlog_enforcer/_playtime_budget.py b/steam_backlog_enforcer/_playtime_budget.py index c7c9312..2f940c6 100644 --- a/steam_backlog_enforcer/_playtime_budget.py +++ b/steam_backlog_enforcer/_playtime_budget.py @@ -51,6 +51,39 @@ def accumulate( charged to a guess, which is why ``per_game`` only ever sums to at most ``seconds``. + Returns: + The updated state. + """ + return bill( + state, + now=now, + billed=bool(qualifying), + interval=interval, + credited_key=credited_key, + ) + + +def bill( + state: PlaytimeState, + *, + now: datetime, + billed: bool, + interval: float, + credited_key: str = "", +) -> PlaytimeState: + """:func:`accumulate` with the "anything qualifies?" answer given directly. + + The leisure budget (Phase 2) decides billing from the shadow verdict, not + from game PIDs; everything else -- the clamp, the first tick, a backward + step, the attribution -- is the same arithmetic. + + Args: + state: Current accounting state. + now: Timezone-aware local timestamp for this tick. + billed: Whether this tick counts against the budget. + interval: Nominal seconds between ticks. + credited_key: Attribution key to charge, or ``""``. + Returns: The updated state. """ @@ -59,7 +92,7 @@ def accumulate( return replace(state, last_tick_at=stamp) delta = stamp - state.last_tick_at - if delta <= 0.0 or not qualifying: + if delta <= 0.0 or not billed: return replace(state, last_tick_at=stamp) credited = min(delta, interval * _DELTA_CLAMP_FACTOR) diff --git a/steam_backlog_enforcer/_playtime_cutoff.py b/steam_backlog_enforcer/_playtime_cutoff.py index 89cf69b..d541955 100644 --- a/steam_backlog_enforcer/_playtime_cutoff.py +++ b/steam_backlog_enforcer/_playtime_cutoff.py @@ -53,6 +53,19 @@ def reconcile_for(rules: PlaytimeRules, *, should_block: bool) -> None: reconcile(should_block=should_block) +_CUTOFF_BODY = { + False: "Daily budget used up. Steam and games are shutting down. " + "Unblocks at 06:00.", + True: "Daily leisure budget used up. Steam, games and leisure apps are " + "shutting down. Unblocks at 06:00.", +} + + +def _noun(rules: PlaytimeRules) -> str: + """What the budget is called: "Leisure" from the go-live day (Phase 2).""" + return "Leisure" if rules.leisure else "Gaming" + + def _warn(state: PlaytimeState, rules: PlaytimeRules) -> PlaytimeState: """Fire the due warning, if any, and record that it fired. @@ -67,14 +80,16 @@ def _warn(state: PlaytimeState, rules: PlaytimeRules) -> PlaytimeState: if threshold is None: return state - left = f"{_humanise(threshold)} of today's gaming budget left." + noun = _noun(rules) + left = f"{_humanise(threshold)} of today's {noun.lower()} budget left." logger.info( - "Gaming warning: %s (billed %.0fs of %.0fs)", + "%s warning: %s (billed %.0fs of %.0fs)", + noun, left, state.seconds, rules.budget_seconds, ) - notify_desktop_user("Gaming time running out", left) + notify_desktop_user(f"{noun} time running out", left) return replace(state, warned_seconds=[*state.warned_seconds, threshold]) @@ -98,11 +113,10 @@ def _begin_cutoff( Returns: The updated state. """ - logger.warning("Daily gaming budget exhausted; shutting Steam down.") - notify_desktop_user( - "Gaming time is up", - "Daily budget used up. Steam and games are shutting down. Unblocks at 06:00.", + logger.warning( + "Daily %s budget exhausted; shutting Steam down.", _noun(rules).lower() ) + notify_desktop_user(f"{_noun(rules)} time is up", _CUTOFF_BODY[rules.leisure]) request_steam_shutdown() kill_gaming_processes(_kill_set(rules), force=False) if rules.demo: diff --git a/steam_backlog_enforcer/_playtime_session.py b/steam_backlog_enforcer/_playtime_session.py index 248e661..bb58f36 100644 --- a/steam_backlog_enforcer/_playtime_session.py +++ b/steam_backlog_enforcer/_playtime_session.py @@ -12,9 +12,10 @@ would otherwise have caught. from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import TYPE_CHECKING, Protocol +from steam_backlog_enforcer._leisure_kill import LeisureKiller from steam_backlog_enforcer._playtime_history import HistoryWriter from steam_backlog_enforcer._playtime_log import ( BudgetLog, @@ -56,6 +57,20 @@ class ShadowRecorder(Protocol): """Record the shadow verdict if it changed.""" +class LeisureEnforcer(Protocol): + """What the budget tick needs from the Phase 2 leisure killer.""" + + def enforce( + self, + state: PlaytimeState, + rules: PlaytimeRules, + *, + verdict: ShadowVerdict, + stamp: float, + ) -> None: + """Signal leisure apps if the leisure budget is spent.""" + + @dataclass(frozen=True) class PlaytimeSession: """Per-daemon logging state.""" @@ -63,6 +78,9 @@ class PlaytimeSession: journal: TickRecorder history: HistoryRecorder shadow: ShadowRecorder + # Defaulted (records nowhere) so a session built without it still kills; + # new_session wires it to the budget log. + leisure: LeisureEnforcer = field(default_factory=LeisureKiller) def new_session(*, demo: bool = False) -> PlaytimeSession: @@ -81,4 +99,5 @@ def new_session(*, demo: bool = False) -> PlaytimeSession: journal=TickJournal(log), history=HistoryWriter(), shadow=ShadowJournal(log), + leisure=LeisureKiller(log), ) diff --git a/steam_backlog_enforcer/_playtime_state.py b/steam_backlog_enforcer/_playtime_state.py index 6386417..1b134f2 100644 --- a/steam_backlog_enforcer/_playtime_state.py +++ b/steam_backlog_enforcer/_playtime_state.py @@ -97,6 +97,9 @@ class PlaytimeRules: budget_reason: str = "" # The shadow leisure budget (C4), held like budget_seconds; never enforced. shadow_budget_seconds: float = 0.0 + # Phase 2: the billed gaming day is on the leisure budget (set per tick by + # _leisure_mode.leisure_billed, never by rules_for). Only wording and kills. + leisure: bool = False def rules_for(config: Config, *, demo: bool) -> PlaytimeRules: diff --git a/steam_backlog_enforcer/_shadow.py b/steam_backlog_enforcer/_shadow.py index 978d17b..796885d 100644 --- a/steam_backlog_enforcer/_shadow.py +++ b/steam_backlog_enforcer/_shadow.py @@ -23,7 +23,7 @@ Overlapping sources bill the wall clock once and split it evenly, so from __future__ import annotations -from dataclasses import dataclass, replace +from dataclasses import dataclass, field, replace import logging from pathlib import Path import time @@ -67,6 +67,11 @@ class ShadowVerdict: """``gatelock:`` / ``i3lock`` while paused, else ``None``.""" keys: tuple[str, ...] """Sources billed this tick; empty = free (or paused).""" + fact: Now | None = field(default=None, compare=False, repr=False) + """The trusted watcher fact (Phase 2 kills); ``None`` if paused or untrusted. + + Out of equality: ts/seq change every 5 s and the journal dedups on it. + """ @property def state(self) -> str: @@ -121,7 +126,7 @@ def observe_shadow(owners: dict[int, str], *, stamp: float) -> ShadowVerdict: "Shadow inputs unreadable (%r); failing closed.", exc, exc_info=True ) fact = None - return ShadowVerdict(paused_by=None, keys=shadow_keys(owners, fact)) + return ShadowVerdict(paused_by=None, keys=shadow_keys(owners, fact), fact=fact) def _should_warn(exc: BaseException) -> bool: diff --git a/steam_backlog_enforcer/tests/_isolate_leisure.py b/steam_backlog_enforcer/tests/_isolate_leisure.py new file mode 100644 index 0000000..7845af3 --- /dev/null +++ b/steam_backlog_enforcer/tests/_isolate_leisure.py @@ -0,0 +1,52 @@ +"""Autouse isolation of the Phase-2 leisure switch and the leisure kill scan. + +The switch file ``/etc/leisure-watch/leisure_from`` is read twice per tick: +by ``earned_time`` itself (``is_leisure``, looked up at call time from +``earned_time._leisure``) and by the enforcer's gate, which passes its own +constant explicitly. Both point at an absent file under ``tmp_path``, so no +test ever reads the real one and every test starts dormant. A test turns +Phase 2 on by writing a date to :func:`switch_file`. + +The leisure kill scan reads ``/proc`` through its own constant; it gets an +empty directory (:func:`leisure_proc`), so no test can see -- or signal -- a +live browser or player. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_SWITCH_NAME = "leisure_from" +_PROC_NAME = "leisure_proc" + + +def switch_file(tmp_path: Path) -> Path: + """The redirected switch file for this test (absent until written).""" + return tmp_path / "etc-leisure-watch" / _SWITCH_NAME + + +def leisure_proc(tmp_path: Path) -> Path: + """The redirected ``/proc`` the leisure kill scan reads for this test.""" + return tmp_path / _PROC_NAME + + +@pytest.fixture(autouse=True) +def _isolate_leisure(tmp_path: Path) -> Iterator[None]: + """Point both switch readers and the kill scan at ``tmp_path``.""" + switch = switch_file(tmp_path) + switch.parent.mkdir(exist_ok=True) + proc = leisure_proc(tmp_path) + proc.mkdir(exist_ok=True) + with ( + patch("earned_time._leisure.LEISURE_FROM_PATH", switch), + patch("steam_backlog_enforcer._leisure_mode.LEISURE_FROM_PATH", switch), + patch("steam_backlog_enforcer._leisure_procs._PROC", proc), + ): + yield diff --git a/steam_backlog_enforcer/tests/_leisure_fake_proc.py b/steam_backlog_enforcer/tests/_leisure_fake_proc.py new file mode 100644 index 0000000..c65b9e8 --- /dev/null +++ b/steam_backlog_enforcer/tests/_leisure_fake_proc.py @@ -0,0 +1,48 @@ +"""A tiny fake ``/proc`` for the leisure kill tests: one directory per pid.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pathlib import Path + +# Fields after ``comm``: state, ppid, then 17 more, then starttime (index 19). +_FILLER_BEFORE_START = 17 +_FILLER_AFTER_START = 5 + + +@dataclass(frozen=True) +class FakeProc: + """What one fake ``/proc/`` holds.""" + + pid: int + exe: str = "/usr/bin/mpv" + argv: tuple[str, ...] = ("mpv",) + comm: str = "mpv" + uid: int = 1000 + ppid: int = 1 + start: int = 100 + + +def add_proc(root: Path, spec: FakeProc) -> None: + """Create ``root/`` with exe, cmdline, comm, status and stat. + + Args: + root: The fake ``/proc``. + spec: The process to plant (a dangling ``exe`` target is fine). + """ + proc = root / str(spec.pid) + proc.mkdir(parents=True) + (proc / "exe").symlink_to(spec.exe) + (proc / "cmdline").write_bytes(b"".join(a.encode() + b"\0" for a in spec.argv)) + (proc / "comm").write_text(spec.comm + "\n", encoding="utf-8") + uid = spec.uid + (proc / "status").write_text( + f"Name:\t{spec.comm}\nUid:\t{uid}\t{uid}\t{uid}\t{uid}\n", encoding="utf-8" + ) + rest = ["S", str(spec.ppid), *["0"] * _FILLER_BEFORE_START, str(spec.start)] + rest += ["0"] * _FILLER_AFTER_START + line = f"{spec.pid} ({spec.comm}) {' '.join(rest)}\n" + (proc / "stat").write_text(line, encoding="utf-8") diff --git a/steam_backlog_enforcer/tests/conftest.py b/steam_backlog_enforcer/tests/conftest.py index 94e65f1..754f3b6 100644 --- a/steam_backlog_enforcer/tests/conftest.py +++ b/steam_backlog_enforcer/tests/conftest.py @@ -26,6 +26,7 @@ from steam_backlog_enforcer.tests._isolate_control_plane import ( _isolate_cover_art, ) from steam_backlog_enforcer.tests._isolate_incidents import _isolate_incidents +from steam_backlog_enforcer.tests._isolate_leisure import _isolate_leisure from steam_backlog_enforcer.tests._isolate_playtime import _isolate_playtime from steam_backlog_enforcer.tests._isolate_reading import _isolate_reading from steam_backlog_enforcer.tests._isolate_shadow import _isolate_shadow @@ -46,6 +47,7 @@ __all__ = [ "_isolate_control_plane", "_isolate_cover_art", "_isolate_incidents", + "_isolate_leisure", "_isolate_playtime", "_isolate_reading", "_isolate_shadow", diff --git a/steam_backlog_enforcer/tests/test_leisure_kill.py b/steam_backlog_enforcer/tests/test_leisure_kill.py new file mode 100644 index 0000000..618cfca --- /dev/null +++ b/steam_backlog_enforcer/tests/test_leisure_kill.py @@ -0,0 +1,247 @@ +"""Phase 2 at 0: which leisure pids die, how, and what the audit log says.""" + +from __future__ import annotations + +from dataclasses import replace +import os +import signal +from typing import TYPE_CHECKING +from unittest.mock import patch + +from leisure_watch.facts import Focus, Now, Reason +import pytest + +from steam_backlog_enforcer import _leisure_kill +from steam_backlog_enforcer._leisure_kill import ( + EVENT_LEISURE_KILL, + LeisureKiller, + fact_targets, +) +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 Iterator + from pathlib import Path + from unittest.mock import MagicMock + +_HELIUM = "/opt/helium-browser-bin/helium" +_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) + + +class StubLog: + def __init__(self) -> None: + self.records: list[tuple[str, dict[str, object]]] = [] + + def record(self, event: str, **fields: object) -> None: + self.records.append((event, fields)) + + +def _now(*reasons: Reason) -> Now: + return Now( + ts=0.0, + seq=1, + state="leisure", + paused_by=None, + reasons=reasons, + focus=Focus("Helium", "", None, idle=False), + ) + + +def _verdict(fact: Now | None, paused: str | None = None) -> ShadowVerdict: + return ShadowVerdict(paused_by=paused, keys=("k",), fact=fact) + + +@pytest.fixture +def proc(tmp_path: Path) -> Path: + root = leisure_proc(tmp_path) + add_proc(root, FakeProc(10, exe="/usr/bin/mpv")) + add_proc(root, FakeProc(20, exe=_HELIUM, comm="helium")) + add_proc(root, FakeProc(30, exe="/usr/bin/chromium", comm="chromium")) + add_proc(root, FakeProc(40, exe="/usr/bin/vesktop", uid=0)) + return root + + +@pytest.fixture +def kill() -> Iterator[MagicMock]: + with patch.object(os, "kill") as mock: + yield mock + + +class TestFactTargets: + def test_media_and_focus_apps_die_browsers_and_games_do_not( + self, proc: Path + ) -> None: + del proc + fact = _now( + Reason("media", "app:mpv", 10, "mpv"), + Reason("focus", "domain:youtube.com", 20, "YouTube"), + Reason("media", "domain:youtube.com", None, "url"), + Reason("game", "steam:440", 50, "tf2"), + Reason("mystery", "x", 10, "?"), + ) + assert fact_targets(fact) == {10: "app:mpv"} + + @pytest.mark.parametrize( + ("pid", "key"), + [ + (20, "failclosed:helium"), + (20, "failclosed:helium:unreported"), + (20, "failclosed:helium:audio"), + (30, "browser:chromium"), + ], + ) + def test_a_non_reporting_browser_dies(self, proc: Path, pid: int, key: str) -> None: + del proc + fact = _now(Reason("failclosed", key, pid, "x")) + assert fact_targets(fact) == {pid: key} + + def test_failclosed_without_a_pid_names_nothing(self) -> None: + fact = _now(Reason("failclosed", "failclosed:display", None, "X")) + assert fact_targets(fact) == {} + + +class TestEnforce: + def test_mpv_is_terminated_then_killed_after_30_s( + self, proc: Path, kill: MagicMock + ) -> None: + del proc + log = StubLog() + killer = LeisureKiller(log) + verdict = _verdict( + _now( + Reason("media", "app:mpv", 10, "mpv"), + Reason("focus", "domain:x.com", 20, "x"), + ) + ) + for stamp in (1000.0, 1003.0, 1029.0, 1030.0, 1033.0): + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=stamp) + assert [c.args for c in kill.call_args_list] == [ + (10, signal.SIGTERM), + (10, signal.SIGKILL), + (10, signal.SIGKILL), + ] + assert [(e, f["pid"], f["signal"]) for e, f in log.records] == [ + (EVENT_LEISURE_KILL, 10, "SIGTERM"), + (EVENT_LEISURE_KILL, 10, "SIGKILL"), + ] + first = log.records[0][1] + assert first["key"] == "app:mpv" + assert first["name"] == "mpv" + assert first["why"] == "now.json" + assert first["day_key"] == "2026-11-02" + + def test_untrusted_watcher_sweeps_but_spares_root( + 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 + ) + 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: + del proc + LeisureKiller(StubLog()).enforce( + _SPENT, _RULES, verdict=_verdict(None, paused="i3lock"), stamp=1.0 + ) + kill.assert_not_called() + + def test_under_budget_forgets_the_timers(self, proc: Path, kill: MagicMock) -> None: + del proc + killer = LeisureKiller(None) + verdict = _verdict(_now(Reason("media", "app:mpv", 10, "mpv"))) + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=0.0) + under = PlaytimeState(day_key="2026-11-02", seconds=50.0) + killer.enforce(under, _RULES, verdict=verdict, stamp=10.0) + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=40.0) + assert [c.args[1] for c in kill.call_args_list] == [signal.SIGTERM] * 2 + + def test_a_target_that_drops_out_restarts_its_timer( + self, proc: Path, kill: MagicMock + ) -> None: + del proc + killer = LeisureKiller(None) + listed = _verdict(_now(Reason("media", "app:mpv", 10, "mpv"))) + killer.enforce(_SPENT, _RULES, verdict=listed, stamp=0.0) + killer.enforce(_SPENT, _RULES, verdict=_verdict(_now()), stamp=3.0) + killer.enforce(_SPENT, _RULES, verdict=listed, stamp=40.0) + assert [c.args[1] for c in kill.call_args_list] == [signal.SIGTERM] * 2 + + def test_a_pid_reused_by_a_new_process_starts_over( + self, proc: Path, kill: MagicMock + ) -> None: + killer = LeisureKiller(None) + verdict = _verdict(_now(Reason("media", "app:mpv", 10, "mpv"))) + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=0.0) + (proc / "10" / "stat").write_text( + "10 (mpv) S 1 " + "0 " * 17 + "999 0", encoding="utf-8" + ) + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=40.0) + assert [c.args[1] for c in kill.call_args_list] == [signal.SIGTERM] * 2 + + def test_a_vanished_or_unsignallable_pid_is_not_logged( + self, proc: Path, kill: MagicMock + ) -> None: + add_proc(proc, FakeProc(11, exe="/usr/bin/vlc")) + (proc / "12").mkdir() + (proc / "12" / "status").write_text("Uid:\t1000\t1000\n", encoding="utf-8") + kill.side_effect = ProcessLookupError + log = StubLog() + verdict = _verdict( + _now( + Reason("media", "app:vlc", 11, "vlc"), + Reason("media", "app:ghost", 12, "?"), + ) + ) + LeisureKiller(log).enforce(_SPENT, _RULES, verdict=verdict, stamp=0.0) + assert [c.args[0] for c in kill.call_args_list] == [11] + assert log.records == [] + + @pytest.mark.parametrize( + "rules", + [ + replace(_RULES, enforcement=False), + replace(_RULES, demo=True), + replace(_RULES, leisure=False), + ], + ) + def test_never_with_enforcement_off_in_a_demo_or_dormant( + self, proc: Path, kill: MagicMock, rules: PlaytimeRules + ) -> None: + del proc + LeisureKiller(None).enforce(_SPENT, rules, verdict=_verdict(None), stamp=0.0) + kill.assert_not_called() + + def test_a_bug_never_escapes_into_the_loop( + self, caplog: pytest.LogCaptureFixture + ) -> 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) + 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( + self, proc: Path, kill: MagicMock + ) -> None: + del proc + kill.side_effect = [None, ProcessLookupError, None, None] + log = StubLog() + killer = LeisureKiller(log) + verdict = _verdict(_now(Reason("media", "app:mpv", 10, "mpv"))) + for stamp in (0.0, 30.0, 33.0, 36.0): + killer.enforce(_SPENT, _RULES, verdict=verdict, stamp=stamp) + assert [f["signal"] for _e, f in log.records] == ["SIGTERM", "SIGKILL"] + assert kill.call_count == 4 diff --git a/steam_backlog_enforcer/tests/test_leisure_mode.py b/steam_backlog_enforcer/tests/test_leisure_mode.py new file mode 100644 index 0000000..ebac3f9 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_leisure_mode.py @@ -0,0 +1,160 @@ +"""Phase 2 gate and billing: ``leisure_on`` and ``leisure_billed``.""" + +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone +import importlib +from pathlib import Path +from unittest.mock import patch + +import earned_time +import pytest + +from steam_backlog_enforcer import _leisure_mode +from steam_backlog_enforcer._budget_resolve import resolve_budget +from steam_backlog_enforcer._leisure_mode import ( + Tick, + bill_tick, + leisure_billed, + leisure_on, +) +from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState +from steam_backlog_enforcer._shadow import ShadowVerdict, shadow_accumulate +from steam_backlog_enforcer.config import Config +from steam_backlog_enforcer.tests._budget_dates import AFTER_CUT, pin_today +from steam_backlog_enforcer.tests._isolate_leisure import switch_file + +_LOCAL = timezone(timedelta(hours=2)) +_NOW = datetime(2026, 11, 2, 20, 0, tzinfo=_LOCAL) +_DAY = "2026-11-02" +_REAL_SWITCH = Path("/etc/leisure-watch/leisure_from") +_RULES = PlaytimeRules( + budget_seconds=100.0, + warn_at=(), + sigkill_after=30.0, + count_launchers=False, + enforcement=True, + demo=False, +) + + +def _switch_on(tmp_path: Path, day: str) -> None: + switch_file(tmp_path).write_text(day + "\n", encoding="ascii") + + +class TestHygiene: + def test_no_test_reads_the_real_switch_file(self) -> None: + assert _leisure_mode.LEISURE_FROM_PATH != _REAL_SWITCH + reader = importlib.import_module("earned_time._leisure") + assert reader.LEISURE_FROM_PATH != _REAL_SWITCH + + def test_the_redirect_reaches_both_readers(self, tmp_path: Path) -> None: + with pin_today(AFTER_CUT): + dormant = resolve_budget(Config()).base_seconds + _switch_on(tmp_path, AFTER_CUT.isoformat()) + live = resolve_budget(Config()).base_seconds + assert leisure_on(AFTER_CUT.isoformat()) + assert live - dormant == 2 * 3600.0 + + +class TestLeisureOn: + def test_absent_switch_is_dormant(self) -> None: + assert not leisure_on(_DAY) + + def test_the_switch_day_and_later_are_on(self, tmp_path: Path) -> None: + _switch_on(tmp_path, _DAY) + assert leisure_on(_DAY) + assert leisure_on("2026-11-03") + + def test_the_day_before_is_off(self, tmp_path: Path) -> None: + _switch_on(tmp_path, _DAY) + assert not leisure_on("2026-11-01") + + def test_a_garbage_day_key_is_dormant(self, tmp_path: Path) -> None: + _switch_on(tmp_path, _DAY) + assert not leisure_on("not-a-day") + + def test_the_gate_asks_earned_time(self) -> None: + with patch.object(earned_time, "leisure_active", return_value=True) as ask: + assert leisure_on(_DAY) + ask.assert_called_once_with(date(2026, 11, 2), _leisure_mode.LEISURE_FROM_PATH) + + +def _start() -> PlaytimeState: + return PlaytimeState(day_key=_DAY, last_tick_at=_NOW.timestamp() - 3.0) + + +def _tick(verdict: ShadowVerdict, owners: dict[int, str] | None = None) -> Tick: + return Tick(now=_NOW, interval=3.0, owners=owners or {}, verdict=verdict) + + +def _bill( + verdict: ShadowVerdict, owners: dict[int, str] | None = None +) -> tuple[PlaytimeState, bool]: + state, rules = bill_tick(_start(), _RULES, _tick(verdict, owners)) + return state, rules.leisure + + +@pytest.fixture +def live(tmp_path: Path) -> None: + _switch_on(tmp_path, _DAY) + + +@pytest.mark.usefixtures("live") +class TestLeisureBilled: + def test_leisure_bills_enforced_seconds_unattributed(self) -> None: + state, leisure = _bill(ShadowVerdict(paused_by=None, keys=("app:mpv",))) + assert (state.seconds, state.per_game, leisure) == (3.0, {}, True) + assert state.last_tick_at == _NOW.timestamp() + + def test_a_game_tick_credits_the_game(self) -> None: + verdict = ShadowVerdict(paused_by=None, keys=("app:440",)) + state, _ = _bill(verdict, {7: "app:440"}) + assert state.per_game == {"app:440": 3.0} + + def test_paused_bills_nothing_even_with_a_game(self) -> None: + verdict = ShadowVerdict(paused_by="i3lock", keys=()) + state, leisure = _bill(verdict, {7: "app:440"}) + assert (state.seconds, state.per_game, leisure) == (0.0, {}, True) + assert state.last_tick_at == _NOW.timestamp() + + def test_free_bills_nothing(self) -> None: + state, _ = _bill(ShadowVerdict(paused_by=None, keys=())) + assert state.seconds == 0.0 + + def test_enforced_and_shadow_move_together(self) -> None: + verdicts = ( + ShadowVerdict(paused_by=None, keys=("domain:youtube.com",)), + ShadowVerdict(paused_by="i3lock", keys=()), + ShadowVerdict(paused_by=None, keys=()), + ShadowVerdict(paused_by=None, keys=("app:440", "app:mpv")), + ) + state = _start() + for i, verdict in enumerate(verdicts): + now = _NOW + timedelta(seconds=3 * i) + before = state + state = shadow_accumulate( + state, stamp=now.timestamp(), interval=3.0, verdict=verdict + ) + tick = Tick(now=now, interval=3.0, owners={}, verdict=verdict) + state = leisure_billed(state, tick) + moved = state.seconds - before.seconds + assert moved == state.shadow_seconds - before.shadow_seconds + + +class TestDormantBilling: + def test_games_only_and_rules_untouched(self) -> None: + verdict = ShadowVerdict(paused_by=None, keys=("app:mpv",)) + state, leisure = _bill(verdict) + assert (state.seconds, leisure) == (0.0, False) + state, _ = _bill(ShadowVerdict(paused_by="i3lock", keys=()), {7: "app:440"}) + assert state.per_game == {"app:440": 3.0} + + def test_a_demo_is_never_live(self, live: None) -> None: + del live + demo = PlaytimeRules( + 1.0, (), 1.0, count_launchers=False, enforcement=True, demo=True + ) + verdict = ShadowVerdict(paused_by=None, keys=("app:mpv",)) + state, rules = bill_tick(_start(), demo, _tick(verdict)) + assert (state.seconds, rules) == (0.0, demo) diff --git a/steam_backlog_enforcer/tests/test_leisure_procs.py b/steam_backlog_enforcer/tests/test_leisure_procs.py new file mode 100644 index 0000000..24920f2 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_leisure_procs.py @@ -0,0 +1,174 @@ +"""Process identity for leisure kills: browsers, players, lockers, ownership.""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING + +import pytest + +from steam_backlog_enforcer import _leisure_procs as procs +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 + + +@pytest.fixture +def proc(tmp_path: Path) -> Path: + return leisure_proc(tmp_path) + + +class TestNames: + @pytest.mark.parametrize( + "exe", + [ + "/opt/helium-browser-bin/helium", + "/usr/lib/librewolf/librewolf", + "/usr/lib/chromium/chromium", + "/opt/google/chrome/chrome", + "/opt/brave.com/brave/brave", + "/opt/vivaldi/vivaldi-bin", + "/usr/lib/opera/opera", + "/opt/microsoft/msedge/msedge", + "/usr/lib/firefox/firefox-bin", + "/opt/zen-browser-bin/zen-bin", + "/opt/helium-browser-bin/helium_crashpad_handler", + ], + ) + def test_browser_families_by_exe(self, proc: Path, exe: str) -> None: + add_proc(proc, FakeProc(10, exe=exe, comm="Isolated Web Co")) + assert procs.is_browser(10) + + @pytest.mark.parametrize( + "exe", ["/usr/bin/mpv", "/usr/bin/operator", "/usr/bin/frozen", "/x/knowledge"] + ) + def test_non_browsers(self, proc: Path, exe: str) -> None: + add_proc(proc, FakeProc(10, exe=exe)) + assert not procs.is_browser(10) + + def test_unreadable_exe_falls_back_to_comm(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, comm="helium")) + (proc / "10" / "exe").unlink() + assert procs.exe_name(10) == "helium" + assert procs.is_browser(10) + + def test_a_vanished_pid_has_no_name(self) -> None: + assert procs.exe_name(99) == "" + assert procs.argv(99) == () + assert procs.real_uid(99) is None + assert procs.start_time(99) is None + assert procs.parent(99) is None + + def test_reads_uid_start_and_parent(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, uid=1000, ppid=7, start=4242)) + assert (procs.real_uid(10), procs.start_time(10), procs.parent(10)) == ( + 1000, + 4242, + 7, + ) + + def test_malformed_status_and_stat(self, proc: Path) -> None: + add_proc(proc, FakeProc(10)) + (proc / "10" / "status").write_text("Name:\tmpv\nUid:\tx\n", encoding="utf-8") + (proc / "10" / "stat").write_text("10 (mpv) S", encoding="utf-8") + assert procs.real_uid(10) is None + assert procs.start_time(10) is None + (proc / "10" / "stat").write_text("10 (mpv) S x", encoding="utf-8") + assert procs.parent(10) is None + + def test_status_without_a_uid_line(self, proc: Path) -> None: + add_proc(proc, FakeProc(10)) + (proc / "10" / "status").write_text("Name:\tmpv\n", encoding="utf-8") + assert procs.real_uid(10) is None + + +class TestProtected: + def test_a_plain_user_player_is_fair_game(self, proc: Path) -> None: + add_proc(proc, FakeProc(10)) + assert not procs.is_protected(10, set()) + + @pytest.mark.parametrize("uid", [0, 999, 1001]) + def test_anything_not_the_desktop_users(self, proc: Path, uid: int) -> None: + add_proc(proc, FakeProc(10, uid=uid)) + assert procs.is_protected(10, set()) + + def test_a_vanished_pid_is_never_signalled(self) -> None: + assert procs.is_protected(99, set()) + + def test_own_chain(self, proc: Path) -> None: + add_proc(proc, FakeProc(10)) + assert procs.is_protected(10, {10}) + + @pytest.mark.parametrize( + "argv", + [ + ("i3lock", "--nofork"), + ("/usr/bin/python3", "-m", "screen_locker._web_server"), + ("python3", "/home/kuhy/.config/i3/scripts/leetcode_guard_tray.py"), + ("/usr/bin/book-guard",), + ], + ) + def test_lockers_and_their_children( + self, proc: Path, argv: tuple[str, ...] + ) -> None: + add_proc(proc, FakeProc(5, exe="/usr/bin/python3", argv=argv, comm="x")) + add_proc(proc, FakeProc(10, ppid=5)) + assert procs.is_protected(5, set()) + assert procs.is_protected(10, set()) + + def test_a_parent_loop_terminates(self, proc: Path) -> None: + add_proc(proc, FakeProc(10, ppid=11)) + add_proc(proc, FakeProc(11, ppid=10)) + assert not procs.is_protected(10, set()) + + def test_own_chain_walks_up(self, proc: Path) -> None: + me = os.getpid() + add_proc(proc, FakeProc(me, ppid=77)) + add_proc(proc, FakeProc(77, ppid=1)) + assert procs.own_chain() == {me, 77, 1} + + +class TestSweep: + def test_untrusted_watcher_sweeps_browsers_players_and_the_reviewer( + self, proc: Path + ) -> None: + add_proc(proc, FakeProc(10, exe="/opt/helium-browser-bin/helium")) + add_proc(proc, FakeProc(11, exe="/usr/bin/mpv")) + add_proc(proc, FakeProc(12, exe="/opt/FreeTube/freetube")) + add_proc( + proc, + FakeProc( + 13, exe="/usr/bin/python3.14", argv=("python3", "-m", "wsg_grabber") + ), + ) + add_proc( + proc, + FakeProc( + 14, + exe="/usr/bin/python3.14", + argv=("python3", "-m", "wsg_grabber", "scrape"), + ), + ) + add_proc(proc, FakeProc(15, exe="/usr/bin/terminator")) + add_proc(proc, FakeProc(16, exe="/usr/bin/wsg-grabber", argv=("wsg-grabber",))) + (proc / "self").mkdir() + assert procs.sweep_targets() == { + 10: "sweep:browser:helium", + 11: "sweep:media:mpv", + 12: "sweep:media:freetube", + 13: "sweep:wsg-grabber", + 16: "sweep:wsg-grabber", + } + + def test_an_unlistable_proc_sweeps_nothing(self, proc: Path) -> None: + proc.rmdir() + assert procs.sweep_targets() == {} + + def test_a_deep_tree_walk_is_bounded(self, proc: Path) -> None: + for pid in range(100, 150): + add_proc(proc, FakeProc(pid, ppid=pid + 1)) + add_proc(proc, FakeProc(150, argv=("i3lock",))) + # The locker is 50 generations up: past the bound, so not seen. + assert not procs.is_protected(100, set()) diff --git a/steam_backlog_enforcer/tests/test_leisure_tick.py b/steam_backlog_enforcer/tests/test_leisure_tick.py new file mode 100644 index 0000000..04ef582 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_leisure_tick.py @@ -0,0 +1,183 @@ +"""The tick with Phase 2 dormant (no new path runs) and live (leisure billing).""" + +from __future__ import annotations + +from contextlib import ExitStack +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from steam_backlog_enforcer import _leisure_mode, _shadow +from steam_backlog_enforcer._leisure_kill import LeisureKiller +from steam_backlog_enforcer._playtime import playtime_tick +from steam_backlog_enforcer._playtime_cutoff import _begin_cutoff, _warn +from steam_backlog_enforcer._playtime_session import new_session +from steam_backlog_enforcer._playtime_state import ( + PlaytimeRules, + PlaytimeState, + load_state, + save_state, +) +from steam_backlog_enforcer.config import Config +from steam_backlog_enforcer.tests._fake_session import fake_session +from steam_backlog_enforcer.tests._isolate_leisure import switch_file +from steam_backlog_enforcer.tests._no_workout_http import fixed_budget + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + from steam_backlog_enforcer._shadow import ShadowVerdict + +_LOCAL = timezone(timedelta(hours=2)) +_NOW = datetime(2026, 7, 27, 20, 0, tzinfo=_LOCAL) +_TODAY = "2026-07-27" +_PKG = "steam_backlog_enforcer._playtime" +_CUTOFF = "steam_backlog_enforcer._playtime_cutoff" + + +class SpyLeisure: + def __init__(self) -> None: + self.calls: list[tuple[PlaytimeState, PlaytimeRules]] = [] + + def enforce( + self, + state: PlaytimeState, + rules: PlaytimeRules, + *, + verdict: ShadowVerdict, + stamp: float, + ) -> None: + del verdict, stamp + self.calls.append((state, rules)) + + +@pytest.fixture +def notify() -> Iterator[MagicMock]: + with ExitStack() as stack: + stack.enter_context(patch(f"{_CUTOFF}.reconcile")) + stack.enter_context(patch(f"{_CUTOFF}.request_steam_shutdown")) + stack.enter_context(patch(f"{_CUTOFF}.kill_gaming_processes")) + stack.enter_context(patch(f"{_PKG}.mounted_targets", return_value=set())) + stack.enter_context(patch(f"{_PKG}.is_total_block_active", return_value=False)) + yield stack.enter_context(patch(f"{_CUTOFF}.notify_desktop_user")) + + +def _tick( + owners: dict[int, str], *, budget: float = 3600.0, demo: bool = False +) -> tuple[PlaytimeState, SpyLeisure]: + save_state( + PlaytimeState(day_key=_TODAY, last_tick_at=_NOW.timestamp() - 3.0), + demo=demo, + ) + spy = SpyLeisure() + session = replace(fake_session(), leisure=spy) + with ( + fixed_budget(budget), + patch(f"{_PKG}.qualifying_pids", return_value=owners), + patch(f"{_PKG}.datetime") as mock_dt, + ): + mock_dt.now.return_value = _NOW + playtime_tick(Config(), interval=3.0, session=session, demo=demo) + state = load_state(demo=demo) + assert state is not None + return state, spy + + +def _go_live(tmp_path: Path) -> None: + switch_file(tmp_path).write_text(_TODAY + "\n", encoding="ascii") + + +@pytest.mark.usefixtures("notify") +class TestDormant: + def test_no_new_code_path_runs_without_the_switch(self) -> None: + with patch.object( + _leisure_mode, "leisure_billed", side_effect=AssertionError("ran") + ): + state, spy = _tick({}) + assert spy.calls == [] + # Games-only, exactly as before: the fail-closed shadow does not bill. + assert (state.seconds, state.shadow_seconds) == (0.0, 3.0) + + def test_games_still_bill_games_only(self) -> None: + state, spy = _tick({7: "app:440"}) + assert (state.seconds, state.per_game, spy.calls) == (3.0, {"app:440": 3.0}, []) + + def test_a_demo_ignores_the_switch(self, tmp_path: Path) -> None: + _go_live(tmp_path) + state, spy = _tick({}, demo=True) + assert (state.seconds, spy.calls) == (0.0, []) + + +@pytest.mark.usefixtures("notify") +class TestLive: + def test_leisure_bills_the_enforced_count(self, tmp_path: Path) -> None: + _go_live(tmp_path) + state, spy = _tick({}) + assert state.seconds == state.shadow_seconds == 3.0 + assert state.shadow_per_source == {_shadow.FAILCLOSED_KEY: 3.0} + [(_seen, rules)] = spy.calls + assert rules.leisure + + def test_a_lock_pauses_the_enforced_count_too(self, tmp_path: Path) -> None: + _go_live(tmp_path) + with patch.object(_shadow, "paused_by", return_value="i3lock"): + state, _ = _tick({7: "app:440"}) + assert (state.seconds, state.shadow_paused_seconds) == (0.0, 3.0) + + def test_at_zero_the_cutoff_says_leisure_and_kills_run( + self, tmp_path: Path, notify: MagicMock + ) -> None: + _go_live(tmp_path) + state, spy = _tick({}, budget=3.0) + assert state.is_blocked() + assert notify.call_args.args[0] == "Leisure time is up" + assert "leisure" in notify.call_args.args[1] + assert spy.calls[0][0].seconds >= spy.calls[0][1].budget_seconds + + +_RULES = PlaytimeRules( + budget_seconds=3600.0, + warn_at=(3600, 1800), + sigkill_after=30.0, + count_launchers=False, + enforcement=True, + demo=False, +) + + +class TestWording: + def test_warnings_say_gaming_while_dormant(self) -> None: + with patch(f"{_CUTOFF}.notify_desktop_user") as notify: + _warn(PlaytimeState(day_key=_TODAY, seconds=10.0), _RULES) + notify.assert_called_once_with( + "Gaming time running out", "1 hour of today's gaming budget left." + ) + + def test_warnings_say_leisure_when_live(self) -> None: + with patch(f"{_CUTOFF}.notify_desktop_user") as notify: + _warn( + PlaytimeState(day_key=_TODAY, seconds=10.0), + replace(_RULES, leisure=True), + ) + notify.assert_called_once_with( + "Leisure time running out", "1 hour of today's leisure budget left." + ) + + @pytest.mark.usefixtures("notify") + def test_the_dormant_cutoff_text_is_unchanged(self, notify: MagicMock) -> None: + _begin_cutoff(PlaytimeState(day_key=_TODAY), _RULES, now=_NOW) + notify.assert_called_once_with( + "Gaming time is up", + "Daily budget used up. Steam and games are shutting down. " + "Unblocks at 06:00.", + ) + + +def test_new_session_kills_through_the_budget_log() -> None: + session = new_session() + assert isinstance(session.leisure, LeisureKiller) + assert session.leisure.log is not None