kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-10 19:27:04 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

playtime: shadow leisure billing (C4); /proc scan from leisure_watch

Browse Source 
The enforcer now imports the Steam/launcher/counted-process scan from
leisure_watch.procscan (v0.1.0, shared with the leisure watcher) and drops
its own copies (_playtime_procs, _counted_procs, _total_block_launchers and
enforcer's two scanners). Every counted-process read passes
config.CONFIG_FILE explicitly, as before.

Each tick also bills a shadow leisure budget beside the enforced one, from
the same incoming stamp and delta clamp, before accumulate: paused (held
gatelock grab or i3lock) -> shadow_paused_seconds; else a qualifying game,
a "leisure" now.json, or an untrusted/missing now.json (fail closed) bills
shadow_seconds once, split over shadow_per_source. shadow_budget_seconds is
earned_time.resolve(leisure=True) from the same answers, held as a same-day
high-water mark. Not billed under a total block. Shadow verdict changes go
to budget.jsonl as shadow_change (the live game view skips them); /api/budget
and MCP get_gaming_time carry a "shadow" block. Enforced fields, cutoff,
warnings and existing log records are unchanged.

Also fixes two tests that assumed earned_time < 0.14 internals.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N
This commit is contained in:
Krzysztof Rudnicki 2026-10-19 12:37:01 +02:00
parent 5e1fc16ea6
commit e82ddd9a4b
45 changed files with 958 additions and 1267 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -111,6 +111,7 @@ repos:
additional_dependencies:
- types-requests
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time
- leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.0
# ===========================================================================
# PYLINT
@ -131,6 +132,7 @@ repos:
additional_dependencies:
- aiohttp==3.14.5
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time
- leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.0
- howlongtobeatpy==1.0.24
- mcp==2.3.0
- pytest

4
pyproject.toml
View File

@ -17,6 +17,10 @@ dependencies = [
# 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",
# 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.
"leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.0",
]
# Optional extras. The `mcp` group is ONLY needed by the MCP server

1
requirements.txt
View File

@ -7,6 +7,7 @@ codespell==2.4.3
coverage==7.16.2
earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.14.0#subdirectory=earned_time
howlongtobeatpy==1.0.24
leisure-watch @ git+https://github.com/kuhyx/leisure-watch@leisure-watch-v0.1.0
mcp==2.3.0
mypy==2.4.0
pre-commit==4.7.0

7
steam_backlog_enforcer/_attribution_labels.py
View File

@ -16,8 +16,9 @@ import contextlib
import re
from typing import TYPE_CHECKING, Final
from steam_backlog_enforcer import _steam_state
from steam_backlog_enforcer._counted_procs import labels_by_key, load_counted_processes
from leisure_watch.procscan import labels_by_key, load_counted_processes
from steam_backlog_enforcer import _steam_state, config
if TYPE_CHECKING:
from collections.abc import Iterable
@ -74,7 +75,7 @@ def label_for(key: str) -> str:
return _steam_name(int(raw)) or key
return key
if key.startswith(_PROC_PREFIX):
return labels_by_key(load_counted_processes()).get(key, key)
return labels_by_key(load_counted_processes(config.CONFIG_FILE)).get(key, key)
if key.startswith(_LAUNCHER_PREFIX):
return key[len(_LAUNCHER_PREFIX) :]
return key

8
steam_backlog_enforcer/_budget_log_tail.py
View File

@ -23,17 +23,20 @@ import json
import logging
from typing import Any, Final
from leisure_watch.procscan import process_name
from steam_backlog_enforcer._playtime_log import (
EVENT_MANUAL_ADJUSTMENT,
budget_log_path,
)
from steam_backlog_enforcer._playtime_procs import process_name
from steam_backlog_enforcer._shadow_log import EVENT_SHADOW_CHANGE
logger = logging.getLogger(__name__)
# Enough for several records at any plausible size, small enough to read on
# every poll without thinking about it.
_TAIL_BYTES: Final = 64 * 1024
_NOT_VERDICTS: Final = frozenset({EVENT_MANUAL_ADJUSTMENT, EVENT_SHADOW_CHANGE})
@dataclass(frozen=True)
@ -85,7 +88,8 @@ def _last_record(*, demo: bool) -> dict[str, Any] | None:
# A manual adjustment (e.g. a gaming reset) is an audit note, not a
# tick verdict: returning it would blank the live status until the
# next verdict change or heartbeat.
if isinstance(record, dict) and record.get("event") != EVENT_MANUAL_ADJUSTMENT:
# A shadow verdict (C4) is leisure, not a game tick: same reasoning.
if isinstance(record, dict) and record.get("event") not in _NOT_VERDICTS:
return record
return None

31
steam_backlog_enforcer/_budget_resolve.py
View File

@ -58,6 +58,8 @@ from steam_backlog_enforcer._reading_bonus import read_today
from steam_backlog_enforcer._workout_budget import workout_logged_today
if TYPE_CHECKING:
from datetime import date
from steam_backlog_enforcer.config import Config
logger = logging.getLogger(__name__)
@ -101,6 +103,9 @@ class BudgetResolution:
reading_seconds: Seconds added by a credited reading session, or 0.
earned_seconds: Seconds added per earner name, every registered earner
included -- the generic form of the three fields above.
shadow_seconds: The same answers priced as leisure
(``earned_time.resolve(leisure=True)``, contract C4). Shown, never
enforced.
"""
seconds: float
@ -110,6 +115,23 @@ class BudgetResolution:
reason: str
reading_seconds: float = 0.0
earned_seconds: dict[str, float] = field(default_factory=dict)
shadow_seconds: float = 0.0
def _price(
answers: dict[str, int | bool | None],
today: date,
registry: tuple[earned_time.Earner, ...],
first_paid: dict[str, date | None] | None,
*,
leisure: bool,
) -> earned_time.Resolution:
"""Price one set of answers; ``leisure=False`` is exactly the old call."""
if first_paid is None:
return earned_time.resolve(answers, today, registry, leisure=leisure)
return earned_time.resolve(
answers, today, registry, first_credits=first_paid, leisure=leisure
)
def _describe(name: str, label: str, answer: int | None) -> str:
@ -164,11 +186,9 @@ def resolve_budget(config: Config) -> BudgetResolution:
answers = _answers(config, registry)
# Maturity only delays a penalty: a gate that never paid out costs nothing.
first_paid = first_credits(registry, today)
day = (
earned_time.resolve(answers, today, registry)
if first_paid is None
else earned_time.resolve(answers, today, registry, first_credits=first_paid)
)
day = _price(answers, today, registry, first_paid, leisure=False)
# The same answers, asked once: the shadow can never see a different day.
shadow = _price(answers, today, registry, first_paid, leisure=True)
earned = {t.earner.name: t.gaming_minutes * _SECONDS_PER_MINUTE for t in day.terms}
total = day.gaming_minutes * _SECONDS_PER_MINUTE
parts = ", ".join(
@ -188,4 +208,5 @@ def resolve_budget(config: Config) -> BudgetResolution:
reason=reason,
reading_seconds=earned.get("reading", 0.0),
earned_seconds=earned,
shadow_seconds=shadow.gaming_minutes * _SECONDS_PER_MINUTE,
)

2
steam_backlog_enforcer/_budget_view.py
View File

@ -29,6 +29,7 @@ from steam_backlog_enforcer._playtime_state import (
rules_for,
state_path,
)
from steam_backlog_enforcer._shadow_view import shadow_view
from steam_backlog_enforcer.config import Config, State
if TYPE_CHECKING:
@ -136,6 +137,7 @@ def build_today(state: PlaytimeState, rules: PlaytimeRules) -> dict[str, Any]:
"next_warning_seconds": next_warning(state, rules),
"warned_seconds": list(state.warned_seconds),
"games": today_games(state),
"shadow": shadow_view(state),
}

245
steam_backlog_enforcer/_counted_procs.py
View File

@ -1,245 +0,0 @@
"""User-declared non-Steam processes that count against the gaming budget.
Steam games announce themselves through ``SteamAppId``; everything else has to
be named. :mod:`_total_block_launchers` already names sixteen *launchers*, but
that frozenset is consumed by the total-block killer, so adding a game to it
silently makes the game killable. This module is the separate, user-editable
list: ``counted_processes`` in ``config.json``.
Loading is deliberately independent of :class:`~config.Config`. The two kill
paths that need these names — ``_playtime_kill.steam_and_launcher_pids`` and
``_total_block.enforce_total_block_tick`` — take no arguments and never build a
``Config``, so a ``Config``-only accessor could not reach them.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import logging
from typing import Any
from steam_backlog_enforcer import config
logger = logging.getLogger(__name__)
# Names that must never be matched. Every name here is either an interpreter
# that hosts unrelated programs or a shell: `counted_processes` entries are
# kill targets, and `sh` in the set would SIGTERM every shell on the system.
# osu!lazer is launched through a `/bin/sh` wrapper, so this is a live hazard,
# not a hypothetical one.
FORBIDDEN_NAMES = frozenset(
{
"bash",
"dash",
"dotnet",
"env",
"java",
"mono",
"node",
"perl",
"python",
"python3",
"ruby",
"sh",
"wine",
"wine64",
"zsh",
}
)
@dataclass(frozen=True)
class CountedProcess:
"""One user-declared non-Steam game that bills the budget."""
id: str
"""Stable slug; forms the ``proc:<id>`` attribution key."""
label: str
"""Display name shown in the UI and stored in history labels."""
names: frozenset[str]
"""Process basenames to match against ``comm`` and argv."""
# osu!lazer ships as the AUR `osu-lazer-bin` package: `/usr/bin/osu-lazer` is a
# `/bin/sh` wrapper that execs `/opt/osu-lazer/osu.AppImage`, which self-mounts
# a squashfs and execs the game itself. All three layers are named so that
# whichever one the detector sees first, the time is billed.
DEFAULT_COUNTED_PROCESSES: tuple[CountedProcess, ...] = (
CountedProcess(
id="osu-lazer",
label="osu!lazer",
names=frozenset({"osu-lazer", "osu.AppImage", "osu!"}),
),
)
def parse_counted_processes(raw: object) -> tuple[CountedProcess, ...]:
"""Build validated entries from ``config.json``'s ``counted_processes``.
Invalid entries are dropped with a warning rather than raising: a typo in a
hand-edited config must not take the whole enforcer down, and the failure
mode of dropping one entry (that game stops billing) is visible in the UI.
Args:
raw: The deserialised ``counted_processes`` value.
Returns:
The valid entries, in file order.
"""
if not isinstance(raw, list):
logger.warning("counted_processes is not a list; ignoring it.")
return ()
entries: list[CountedProcess] = []
seen: set[str] = set()
for item in raw:
entry = _parse_entry(item)
if entry is None or entry.id in seen:
continue
seen.add(entry.id)
entries.append(entry)
return tuple(entries)
def _parse_entry(item: object) -> CountedProcess | None:
"""Validate one ``counted_processes`` element.
Args:
item: A single deserialised entry.
Returns:
The entry, or ``None`` if it is unusable.
"""
if not isinstance(item, dict):
logger.warning("counted_processes entry is not an object; skipping.")
return None
entry_id = _text(item.get("id"))
if not entry_id:
logger.warning("counted_processes entry has no id; skipping.")
return None
names = _clean_names(entry_id, item.get("names"))
if not names:
logger.warning("counted_processes entry %r has no usable names.", entry_id)
return None
return CountedProcess(
id=entry_id,
label=_text(item.get("label")) or entry_id,
names=names,
)
def _clean_names(entry_id: str, raw: object) -> frozenset[str]:
"""Return the matchable names in *raw*, dropping forbidden ones.
Args:
entry_id: Owning entry id, for the warning message.
raw: The deserialised ``names`` value.
Returns:
The usable names.
"""
if not isinstance(raw, list):
return frozenset()
names: set[str] = set()
for value in raw:
name = _text(value)
if not name:
continue
if name in FORBIDDEN_NAMES:
logger.warning(
"counted_processes entry %r names %r, which hosts unrelated "
"programs and is a kill target; refusing it.",
entry_id,
name,
)
continue
names.add(name)
return frozenset(names)
def _text(value: object) -> str:
"""Return *value* as a stripped string, or ``""`` if it is not one.
Args:
value: Any deserialised JSON value.
Returns:
The trimmed string.
"""
return value.strip() if isinstance(value, str) else ""
def load_counted_processes() -> tuple[CountedProcess, ...]:
"""Read ``counted_processes`` straight from ``config.json``.
Used by the kill paths, which have no ``Config`` to consult. A missing or
unreadable file yields the defaults, so a fresh install still bills
osu!lazer.
Returns:
The configured entries, or :data:`DEFAULT_COUNTED_PROCESSES`.
"""
try:
data: Any = json.loads(config.CONFIG_FILE.read_text(encoding="utf-8"))
except OSError, ValueError:
return DEFAULT_COUNTED_PROCESSES
if not isinstance(data, dict) or "counted_processes" not in data:
return DEFAULT_COUNTED_PROCESSES
return parse_counted_processes(data["counted_processes"])
def counted_process_names(entries: tuple[CountedProcess, ...]) -> frozenset[str]:
"""Return every matchable name across *entries*.
Args:
entries: Configured entries.
Returns:
The union of their names.
"""
if not entries:
return frozenset()
return frozenset().union(*(entry.names for entry in entries))
def key_by_name(entries: tuple[CountedProcess, ...]) -> dict[str, str]:
"""Map each matchable name to its ``proc:<id>`` attribution key.
Args:
entries: Configured entries.
Returns:
Mapping of process name to attribution key.
"""
return {name: f"proc:{entry.id}" for entry in entries for name in entry.names}
def labels_by_key(entries: tuple[CountedProcess, ...]) -> dict[str, str]:
"""Map each ``proc:<id>`` key to its display label.
Args:
entries: Configured entries.
Returns:
Mapping of attribution key to label.
"""
return {f"proc:{entry.id}": entry.label for entry in entries}
def kill_target_names() -> frozenset[str]:
"""Return the counted-process names the kill paths should also target.
The budget cutoff and the total block kill these; the
``kill_unauthorized_games`` path does not — it matches on ``SteamAppId``,
which a non-Steam game never sets, so it excludes them for free.
Returns:
Every configured matchable name.
"""
return counted_process_names(load_counted_processes())

17
steam_backlog_enforcer/_gaming_days.py
View File

@ -59,3 +59,20 @@ def held_today(stored: PlaytimeState | None, now: datetime) -> float:
if stored is None or stored.day_key != gaming_day_key(now):
return 0.0
return stored.budget_seconds
def held_shadow_today(stored: PlaytimeState | None, now: datetime) -> float:
"""Return the shadow leisure budget already granted in *now*'s gaming day.
The same high-water rule as :func:`held_today`, for ``shadow_budget_seconds``.
Args:
stored: The persisted production state, or ``None`` if unreadable.
now: Local timestamp for this tick.
Returns:
The recorded shadow high-water budget; 0 when the state is another day's.
"""
if stored is None or stored.day_key != gaming_day_key(now):
return 0.0
return stored.shadow_budget_seconds

2
steam_backlog_enforcer/_mcp_gaming_time.py
View File

@ -66,6 +66,8 @@ def get_gaming_time() -> dict[str, Any]:
# Which games make up seconds_used, largest first, with whatever the
# focus probe could not attribute as a trailing "unattributed" slice.
"games": today["games"],
# The shadow leisure budget (C4): shown, never enforced.
"shadow": today["shadow"],
}

27
steam_backlog_enforcer/_playtime.py
View File

@ -20,9 +20,12 @@ suspends, missed ticks, daemon restarts and NTP steps all behave.
**Launchers are matched by cmdline, not by ``comm``.** ``/usr/bin/lutris`` is a
``#!/usr/bin/env python3`` script, so the kernel records its ``comm`` as
``python3`` and the ``comm``-based scan in :mod:`enforcer` never sees it. That
is why :func:`get_pids_by_cmdline_names` exists alongside
``enforcer.get_pids_by_process_names`` rather than replacing it.
``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.
"""
from __future__ import annotations
@ -33,6 +36,9 @@ import logging
import time
from typing import TYPE_CHECKING
from leisure_watch.procscan import attributed_key, 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._playtime_block import mounted_targets
from steam_backlog_enforcer._playtime_budget import (
@ -45,7 +51,6 @@ from steam_backlog_enforcer._playtime_cutoff import (
_warn,
reconcile_for,
)
from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pids
from steam_backlog_enforcer._playtime_state import (
PlaytimeRules,
PlaytimeState,
@ -53,6 +58,7 @@ from steam_backlog_enforcer._playtime_state import (
rules_for,
save_state,
)
from steam_backlog_enforcer._shadow import observe_shadow, shadow_accumulate
from steam_backlog_enforcer._total_block import is_total_block_active
if TYPE_CHECKING:
@ -96,7 +102,11 @@ def playtime_tick(
# because earners can reset at midnight -- before the gaming day ends --
# and a 05:59 dip must not erase the evening's earned-but-unspent time.
state = replace(
state, budget_seconds=max(state.budget_seconds, rules.budget_seconds)
state,
budget_seconds=max(state.budget_seconds, rules.budget_seconds),
shadow_budget_seconds=max(
state.shadow_budget_seconds, rules.shadow_budget_seconds
),
)
if is_total_block_active():
@ -108,9 +118,13 @@ def playtime_tick(
session.history.observe(state, demo=demo)
return
owners = qualifying_pids(rules)
owners = qualifying_pids(rules, config_mod.CONFIG_FILE)
pids = set(owners)
shadow = observe_shadow(owners, stamp=now.timestamp())
state = shadow_accumulate(
state, stamp=now.timestamp(), interval=interval, verdict=shadow
)
state = accumulate(
state,
now=now,
@ -120,6 +134,7 @@ def playtime_tick(
)
state = _policy(state, rules, now=now)
session.journal.observe(pids, state, rules=rules, now_monotonic=time.monotonic())
session.shadow.observe(shadow, state)
save_state(state, demo=demo)
session.history.observe(state, demo=demo)

8
steam_backlog_enforcer/_playtime_cutoff.py
View File

@ -12,6 +12,9 @@ from dataclasses import replace
import logging
from typing import TYPE_CHECKING
from leisure_watch.procscan import qualifying_pids
from steam_backlog_enforcer import config as config_mod
from steam_backlog_enforcer._playtime_block import (
reconcile,
)
@ -22,7 +25,6 @@ from steam_backlog_enforcer._playtime_kill import (
steam_and_launcher_pids,
)
from steam_backlog_enforcer._playtime_notify import _humanise, notify_desktop_user
from steam_backlog_enforcer._playtime_procs import qualifying_pids
from steam_backlog_enforcer._playtime_state import (
_SHUTDOWN_GRACE_SECONDS,
PlaytimeRules,
@ -148,4 +150,6 @@ def _kill_set(rules: PlaytimeRules) -> set[int]:
Returns:
PIDs to signal.
"""
return set(qualifying_pids(rules)) | steam_and_launcher_pids()
return (
set(qualifying_pids(rules, config_mod.CONFIG_FILE)) | steam_and_launcher_pids()
)

16
steam_backlog_enforcer/_playtime_kill.py
View File

@ -15,13 +15,15 @@ from pathlib import Path
import shutil
import signal
from steam_backlog_enforcer._counted_procs import kill_target_names
from steam_backlog_enforcer._playtime_run import _run
from steam_backlog_enforcer._total_block_launchers import (
from leisure_watch.procscan import (
LAUNCHER_PROCESS_NAMES,
STEAM_CLIENT_PROCESS_NAMES,
get_pids_by_process_names,
kill_target_names,
)
from steam_backlog_enforcer.enforcer import get_pids_by_process_names
from steam_backlog_enforcer import config
from steam_backlog_enforcer._playtime_run import _run
logger = logging.getLogger(__name__)
@ -54,7 +56,11 @@ def steam_and_launcher_pids() -> set[int]:
Returns:
The launcher and client PIDs.
"""
names = STEAM_CLIENT_PROCESS_NAMES | LAUNCHER_PROCESS_NAMES | kill_target_names()
names = (
STEAM_CLIENT_PROCESS_NAMES
| LAUNCHER_PROCESS_NAMES
| kill_target_names(config.CONFIG_FILE)
)
return set(get_pids_by_process_names(names))

190
steam_backlog_enforcer/_playtime_procs.py
View File

@ -1,190 +0,0 @@
"""Identifying gaming processes by their /proc cmdline.
Process-name matching alone misses launchers run through an interpreter
(``lutris`` is a Python script; several Minecraft launchers are ``java -jar``),
so this reads argv and looks past the interpreter to the real program.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import TYPE_CHECKING
from steam_backlog_enforcer._counted_procs import key_by_name, load_counted_processes
from steam_backlog_enforcer._total_block_launchers import LAUNCHER_PROCESS_NAMES
from steam_backlog_enforcer.enforcer import (
get_pids_by_process_names,
get_running_steam_game_pids,
)
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeRules
_PROC = Path("/proc")
# argv[0] basenames that tell us nothing: the real program is argv[1].
_INTERPRETERS = frozenset({"bash", "env", "java", "perl", "python", "python3", "sh"})
# An interpreter invocation needs at least `<interp> <script>` to name a program.
_MIN_ARGV_FOR_INTERPRETER = 2
def get_pids_by_cmdline_names(names: frozenset[str]) -> dict[int, str]:
"""Scan ``/proc/*/cmdline`` for processes whose program name is in *names*.
Complements ``enforcer.get_pids_by_process_names``, which matches on
``comm`` and therefore cannot see interpreter-launched programs: the kernel
records ``/usr/bin/lutris`` as ``python3``. When ``argv[0]``'s basename is a
known interpreter this falls through to ``argv[1]``.
The daemon itself runs under ``python3``, so its own PID is skipped.
Args:
names: Program basenames to match.
Returns:
Mapping of PID to the matched name.
"""
own_pid = os.getpid()
found: dict[int, str] = {}
for entry in _PROC.iterdir():
if not entry.name.isdigit():
continue
pid = int(entry.name)
if pid == own_pid:
continue
matched = _match_cmdline(entry, names)
if matched is not None:
found[pid] = matched
return found
def _match_cmdline(entry: Path, names: frozenset[str]) -> str | None:
"""Return the name in *names* that *entry*'s cmdline runs, if any.
Args:
entry: A ``/proc/<pid>`` directory.
names: Program basenames to match.
Returns:
The matched name, or ``None``.
"""
try:
raw = (entry / "cmdline").read_bytes()
except OSError, ValueError:
return None
argv = [
part for part in raw.decode("utf-8", errors="replace").split("\x00") if part
]
if not argv:
return None
first = Path(argv[0]).name
if first in names:
return first
if first not in _INTERPRETERS or len(argv) < _MIN_ARGV_FOR_INTERPRETER:
return None
second = Path(argv[1]).name
return second if second in names else None
def process_name(pid: int) -> str | None:
"""Return *pid*'s program name, or ``None`` if it is no longer running.
Existence is the point as much as the name: callers resolve PIDs recorded
earlier, and a PID that has since been recycled would otherwise be reported
under whatever process now holds the number.
Args:
pid: Process id to look up.
Returns:
The contents of ``/proc/<pid>/comm``, or ``None``.
"""
try:
return (_PROC / str(pid) / "comm").read_text(encoding="utf-8").strip() or None
except OSError, ValueError:
return None
def _merge_named(found: dict[int, str], keys: dict[str, str]) -> None:
"""Add name-matched PIDs to *found* without displacing existing keys.
``setdefault`` is the point: a Steam game already attributed by its
``SteamAppId`` must keep that key even if its ``comm`` also happens to match
a launcher name.
A scanner can only return names it was asked for, so an unmapped name is
unreachable in production — it is skipped rather than raising, because the
alternative is a ``KeyError`` inside the enforce loop.
Args:
found: Accumulating PID-to-key mapping, mutated in place.
keys: Program basename to attribution key.
"""
names = frozenset(keys)
for source in (get_pids_by_process_names(names), get_pids_by_cmdline_names(names)):
for pid, name in source.items():
key = keys.get(name)
if key is not None:
found.setdefault(pid, key)
def qualifying_pids(rules: PlaytimeRules) -> dict[int, str]:
"""Return PIDs whose runtime counts against the daily budget, with owners.
Steam games are identified by the ``SteamAppId`` environment variable, using
the same ``!= 0`` predicate ``enforcer.enforce_allowed_game`` uses to exclude
the Steam client tree — browsing the store is not gaming.
The value is an attribution key (``app:<id>``, ``launcher:<name>`` or
``proc:<id>``) so that the budget can record *which* game it billed. The
app id was previously destructured and dropped on the first line here, one
step before accumulation.
``counted_processes`` is deliberately *not* gated on ``count_launchers``:
that switch covers the sixteen hardcoded launchers, and turning it off must
not silently stop billing a game the user explicitly listed.
Args:
rules: Policy for this tick.
Returns:
Mapping of qualifying PID to its attribution key.
"""
found: dict[int, str] = {
pid: f"app:{app_id}"
for pid, app_id in get_running_steam_game_pids().items()
if app_id != 0
}
if rules.count_launchers:
_merge_named(
found, {name: f"launcher:{name}" for name in LAUNCHER_PROCESS_NAMES}
)
counted = key_by_name(load_counted_processes())
if counted:
_merge_named(found, counted)
return found
def attributed_key(qualifying: dict[int, str]) -> str:
"""Return the single attribution key this tick should be credited to.
With exactly one game qualifying, it gets the credit. With two or more
running at once there is no honest way to say which one earned the tick,
so this returns ``""``: the tick still bills, and the gap shows up as
Unattributed rather than being charged to a guess.
Args:
qualifying: Mapping of qualifying PID to attribution key.
Returns:
The attribution key, or ``""`` if none can be determined.
"""
distinct = set(qualifying.values())
return distinct.pop() if len(distinct) == 1 else ""

72
steam_backlog_enforcer/_playtime_record.py Normal file
View File

@ -0,0 +1,72 @@
"""The persisted per-gaming-day record, split from ``_playtime_state``.
Split at the 250-line cap when the shadow leisure fields (contract C4) were
added. ``_playtime_state`` re-imports :class:`PlaytimeState`, so every
existing ``from _playtime_state import PlaytimeState`` keeps working.
"""
from __future__ import annotations
from dataclasses import dataclass, field
# Unchanged by per-game attribution: `load_state` keeps only keys that are
# dataclass fields, so a record written before `per_game` existed loads with
# the field's empty default. A bump would buy nothing and would route the
# live file through the fail-closed unknown-schema path.
_SCHEMA_VERSION = 1
@dataclass
class PlaytimeState:
"""Per-gaming-day playtime accounting, persisted between ticks."""
schema_version: int = _SCHEMA_VERSION
day_key: str = ""
"""Gaming day this record covers, ``YYYY-MM-DD``, 06:00-shifted."""
seconds: float = 0.0
"""Qualifying seconds accrued during ``day_key``."""
last_tick_at: float = 0.0
"""Epoch seconds of the last accounting tick; ``0.0`` means never ticked."""
blocked_at: float = 0.0
"""Epoch seconds when the cutoff engaged; ``0.0`` means not blocked."""
per_game: dict[str, float] = field(default_factory=dict)
"""Attribution key to seconds credited during ``day_key``.
Only ever a *subset* of ``seconds``: a tick that cannot be attributed to one
game still bills, and ``backdate`` refunds can outpace what one key holds.
The difference is rendered as "Unattributed" rather than stored, so the
invariant is ``sum(per_game.values()) <= seconds`` — never equality.
"""
last_credited_key: str = ""
"""Key credited by the most recent billing tick.
Exists so ``backdate``'s idle-grace refund can debit the same key it
credited; without it every engaged-to-paused edge would drift the parts
above the whole.
"""
warned_seconds: list[int] = field(default_factory=list)
"""Warning thresholds already fired during ``day_key``."""
budget_seconds: float = 0.0
"""Highest budget in force during ``day_key``; ``0.0`` if unset.
What ``held_today`` holds the day at when a live re-resolution dips.
"""
# Shadow leisure accounting (C4): measured, never enforced. No field
# above is ever written by the shadow path, and these never feed a cutoff.
shadow_seconds: float = 0.0
"""Leisure seconds billed during ``day_key`` (wall clock, once per tick)."""
shadow_budget_seconds: float = 0.0
"""Highest ``resolve(leisure=True)`` budget of ``day_key``, in seconds."""
shadow_per_source: dict[str, float] = field(default_factory=dict)
"""Source key to shadow seconds; overlapping sources split a tick."""
shadow_paused_seconds: float = 0.0
"""Seconds a held locker grab or i3lock paused shadow billing."""
def is_blocked(self) -> bool:
"""Whether the cutoff has engaged for this gaming day.
Deliberately a method rather than a field: ``save`` serialises
``self.__dict__``, so a derived field would be persisted and could
drift out of step with ``blocked_at``.
"""
return self.blocked_at > 0.0

14
steam_backlog_enforcer/_playtime_session.py
View File

@ -21,9 +21,11 @@ from steam_backlog_enforcer._playtime_log import (
TickJournal,
budget_log_path,
)
from steam_backlog_enforcer._shadow_log import ShadowJournal
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
from steam_backlog_enforcer._shadow import ShadowVerdict
class TickRecorder(Protocol):
@ -47,12 +49,20 @@ class HistoryRecorder(Protocol):
"""Record today's running total if it has moved far enough."""
class ShadowRecorder(Protocol):
"""What the budget tick needs from the shadow verdict journal."""
def observe(self, verdict: ShadowVerdict, state: PlaytimeState) -> None:
"""Record the shadow verdict if it changed."""
@dataclass(frozen=True)
class PlaytimeSession:
"""Per-daemon logging state."""
journal: TickRecorder
history: HistoryRecorder
shadow: ShadowRecorder
def new_session(*, demo: bool = False) -> PlaytimeSession:
@ -66,7 +76,9 @@ def new_session(*, demo: bool = False) -> PlaytimeSession:
Returns:
A session for this daemon run.
"""
log = BudgetLog(path=budget_log_path(demo=demo))
return PlaytimeSession(
journal=TickJournal(BudgetLog(path=budget_log_path(demo=demo))),
journal=TickJournal(log),
history=HistoryWriter(),
shadow=ShadowJournal(log),
)

74
steam_backlog_enforcer/_playtime_state.py
View File

@ -14,8 +14,9 @@ import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._budget_resolve import resolve_budget
from steam_backlog_enforcer._gaming_days import held_today
from steam_backlog_enforcer._gaming_days import held_shadow_today, held_today
from steam_backlog_enforcer._playtime_legacy import without_legacy_carry
from steam_backlog_enforcer._playtime_record import _SCHEMA_VERSION, PlaytimeState
from steam_backlog_enforcer._whitelist_locking import (
_try_set_immutable,
unlock_for_write,
@ -27,17 +28,25 @@ if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config
# Explicit because PlaytimeState now lives in _playtime_record: every existing
# `from _playtime_state import PlaytimeState` keeps type-checking as a re-export.
__all__ = [
"PLAYTIME_DEMO_STATE_FILE",
"PLAYTIME_STATE_FILE",
"STATE_MODE",
"PlaytimeRules",
"PlaytimeState",
"load_state",
"rules_for",
"save_state",
"state_path",
]
logger = logging.getLogger(__name__)
PLAYTIME_STATE_FILE = CONFIG_DIR / "playtime_state.json"
PLAYTIME_DEMO_STATE_FILE = CONFIG_DIR / "playtime_demo_state.json"
# Unchanged by per-game attribution: `load_state` keeps only keys that are
# dataclass fields, so a record written before `per_game` existed loads with
# the field's empty default. A bump would buy nothing and would route the
# live file through the fail-closed unknown-schema path.
_SCHEMA_VERSION = 1
# The daemon runs as root, and ``mkstemp`` would leave this at 0600 — which
# silently broke every unprivileged reader: ``load_state`` swallows the
# resulting ``PermissionError`` and reports "no state recorded yet", so both the
@ -65,52 +74,6 @@ _SIGKILL_AFTER_SECONDS = 30.0
_DEMO_SIGKILL_AFTER_SECONDS = 10.0
@dataclass
class PlaytimeState:
"""Per-gaming-day playtime accounting, persisted between ticks."""
schema_version: int = _SCHEMA_VERSION
day_key: str = ""
"""Gaming day this record covers, ``YYYY-MM-DD``, 06:00-shifted."""
seconds: float = 0.0
"""Qualifying seconds accrued during ``day_key``."""
last_tick_at: float = 0.0
"""Epoch seconds of the last accounting tick; ``0.0`` means never ticked."""
blocked_at: float = 0.0
"""Epoch seconds when the cutoff engaged; ``0.0`` means not blocked."""
per_game: dict[str, float] = field(default_factory=dict)
"""Attribution key to seconds credited during ``day_key``.
Only ever a *subset* of ``seconds``: a tick that cannot be attributed to one
game still bills, and ``backdate`` refunds can outpace what one key holds.
The difference is rendered as "Unattributed" rather than stored, so the
invariant is ``sum(per_game.values()) <= seconds`` — never equality.
"""
last_credited_key: str = ""
"""Key credited by the most recent billing tick.
Exists so ``backdate``'s idle-grace refund can debit the same key it
credited; without it every engaged-to-paused edge would drift the parts
above the whole.
"""
warned_seconds: list[int] = field(default_factory=list)
"""Warning thresholds already fired during ``day_key``."""
budget_seconds: float = 0.0
"""Highest budget in force during ``day_key``; ``0.0`` if unset.
What ``held_today`` holds the day at when a live re-resolution dips.
"""
def is_blocked(self) -> bool:
"""Whether the cutoff has engaged for this gaming day.
Deliberately a method rather than a field: ``save`` serialises
``self.__dict__``, so a derived field would be persisted and could
drift out of step with ``blocked_at``.
"""
return self.blocked_at > 0.0
@dataclass(frozen=True)
class PlaytimeRules:
"""Resolved budget policy for one tick.
@ -132,6 +95,8 @@ class PlaytimeRules:
reading_seconds: float = 0.0
earned_seconds: dict[str, float] = field(default_factory=dict)
budget_reason: str = ""
# The shadow leisure budget (C4), held like budget_seconds; never enforced.
shadow_budget_seconds: float = 0.0
def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
@ -176,6 +141,9 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
if resolved is None
else resolved.reason + (held_note if held > live else "")
),
shadow_budget_seconds=0.0
if resolved is None
else max(resolved.shadow_seconds, held_shadow_today(stored, now)),
)

141
steam_backlog_enforcer/_shadow.py Normal file
View File

@ -0,0 +1,141 @@
"""Shadow leisure billing (contract C4): measured every tick, never enforced.
Phase 1 of the leisure rollout counts every non-productive minute on the PC
beside the enforced gaming count, so a week of real data exists before the
leisure budget is switched on. Nothing here feeds a warning, a cutoff or a
kill, and nothing here writes an enforced field (``seconds``, ``per_game``,
``last_tick_at``, ...): the shadow is computed from the incoming tick stamp
*before* :func:`~steam_backlog_enforcer._playtime_budget.accumulate` moves it.
Per tick, in order:
1. A locker whose X grab is actually held, or i3lock -> paused: the delta
goes to ``shadow_paused_seconds`` and nothing bills, games included (Q7).
2. Any qualifying game process (the enforcer's own scan) -> leisure, billed
to the game keys.
3. Else the watcher's signed ``now.json``: ``leisure`` bills its reason keys;
a missing, stale or badly signed file fails closed (Q8) and bills
``failclosed:watcher``.
Overlapping sources bill the wall clock once and split it evenly, so
``sum(shadow_per_source) == shadow_seconds`` up to float rounding.
"""
from __future__ import annotations
from dataclasses import dataclass, replace
from pathlib import Path
from typing import TYPE_CHECKING, Final
from leisure_watch.facts import read_now
from leisure_watch.lockstate import paused_by
from steam_backlog_enforcer._playtime_state import _DELTA_CLAMP_FACTOR
if TYPE_CHECKING:
from leisure_watch.facts import Now
from steam_backlog_enforcer._playtime_state import PlaytimeState
# Explicit, because the daemon is root: these are the desktop user's (uid 1000).
NOW_PATH: Final = Path("/run/user/1000/leisure-watch/now.json")
CLAIMS_DIR: Final = Path("/run/user/1000/gatelock/claims")
# now.json older than this is not trusted (contract C3/Q8).
_MAX_AGE_SECONDS: Final = 15
FAILCLOSED_KEY: Final = "failclosed:watcher"
@dataclass(frozen=True)
class ShadowVerdict:
"""What one tick means for the shadow budget."""
paused_by: str | None
"""``gatelock:<app>`` / ``i3lock`` while paused, else ``None``."""
keys: tuple[str, ...]
"""Sources billed this tick; empty = free (or paused)."""
@property
def state(self) -> str:
"""``paused``, ``leisure`` or ``free``."""
if self.paused_by is not None:
return "paused"
return "leisure" if self.keys else "free"
def shadow_keys(owners: dict[int, str], fact: Now | None) -> tuple[str, ...]:
"""Return the sources this unpaused tick bills, or ``()`` when it is free.
Args:
owners: Qualifying game PIDs to attribution keys (``app:<id>`` ...).
fact: The verified watcher fact, or ``None`` if it cannot be trusted.
Returns:
Game keys (sorted) if a game runs; else the watcher's reason keys in
its order; ``(FAILCLOSED_KEY,)`` when the watcher cannot be trusted.
"""
if owners:
return tuple(sorted(set(owners.values())))
if fact is None:
return (FAILCLOSED_KEY,)
if fact.state != "leisure":
return ()
return tuple(dict.fromkeys(r.key for r in fact.reasons)) or (FAILCLOSED_KEY,)
def observe_shadow(owners: dict[int, str], *, stamp: float) -> ShadowVerdict:
"""Read the lock state and (unless paused) ``now.json`` once for this tick.
Args:
owners: This tick's qualifying game PIDs and their keys.
stamp: Epoch seconds of this tick, the freshness reference.
Returns:
The tick's shadow verdict.
"""
paused = paused_by(CLAIMS_DIR)
if paused is not None:
return ShadowVerdict(paused_by=paused, keys=())
fact = read_now(NOW_PATH, stamp, max_age=_MAX_AGE_SECONDS)
return ShadowVerdict(paused_by=None, keys=shadow_keys(owners, fact))
def shadow_accumulate(
state: PlaytimeState,
*,
stamp: float,
interval: float,
verdict: ShadowVerdict,
) -> PlaytimeState:
"""Add this tick's shadow time to *state*; pure, enforced fields untouched.
Measured from the same incoming ``last_tick_at`` and clamped by the same
factor as the enforced ``accumulate``, so a first tick, a backward clock
step or a suspend behave identically in both counts.
Args:
state: Accounting state *before* this tick's enforced accumulation.
stamp: Epoch seconds of this tick.
interval: Nominal seconds between ticks.
verdict: This tick's shadow verdict.
Returns:
The updated state.
"""
delta = stamp - state.last_tick_at
if state.last_tick_at <= 0.0 or delta <= 0.0:
return state
credited = min(delta, interval * _DELTA_CLAMP_FACTOR)
if verdict.paused_by is not None:
paused = state.shadow_paused_seconds + credited
return replace(state, shadow_paused_seconds=paused)
if not verdict.keys:
return state
share = credited / len(verdict.keys)
per_source = dict(state.shadow_per_source)
for key in verdict.keys:
per_source[key] = per_source.get(key, 0.0) + share
return replace(
state,
shadow_seconds=state.shadow_seconds + credited,
shadow_per_source=per_source,
)

58
steam_backlog_enforcer/_shadow_log.py Normal file
View File

@ -0,0 +1,58 @@
"""The shadow verdict in the budget audit log: one record per change.
The game journal (:class:`~steam_backlog_enforcer._playtime_log.TickJournal`)
only writes when the *game* verdict changes, so browsing with no game running
would never show up there. Shadow changes get their own event in the same
``budget.jsonl``, written only when the verdict (paused/leisure/free plus the
billed sources) changes -- no heartbeat; the state file has the running total.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_log import BudgetLog
from steam_backlog_enforcer._playtime_state import PlaytimeState
from steam_backlog_enforcer._shadow import ShadowVerdict
EVENT_SHADOW_CHANGE: Final = "shadow_change"
class ShadowJournal:
"""Records the shadow verdict whenever it changes."""
def __init__(self, log: BudgetLog) -> None:
"""Start with nothing observed.
Args:
log: Sink to write through (the game journal's own trail).
"""
self._log = log
self._last: ShadowVerdict | None = None
def observe(self, verdict: ShadowVerdict, state: PlaytimeState) -> None:
"""Record *verdict* with the shadow totals if it differs from the last.
Args:
verdict: This tick's shadow verdict.
state: Accounting state after this tick.
"""
if verdict == self._last:
return
self._last = verdict
remaining = state.shadow_budget_seconds - state.shadow_seconds
self._log.record(
EVENT_SHADOW_CHANGE,
shadow_state=verdict.state,
paused_by=verdict.paused_by,
sources=list(verdict.keys),
day_key=state.day_key,
shadow_seconds=round(state.shadow_seconds, 3),
shadow_budget_seconds=round(state.shadow_budget_seconds, 3),
shadow_remaining_seconds=round(remaining, 3),
shadow_paused_seconds=round(state.shadow_paused_seconds, 3),
shadow_per_source={
k: round(v, 3) for k, v in state.shadow_per_source.items()
},
)

33
steam_backlog_enforcer/_shadow_view.py Normal file
View File

@ -0,0 +1,33 @@
"""The shadow leisure block (contract C4) for ``/api/budget`` and MCP.
Read-only and display-only: what the leisure budget *would* say, beside the
enforced gaming numbers it never changes.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeState
def shadow_view(state: PlaytimeState) -> dict[str, Any]:
"""Compose the shadow block from stored state.
Args:
state: Loaded accounting state.
Returns:
Used/budget/remaining seconds, whether the leisure budget would block
now, paused seconds, and the per-source split.
"""
remaining = max(0.0, state.shadow_budget_seconds - state.shadow_seconds)
return {
"seconds_used": round(state.shadow_seconds, 1),
"budget_seconds": state.shadow_budget_seconds,
"seconds_remaining": round(remaining, 1),
"would_block": remaining <= 0.0,
"paused_seconds": round(state.shadow_paused_seconds, 1),
"per_source": {k: round(v, 1) for k, v in state.shadow_per_source.items()},
}

3
steam_backlog_enforcer/_steam_restart_guard.py
View File

@ -14,13 +14,14 @@ from __future__ import annotations
import logging
from pathlib import Path
from leisure_watch.procscan import get_running_steam_game_pids
from steam_backlog_enforcer._steam_errors import (
FastInstallInProgressError,
GameInProgressError,
SteamUpdateInProgressError,
)
from steam_backlog_enforcer._steam_state import steam_update_in_progress
from steam_backlog_enforcer.enforcer import get_running_steam_game_pids
logger = logging.getLogger(__name__)

12
steam_backlog_enforcer/_total_block.py
View File

@ -24,7 +24,9 @@ import logging
import shutil
import subprocess
from steam_backlog_enforcer._counted_procs import kill_target_names
from leisure_watch.procscan import LAUNCHER_PROCESS_NAMES, kill_target_names
from steam_backlog_enforcer import config
from steam_backlog_enforcer._total_block_hosts import (
apply_total_block_hosts,
remove_total_block_hosts,
@ -33,7 +35,6 @@ from steam_backlog_enforcer._total_block_iptables import (
apply_total_block_iptables,
remove_total_block_iptables,
)
from steam_backlog_enforcer._total_block_launchers import LAUNCHER_PROCESS_NAMES
from steam_backlog_enforcer._total_block_purge import (
STEAM_PACKAGE,
is_steam_installed,
@ -41,13 +42,12 @@ from steam_backlog_enforcer._total_block_purge import (
purge_steam_and_proton,
uninstall_steam_package,
)
from steam_backlog_enforcer.config import CONFIG_DIR
from steam_backlog_enforcer.enforcer import kill_processes_by_name
from steam_backlog_enforcer.store_blocker import flush_dns_cache
logger = logging.getLogger(__name__)
TOTAL_BLOCK_LOCK_FILE = CONFIG_DIR / "total_block_lock.json"
TOTAL_BLOCK_LOCK_FILE = config.CONFIG_DIR / "total_block_lock.json"
_PACKAGE_BLOCK_NAME = "steam-block"
@ -181,7 +181,7 @@ def start_total_block(days: int) -> bool:
killed = kill_steam_and_launchers(LAUNCHER_PROCESS_NAMES)
# Kill only: kill_steam_and_launchers() uninstalls what it is given, and a
# counted process is a game the user chose to bill, not a launcher to purge.
killed += kill_processes_by_name(kill_target_names())
killed += kill_processes_by_name(kill_target_names(config.CONFIG_FILE))
if killed:
logger.info("Total block: killed %d process(es): %s", len(killed), killed)
@ -209,7 +209,7 @@ def enforce_total_block_tick() -> None:
is True.
"""
kill_steam_and_launchers(LAUNCHER_PROCESS_NAMES)
kill_processes_by_name(kill_target_names())
kill_processes_by_name(kill_target_names(config.CONFIG_FILE))
if is_steam_installed():
logger.warning("Steam reappeared during total block - removing again")

45
steam_backlog_enforcer/_total_block_launchers.py
View File

@ -1,45 +0,0 @@
"""Process-name tables for the game launchers a total block kills.
A data table, kept apart from the killing mechanism in
:mod:`steam_backlog_enforcer._total_block_purge` so that adding a launcher
is a one-line data change.
"""
from __future__ import annotations
STEAM_CLIENT_PROCESS_NAMES = frozenset({"steam", "steamwebhelper", "steam.sh"})
# Third-party game launchers, best-effort match by process name.
LAUNCHER_PROCESS_NAMES = frozenset(
{
"EpicGamesLauncher",
"legendary",
"lutris",
"heroic",
"GalaxyClient",
"itch",
"bottles",
"minecraft-launcher",
"prismlauncher",
"multimc",
"polymc",
"ATLauncher",
"GDLauncher",
"gdlauncher-carbon",
"TLauncher",
"modrinth-app",
}
)
# Known limitation, not engineered around in this pass: any launcher run
# via an interpreter rather than its own compiled binary shows up in
# /proc/*/comm as the INTERPRETER's name, not its own - process-name
# matching won't catch those. Confirmed live for "lutris" (a Python
# script, appears as `python3`), and documented upstream for some
# Minecraft launchers (TLauncher, ATLauncher, GDLauncher - exec'd as
# `java -jar ...`, appear as `java`). Matching the interpreter name
# itself is NOT a fix: kill_processes_by_name runs inside this very
# enforcer process, which is itself `python3` - adding it to the set
# would SIGTERM the daemon and every other Python process on the system.
# Consistent with the "best-effort" framing already agreed for non-Steam
# blocking; the hosts+iptables domain blocking below is the backstop for
# launchers this can't catch by process name.

11
steam_backlog_enforcer/_total_block_purge.py
View File

@ -15,17 +15,18 @@ import logging
from pathlib import Path
import shutil
from leisure_watch.procscan import (
STEAM_CLIENT_PROCESS_NAMES,
get_pids_by_process_names,
)
from steam_backlog_enforcer._pacman import (
is_package_installed,
pacman_owner,
uninstall_package,
)
from steam_backlog_enforcer._total_block_launchers import STEAM_CLIENT_PROCESS_NAMES
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
from steam_backlog_enforcer.enforcer import (
get_pids_by_process_names,
kill_processes_by_name,
)
from steam_backlog_enforcer.enforcer import kill_processes_by_name
logger = logging.getLogger(__name__)

5
steam_backlog_enforcer/config.py
View File

@ -119,8 +119,9 @@ class Config:
Each entry is ``{"id": ..., "label": ..., "names": [...]}``. ``None`` means
"use the built-in defaults" (osu!lazer); an explicit ``[]`` means "none",
which is why this is not simply an empty list. Parsed and validated by
:mod:`steam_backlog_enforcer._counted_procs` — the field exists here only so
that ``save()`` round-trips a hand-edited list instead of dropping it.
``leisure_watch.procscan`` (once ``_counted_procs``) — the field exists
here only so that ``save()`` round-trips a hand-edited list instead of
dropping it.
Deliberately separate from ``LAUNCHER_PROCESS_NAMES``: that frozenset is
also the total-block kill list, so adding a game there has side effects

66
steam_backlog_enforcer/enforcer.py
View File

@ -4,11 +4,17 @@ from __future__ import annotations
import logging
import os
from pathlib import Path
import shutil
import signal
import subprocess
# The /proc scanners live in leisure_watch.procscan (shared with the leisure
# watcher); bound here because the kill/enforce paths below call them.
from leisure_watch.procscan import (
get_pids_by_process_names,
get_running_steam_game_pids,
)
from steam_backlog_enforcer._desktop_env import (
desktop_user_cmd,
resolve_desktop_user,
@ -19,66 +25,12 @@ from steam_backlog_enforcer.game_uninstall import is_protected_app
logger = logging.getLogger(__name__)
def get_running_steam_game_pids() -> dict[int, int]:
"""Scan /proc to find running Steam game processes.
Returns: dict mapping PID -> SteamAppId.
"""
running: dict[int, int] = {}
proc = Path("/proc")
for entry in proc.iterdir():
if not entry.name.isdigit():
continue
try:
environ = (entry / "environ").read_bytes()
pairs = environ.split(b"\x00")
for pair in pairs:
if pair.startswith(b"SteamAppId="):
value = pair.split(b"=", 1)[1].decode("utf-8", errors="replace")
if value.isdigit():
running[int(entry.name)] = int(value)
break
except PermissionError, OSError, ValueError:
continue
return running
def get_pids_by_process_names(names: frozenset[str]) -> dict[int, str]:
"""Scan /proc/*/comm for processes whose command name is in *names*.
The kernel truncates ``/proc/[pid]/comm`` to 15 characters (plus a null
terminator), so *names* longer than that (e.g. ``EpicGamesLauncher``,
``gdlauncher-carbon``) are matched against their own first 15 characters
- matching how the kernel actually stores them, not an exact string
that could never appear in ``comm``.
Returns: dict mapping PID -> matched comm string.
"""
truncated = {name[:15]: name for name in names}
running: dict[int, str] = {}
proc = Path("/proc")
for entry in proc.iterdir():
if not entry.name.isdigit():
continue
try:
comm = (entry / "comm").read_text(encoding="utf-8").strip()
except PermissionError, OSError, ValueError:
continue
if comm in truncated:
running[int(entry.name)] = truncated[comm]
return running
def kill_processes_by_name(names: frozenset[str]) -> list[tuple[int, str]]:
"""Kill (SIGTERM) every running process whose name is in *names*.
Matching is by ``/proc/[pid]/comm`` via :func:`get_pids_by_process_names`,
Matching is by ``/proc/[pid]/comm`` via ``get_pids_by_process_names``,
not the ``SteamAppId`` environment variable (unlike
:func:`get_running_steam_game_pids`) - this is for non-Steam processes
``get_running_steam_game_pids``) - this is for non-Steam processes
(the Steam client itself, and third-party game launchers) that don't set
that variable.

19
steam_backlog_enforcer/tests/_fake_session.py
View File

@ -17,6 +17,7 @@ from steam_backlog_enforcer._playtime_session import PlaytimeSession
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
from steam_backlog_enforcer._shadow import ShadowVerdict
class StubJournal:
@ -63,6 +64,23 @@ class StubHistory:
self.recorded.append((state.day_key, state.seconds))
class StubShadow:
"""A shadow journal that keeps verdicts in memory instead of on disk."""
def __init__(self) -> None:
self.observed: list[ShadowVerdict] = []
def observe(self, verdict: ShadowVerdict, state: PlaytimeState) -> None:
"""Remember *verdict*.
Args:
verdict: This tick's shadow verdict.
state: Ignored.
"""
del state
self.observed.append(verdict)
def fake_session() -> PlaytimeSession:
"""Build a session that never touches the log file or disk.
@ -72,4 +90,5 @@ def fake_session() -> PlaytimeSession:
return PlaytimeSession(
journal=StubJournal(),
history=StubHistory(),
shadow=StubShadow(),
)

1
steam_backlog_enforcer/tests/_isolate_playtime.py
View File

@ -66,7 +66,6 @@ def _isolate_playtime(tmp_path: Path) -> Iterator[None]:
"steam_backlog_enforcer._playtime_log.BUDGET_DEMO_LOG_FILE",
tmp_path / "budget-demo.jsonl",
),
patch("steam_backlog_enforcer._playtime_procs._PROC", fake_proc),
patch("steam_backlog_enforcer._playtime_kill._PROC", fake_proc),
# Never let a test chattr +i a file inside tmp_path — pytest could
# then not clean the directory up.

34
steam_backlog_enforcer/tests/_isolate_shadow.py Normal file
View File

@ -0,0 +1,34 @@
"""Autouse isolation of the shared /proc scan and the shadow leisure inputs.
The scanners moved to ``leisure_watch.procscan`` read ``/proc`` through their
own module constants, and the shadow tick reads the desktop user's gatelock
claims, i3lock and the watcher's signed ``now.json``. Unpatched, a test would
see the developer's live games, lockers and browser.
Default: nothing paused and no trusted watcher fact (the fail-closed reading).
"""
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
@pytest.fixture(autouse=True)
def _isolate_shadow(tmp_path: Path) -> Iterator[None]:
"""Point the scan at an empty /proc and the shadow inputs at nothing."""
fake_proc = tmp_path / "procscan_proc"
fake_proc.mkdir(exist_ok=True)
with (
patch("leisure_watch.procscan._steam._PROC", fake_proc),
patch("leisure_watch.procscan._cmdline._PROC", fake_proc),
patch("steam_backlog_enforcer._shadow.paused_by", return_value=None),
patch("steam_backlog_enforcer._shadow.read_now", return_value=None),
):
yield

4
steam_backlog_enforcer/tests/_no_real_effects.py
View File

@ -175,7 +175,7 @@ def _no_real_effects() -> Iterator[list[str]]:
# Live process-table isolation
#
# Steam games are found by scanning the real /proc for SteamAppId
# (enforcer.get_running_steam_game_pids hardcodes the path). Unpatched,
# (leisure_watch.procscan scans it). Unpatched,
# that reads the *developer's* process table: running the suite while
# playing made three restart tests fail on 2026-08-28, and let a demo
# loop bill and kill the live game on 2026-10-10. Tests that care
@ -192,7 +192,7 @@ def _no_live_games() -> Iterator[None]:
return_value={},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
"leisure_watch.procscan._qualifying.get_running_steam_game_pids",
return_value={},
),
):

7
steam_backlog_enforcer/tests/_no_workout_http.py
View File

@ -60,6 +60,12 @@ def earned_budget(_config: Config) -> BudgetResolution:
earned = {e.name: e.gaming_minutes * _MINUTE for e in earned_time.EARNERS}
ceiling = earned_time.GAMING_CEILING_MINUTES * _MINUTE
total = min(ceiling, base * _MINUTE + sum(earned.values()))
# Shadow (C4): the leisure base is the gaming base plus the same lift.
lift = earned_time.LEISURE_BASE_MINUTES - earned_time.GAMING_BASE_MINUTES
shadow = min(
earned_time.LEISURE_CEILING_MINUTES * _MINUTE,
(base + lift) * _MINUTE + sum(earned.values()),
)
return BudgetResolution(
seconds=total,
base_seconds=base * _MINUTE,
@ -68,6 +74,7 @@ def earned_budget(_config: Config) -> BudgetResolution:
reason="stubbed: fully earned",
reading_seconds=earned.get("reading", 0.0),
earned_seconds=earned,
shadow_seconds=shadow,
)

2
steam_backlog_enforcer/tests/conftest.py
View File

@ -28,6 +28,7 @@ from steam_backlog_enforcer.tests._isolate_control_plane import (
from steam_backlog_enforcer.tests._isolate_incidents import _isolate_incidents
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
from steam_backlog_enforcer.tests._no_fast_install import _no_fast_install
from steam_backlog_enforcer.tests._no_real_effects import (
_no_live_games,
@ -47,6 +48,7 @@ __all__ = [
"_isolate_incidents",
"_isolate_playtime",
"_isolate_reading",
"_isolate_shadow",
"_no_fast_install",
"_no_live_games",
"_no_real_effects",

3
steam_backlog_enforcer/tests/test_attribution_labels.py
View File

@ -4,10 +4,11 @@ from __future__ import annotations
from unittest.mock import patch
from leisure_watch.procscan import CountedProcess
from steam_backlog_enforcer import _attribution_labels as labels_mod
from steam_backlog_enforcer import _steam_state
from steam_backlog_enforcer._attribution_labels import label_for, labels_for
from steam_backlog_enforcer._counted_procs import CountedProcess
def _manifest(app_id: int, name: str) -> None:

8
steam_backlog_enforcer/tests/test_budget_resolve_reading.py
View File

@ -115,7 +115,9 @@ class TestTheBase:
with pin_today(day):
resolved = _resolve(earners=(False, False, False))
later_cuts = 60 * sum(e.max_gaming_minutes for e in _later_penalised(day))
assert resolved.base_seconds == 4 * _HOUR - later_cuts
# From earned_time 0.14 the floor itself is dated (LEISURE_FROM lifts it).
lift = 60 * earned_time.gaming_base_minutes_for(day) - 5 * _HOUR
assert resolved.base_seconds == 4 * _HOUR + lift - later_cuts
class TestTheReadingTerm:
@ -166,8 +168,8 @@ class TestTheCeiling:
"""The ceiling is the registry's, read when the budget resolves."""
with (
pin_today(AFTER_CUT),
# resolve() binds the ceiling from _policy by name, so patch there.
patch("earned_time._resolve.GAMING_CEILING_MINUTES", 6 * 60),
# 0.14 reads the ceiling via _leisure (dated), so patch it there.
patch("earned_time._leisure.GAMING_CEILING_MINUTES", 6 * 60),
):
resolved = _resolve(earners=(True, True, True))
assert resolved.seconds == 6 * _HOUR

146
steam_backlog_enforcer/tests/test_counted_procs.py
View File

@ -1,146 +0,0 @@
"""Tests for _counted_procs — 100% branch coverage.
The validation here is load-bearing rather than cosmetic: these names are fed
to the budget cutoff and the total-block killer, and `_playtime_kill` signals a
matched PID *and its descendants*. A name like `sh` would take the whole
session down with it.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from steam_backlog_enforcer import config as config_mod
from steam_backlog_enforcer._counted_procs import (
DEFAULT_COUNTED_PROCESSES,
CountedProcess,
counted_process_names,
key_by_name,
kill_target_names,
labels_by_key,
load_counted_processes,
parse_counted_processes,
)
if TYPE_CHECKING:
from pathlib import Path
def _write(payload: object) -> None:
config_mod.CONFIG_FILE.write_text(json.dumps(payload), encoding="utf-8")
class TestParse:
"""Turning raw JSON into validated entries."""
def test_reads_a_well_formed_entry(self) -> None:
entries = parse_counted_processes(
[{"id": "osu-lazer", "label": "osu!lazer", "names": ["osu!", "osu-lazer"]}]
)
assert entries == (
CountedProcess(
id="osu-lazer",
label="osu!lazer",
names=frozenset({"osu!", "osu-lazer"}),
),
)
def test_label_defaults_to_the_id(self) -> None:
entries = parse_counted_processes([{"id": "quake", "names": ["quake"]}])
assert entries[0].label == "quake"
def test_rejects_a_non_list(self) -> None:
assert parse_counted_processes({"id": "x"}) == ()
def test_drops_a_non_object_entry(self) -> None:
assert parse_counted_processes(["nope"]) == ()
def test_drops_an_entry_with_no_id(self) -> None:
assert parse_counted_processes([{"names": ["x"]}]) == ()
def test_drops_an_entry_with_no_usable_names(self) -> None:
assert parse_counted_processes([{"id": "x", "names": []}]) == ()
def test_drops_an_entry_whose_names_are_not_a_list(self) -> None:
assert parse_counted_processes([{"id": "x", "names": "osu!"}]) == ()
def test_drops_a_duplicate_id(self) -> None:
entries = parse_counted_processes(
[
{"id": "x", "names": ["a"]},
{"id": "x", "names": ["b"]},
]
)
assert len(entries) == 1
assert entries[0].names == frozenset({"a"})
def test_ignores_non_string_and_blank_names(self) -> None:
entries = parse_counted_processes([{"id": "x", "names": ["a", 7, " ", None]}])
assert entries[0].names == frozenset({"a"})
def test_refuses_a_shell_or_interpreter_name(self) -> None:
"""`sh` would SIGTERM every shell on the system, and its children."""
entries = parse_counted_processes(
[{"id": "osu", "names": ["sh", "dotnet", "osu!"]}]
)
assert entries[0].names == frozenset({"osu!"})
def test_an_entry_of_only_forbidden_names_is_dropped(self) -> None:
assert parse_counted_processes([{"id": "osu", "names": ["sh"]}]) == ()
class TestLoad:
"""Reading straight from config.json, with no Config to consult."""
def test_missing_file_falls_back_to_defaults(self) -> None:
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
def test_unreadable_file_falls_back_to_defaults(self) -> None:
config_mod.CONFIG_FILE.write_text("{not json", encoding="utf-8")
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
def test_non_object_payload_falls_back_to_defaults(self) -> None:
_write([1, 2, 3])
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
def test_absent_key_falls_back_to_defaults(self) -> None:
_write({"steam_id": "1"})
assert load_counted_processes() == DEFAULT_COUNTED_PROCESSES
def test_an_explicit_empty_list_means_none(self) -> None:
"""Distinct from absent: the user can switch the feature off."""
_write({"counted_processes": []})
assert load_counted_processes() == ()
def test_reads_a_configured_list(self) -> None:
_write({"counted_processes": [{"id": "q", "names": ["quake"]}]})
assert load_counted_processes()[0].id == "q"
def test_the_default_covers_every_osu_layer(self) -> None:
"""The sh wrapper, the AppImage and the game each exec a new comm."""
assert counted_process_names(DEFAULT_COUNTED_PROCESSES) == frozenset(
{"osu-lazer", "osu.AppImage", "osu!"}
)
class TestDerivedViews:
"""The three projections the billing and kill paths consume."""
_ENTRIES = (
CountedProcess(id="osu-lazer", label="osu!lazer", names=frozenset({"osu!"})),
)
def test_names_of_no_entries_is_empty(self) -> None:
assert counted_process_names(()) == frozenset()
def test_key_by_name(self) -> None:
assert key_by_name(self._ENTRIES) == {"osu!": "proc:osu-lazer"}
def test_labels_by_key(self) -> None:
assert labels_by_key(self._ENTRIES) == {"proc:osu-lazer": "osu!lazer"}
def test_kill_targets_come_from_the_config(self, tmp_path: Path) -> None:
assert tmp_path.exists()
_write({"counted_processes": [{"id": "q", "names": ["quake"]}]})
assert kill_target_names() == frozenset({"quake"})

77
steam_backlog_enforcer/tests/test_enforcer.py
View File

@ -3,91 +3,14 @@
from __future__ import annotations
import subprocess
from typing import TYPE_CHECKING
from unittest.mock import patch
from steam_backlog_enforcer.enforcer import (
enforce_allowed_game,
get_running_steam_game_pids,
kill_process,
send_notification,
)
if TYPE_CHECKING:
from pathlib import Path
class TestGetRunningPids:
"""Tests for get_running_steam_game_pids."""
def test_finds_steam_pid(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "12345"
pid_dir.mkdir(parents=True)
environ = b"HOME=/home/user\x00SteamAppId=440\x00PATH=/usr/bin"
(pid_dir / "environ").write_bytes(environ)
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_running_steam_game_pids()
assert result == {12345: 440}
def test_skips_non_digit_entries(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
proc_dir.mkdir(parents=True)
(proc_dir / "self").mkdir()
(proc_dir / "cpuinfo").touch()
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_running_steam_game_pids()
assert result == {}
def test_handles_permission_error(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "99"
pid_dir.mkdir(parents=True)
# No environ file -> OSError when reading
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_running_steam_game_pids()
assert result == {}
def test_skips_non_digit_steam_app_id(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "100"
pid_dir.mkdir(parents=True)
environ = b"SteamAppId=notanumber\x00"
(pid_dir / "environ").write_bytes(environ)
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_running_steam_game_pids()
assert result == {}
def test_no_steam_env(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "200"
pid_dir.mkdir(parents=True)
environ = b"HOME=/home/user\x00PATH=/usr/bin\x00"
(pid_dir / "environ").write_bytes(environ)
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_running_steam_game_pids()
assert result == {}
class TestEnforceAllowedGame:
"""Tests for enforce_allowed_game."""

75
steam_backlog_enforcer/tests/test_enforcer_part2.py
View File

@ -2,88 +2,13 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import patch
from steam_backlog_enforcer.enforcer import (
_kill_pid_by_name,
get_pids_by_process_names,
kill_processes_by_name,
)
if TYPE_CHECKING:
from pathlib import Path
class TestGetPidsByProcessNames:
"""Tests for get_pids_by_process_names."""
def test_finds_matching_comm(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "12345"
pid_dir.mkdir(parents=True)
(pid_dir / "comm").write_text("lutris\n", encoding="utf-8")
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_pids_by_process_names(frozenset({"lutris"}))
assert result == {12345: "lutris"}
def test_no_match_returns_empty(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "12345"
pid_dir.mkdir(parents=True)
(pid_dir / "comm").write_text("bash\n", encoding="utf-8")
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_pids_by_process_names(frozenset({"lutris"}))
assert result == {}
def test_skips_non_digit_entries(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
proc_dir.mkdir(parents=True)
(proc_dir / "self").mkdir()
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_pids_by_process_names(frozenset({"lutris"}))
assert result == {}
def test_handles_missing_comm_file(self, tmp_path: Path) -> None:
proc_dir = tmp_path / "proc"
(proc_dir / "42").mkdir(parents=True)
# No comm file -> OSError when reading.
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_pids_by_process_names(frozenset({"lutris"}))
assert result == {}
def test_truncates_long_names_to_15_chars(self, tmp_path: Path) -> None:
"""The kernel truncates /proc/[pid]/comm to 15 chars - matching
must compare against the truncated form, not the full name."""
proc_dir = tmp_path / "proc"
pid_dir = proc_dir / "777"
pid_dir.mkdir(parents=True)
# "EpicGamesLauncher" truncated to 15 chars is "EpicGamesLaunch".
(pid_dir / "comm").write_text("EpicGamesLaunch\n", encoding="utf-8")
with patch(
"steam_backlog_enforcer.enforcer.Path",
return_value=proc_dir,
):
result = get_pids_by_process_names(frozenset({"EpicGamesLauncher"}))
assert result == {777: "EpicGamesLauncher"}
class TestKillProcessesByName:
"""Tests for kill_processes_by_name."""

189
steam_backlog_enforcer/tests/test_playtime_attribution.py
View File

@ -1,189 +0,0 @@
"""Tests for who a billed tick belongs to — 100% branch coverage.
Split from ``test_playtime_procs`` at the 250-line cap. This file owns the
attribution dimension: which PIDs qualify, and which single game a tick is
credited to. ``test_playtime_procs`` keeps the /proc plumbing.
"""
from __future__ import annotations
from unittest.mock import patch
from steam_backlog_enforcer._counted_procs import CountedProcess
from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pids
from steam_backlog_enforcer._playtime_state import rules_for
from steam_backlog_enforcer.config import Config
class TestQualifyingPids:
"""`qualifying_pids` returns pid -> attribution key, not a bare pid set."""
def test_steam_games_only_when_launchers_are_off(self) -> None:
rules = rules_for(Config(count_launcher_processes=False), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={5: 440, 6: 0},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(),
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names"
) as mock_comm,
):
assert qualifying_pids(rules) == {5: "app:440"}
mock_comm.assert_not_called()
def test_steam_client_appid_zero_does_not_count(self) -> None:
"""Browsing the store is not gaming."""
rules = rules_for(Config(count_launcher_processes=False), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={9: 0},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(),
),
):
assert qualifying_pids(rules) == {}
def test_unions_launchers_from_both_scanners(self) -> None:
rules = rules_for(Config(count_launcher_processes=True), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={5: 440},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(),
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
return_value={7: "heroic"},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
return_value={8: "lutris"},
),
):
assert qualifying_pids(rules) == {
5: "app:440",
7: "launcher:heroic",
8: "launcher:lutris",
}
def test_counted_processes_bill_even_with_launchers_off(self) -> None:
"""The user's own list is not gated on the launcher switch.
Turning launcher counting off because a launcher idles in the tray must
not silently stop billing a game the user explicitly listed.
"""
rules = rules_for(Config(count_launcher_processes=False), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(
CountedProcess(
id="osu-lazer", label="osu!lazer", names=frozenset({"osu!"})
),
),
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
return_value={11: "osu!"},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
return_value={},
),
):
assert qualifying_pids(rules) == {11: "proc:osu-lazer"}
def test_steam_attribution_wins_over_a_name_match(self) -> None:
"""A Steam game keeps its app key even if its comm matches a launcher."""
rules = rules_for(Config(count_launcher_processes=True), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={7: 440},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(),
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
return_value={7: "heroic"},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
return_value={},
),
):
assert qualifying_pids(rules) == {7: "app:440"}
def test_a_name_outside_the_map_is_skipped(self) -> None:
"""Unreachable in production; a KeyError in the enforce loop is not."""
rules = rules_for(Config(count_launcher_processes=True), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={},
),
patch(
"steam_backlog_enforcer._playtime_procs.load_counted_processes",
return_value=(),
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
return_value={12: "not-a-launcher"},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
return_value={},
),
):
assert qualifying_pids(rules) == {}
def test_no_processes_at_all(self) -> None:
rules = rules_for(Config(), demo=False)
with (
patch(
"steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
return_value={},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_process_names",
return_value={},
),
patch(
"steam_backlog_enforcer._playtime_procs.get_pids_by_cmdline_names",
return_value={},
),
):
assert qualifying_pids(rules) == {}
class TestAttributedKey:
"""Deciding which single game a tick belongs to."""
def test_falls_back_to_the_only_game_running(self) -> None:
assert attributed_key({7: "app:440", 8: "app:440"}) == "app:440"
def test_two_games_is_unattributable(self) -> None:
# There is no honest way to say which one earned the tick, so the
# tick still bills but the gap shows up as Unattributed rather than
# being charged to a guess.
assert attributed_key({7: "app:440", 8: "app:1"}) == ""
def test_nothing_qualifying(self) -> None:
assert attributed_key({}) == ""

22
steam_backlog_enforcer/tests/test_playtime_notify.py Normal file
View File

@ -0,0 +1,22 @@
"""Tests for _playtime_notify._humanise (kept from test_playtime_procs)."""
from __future__ import annotations
from steam_backlog_enforcer._playtime_notify import _humanise
class TestHumanise:
def test_seconds(self) -> None:
assert _humanise(30) == "30 seconds"
def test_exact_minute_boundary(self) -> None:
assert _humanise(60) == "1 minutes"
def test_minutes(self) -> None:
assert _humanise(300) == "5 minutes"
def test_one_hour_is_special_cased(self) -> None:
assert _humanise(3600) == "1 hour"
def test_thirty_minutes(self) -> None:
assert _humanise(1800) == "30 minutes"

144
steam_backlog_enforcer/tests/test_playtime_procs.py
View File

@ -1,144 +0,0 @@
"""Tests for the main CLI: cmdline-based process detection."""
from datetime import datetime, timedelta, timezone
import os
from typing import TYPE_CHECKING
from steam_backlog_enforcer._playtime_notify import (
_humanise,
)
from steam_backlog_enforcer._playtime_procs import (
_match_cmdline,
get_pids_by_cmdline_names,
process_name,
)
from steam_backlog_enforcer._playtime_state import (
PlaytimeState,
)
if TYPE_CHECKING:
from pathlib import Path
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
NOW = datetime(2026, 7, 27, 20, 0, tzinfo=LOCAL)
INTERVAL = 3.0
def _state_ticked(seconds_ago: float, **kwargs: object) -> PlaytimeState:
return PlaytimeState(last_tick_at=NOW.timestamp() - seconds_ago, **kwargs)
class TestMatchCmdline:
def _proc(self, tmp_path: Path, pid: int, argv: list[str]) -> Path:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(
b"\x00".join(a.encode() for a in argv) + b"\x00"
)
return entry
def test_direct_argv0_match(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 10, ["/usr/bin/lutris", "--foo"])
assert _match_cmdline(entry, frozenset({"lutris"})) == "lutris"
def test_interpreter_falls_through_to_argv1(self, tmp_path: Path) -> None:
"""The real case: /usr/bin/lutris is a python3 script."""
entry = self._proc(tmp_path, 11, ["/usr/bin/python3", "/usr/bin/lutris"])
assert _match_cmdline(entry, frozenset({"lutris"})) == "lutris"
def test_interpreter_with_no_argv1(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 12, ["python3"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_interpreter_running_something_else(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 13, ["python3", "/usr/bin/other"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_non_interpreter_no_match(self, tmp_path: Path) -> None:
entry = self._proc(tmp_path, 14, ["/usr/bin/firefox"])
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_empty_cmdline(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "15"
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(b"")
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_unreadable_cmdline(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "16"
entry.mkdir(parents=True, exist_ok=True)
assert _match_cmdline(entry, frozenset({"lutris"})) is None
def test_undecodable_bytes_do_not_raise(self, tmp_path: Path) -> None:
entry = tmp_path / "playtime_proc" / "17"
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(b"\xff\xfe\x00")
assert _match_cmdline(entry, frozenset({"lutris"})) is None
class TestGetPidsByCmdlineNames:
def _proc(self, tmp_path: Path, pid: int, argv: list[str]) -> None:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "cmdline").write_bytes(
b"\x00".join(a.encode() for a in argv) + b"\x00"
)
def test_finds_matches(self, tmp_path: Path) -> None:
self._proc(tmp_path, 100, ["python3", "/usr/bin/lutris"])
self._proc(tmp_path, 101, ["/usr/bin/firefox"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {100: "lutris"}
def test_skips_non_digit_entries(self, tmp_path: Path) -> None:
(tmp_path / "playtime_proc" / "cpuinfo").touch()
(tmp_path / "playtime_proc" / "self").mkdir(exist_ok=True)
self._proc(tmp_path, 102, ["python3", "/usr/bin/lutris"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {102: "lutris"}
def test_never_matches_our_own_pid(self, tmp_path: Path) -> None:
"""The daemon is itself python3 - it must not count as a launcher."""
self._proc(tmp_path, os.getpid(), ["python3", "/usr/bin/lutris"])
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {}
def test_empty_proc(self) -> None:
assert get_pids_by_cmdline_names(frozenset({"lutris"})) == {}
class TestHumanise:
def test_seconds(self) -> None:
assert _humanise(30) == "30 seconds"
def test_exact_minute_boundary(self) -> None:
assert _humanise(60) == "1 minutes"
def test_minutes(self) -> None:
assert _humanise(300) == "5 minutes"
def test_one_hour_is_special_cased(self) -> None:
assert _humanise(3600) == "1 hour"
def test_thirty_minutes(self) -> None:
assert _humanise(1800) == "30 minutes"
class TestProcessName:
"""Tests for resolving a PID to its program name."""
def _comm(self, tmp_path: Path, pid: int, name: str) -> None:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "comm").write_text(name, encoding="utf-8")
def test_reads_comm(self, tmp_path: Path) -> None:
self._comm(tmp_path, 42, "hollow_knight\n")
assert process_name(42) == "hollow_knight"
def test_missing_pid_is_none(self) -> None:
# A PID recorded earlier may have exited, and the number may since have
# been recycled — "gone" is the only safe answer.
assert process_name(999999) is None
def test_empty_comm_is_none(self, tmp_path: Path) -> None:
self._comm(tmp_path, 43, "\n")
assert process_name(43) is None

7
steam_backlog_enforcer/tests/test_playtime_session.py
View File

@ -19,3 +19,10 @@ class TestNewSession:
demo = new_session(demo=True)
assert demo.journal._log._path == log_mod.BUDGET_DEMO_LOG_FILE
assert live.journal._log._path == log_mod.BUDGET_LOG_FILE
def test_builds_a_shadow_journal_on_the_same_trail(self) -> None:
from steam_backlog_enforcer._shadow_log import ShadowJournal
session = new_session()
assert isinstance(session.shadow, ShadowJournal)
assert session.shadow._log._path == log_mod.BUDGET_LOG_FILE

144
steam_backlog_enforcer/tests/test_shadow.py Normal file
View File

@ -0,0 +1,144 @@
"""Tests for shadow leisure billing (contract C4): verdict and accumulation."""
from __future__ import annotations
from dataclasses import replace
from unittest.mock import patch
from leisure_watch.facts import Focus, Now, Reason
import pytest
from steam_backlog_enforcer import _shadow
from steam_backlog_enforcer._playtime_state import PlaytimeState
from steam_backlog_enforcer._shadow import (
FAILCLOSED_KEY,
ShadowVerdict,
observe_shadow,
shadow_accumulate,
shadow_keys,
)
T0 = 1_760_000_000.0
_FOCUS = Focus(app="Helium", title="t", domain="youtube.com", idle=False)
def _now(state: str, *keys: str) -> Now:
reasons = tuple(Reason("focus", key, None, key) for key in keys)
return Now(ts=T0, seq=1, state=state, paused_by=None, reasons=reasons, focus=_FOCUS)
def _state(**over: object) -> PlaytimeState:
base = PlaytimeState(day_key="2026-10-19", last_tick_at=T0, seconds=50.0)
return replace(base, **over)
class TestShadowKeys:
def test_games_win_over_the_watcher(self) -> None:
owners = {2: "app:9", 1: "app:440", 3: "app:9"}
keys = shadow_keys(owners, _now("leisure", "domain:youtube.com"))
assert keys == ("app:440", "app:9")
def test_a_missing_fact_fails_closed(self) -> None:
assert shadow_keys({}, None) == (FAILCLOSED_KEY,)
def test_leisure_uses_the_reason_keys_deduplicated(self) -> None:
fact = _now("leisure", "domain:youtube.com", "app:mpv", "app:mpv")
assert shadow_keys({}, fact) == ("domain:youtube.com", "app:mpv")
def test_leisure_without_reasons_fails_closed(self) -> None:
assert shadow_keys({}, _now("leisure")) == (FAILCLOSED_KEY,)
@pytest.mark.parametrize("state", ["free", "paused"])
def test_free_or_watcher_paused_bills_nothing(self, state: str) -> None:
assert shadow_keys({}, _now(state, "x")) == ()
class TestVerdictState:
def test_paused(self) -> None:
assert ShadowVerdict(paused_by="i3lock", keys=()).state == "paused"
def test_leisure(self) -> None:
assert ShadowVerdict(paused_by=None, keys=("a",)).state == "leisure"
def test_free(self) -> None:
assert ShadowVerdict(paused_by=None, keys=()).state == "free"
class TestObserveShadow:
def test_paused_skips_the_fact_file(self) -> None:
with (
patch.object(_shadow, "paused_by", return_value="gatelock:x") as lock,
patch.object(_shadow, "read_now") as read,
):
verdict = observe_shadow({1: "app:440"}, stamp=T0)
assert verdict == ShadowVerdict(paused_by="gatelock:x", keys=())
lock.assert_called_once_with(_shadow.CLAIMS_DIR)
read.assert_not_called()
def test_reads_the_fact_once_with_the_contract_age(self) -> None:
fact = _now("leisure", "domain:youtube.com")
with (
patch.object(_shadow, "paused_by", return_value=None),
patch.object(_shadow, "read_now", return_value=fact) as read,
):
verdict = observe_shadow({}, stamp=T0)
assert verdict.keys == ("domain:youtube.com",)
read.assert_called_once_with(_shadow.NOW_PATH, T0, max_age=15)
def test_root_paths_are_the_desktop_users(self) -> None:
assert str(_shadow.NOW_PATH) == "/run/user/1000/leisure-watch/now.json"
assert str(_shadow.CLAIMS_DIR) == "/run/user/1000/gatelock/claims"
class TestShadowAccumulate:
def _bill(self, state: PlaytimeState, verdict: ShadowVerdict) -> PlaytimeState:
return shadow_accumulate(state, stamp=T0 + 3.0, interval=3.0, verdict=verdict)
def test_first_tick_credits_nothing(self) -> None:
state = _state(last_tick_at=0.0)
assert self._bill(state, ShadowVerdict(None, ("a",))) == state
def test_a_backward_clock_credits_nothing(self) -> None:
state = _state(last_tick_at=T0 + 10.0)
assert self._bill(state, ShadowVerdict(None, ("a",))) == state
def test_paused_goes_to_paused_seconds_only(self) -> None:
out = self._bill(_state(), ShadowVerdict("i3lock", ()))
assert out.shadow_paused_seconds == 3.0
assert out.shadow_seconds == 0.0
assert out.shadow_per_source == {}
def test_free_bills_nothing(self) -> None:
state = _state()
assert self._bill(state, ShadowVerdict(None, ())) == state
def test_one_source_takes_the_whole_tick(self) -> None:
out = self._bill(_state(), ShadowVerdict(None, ("app:440",)))
assert out.shadow_seconds == 3.0
assert out.shadow_per_source == {"app:440": 3.0}
def test_overlapping_sources_bill_once_and_split(self) -> None:
state = _state(shadow_per_source={"a": 1.0}, shadow_seconds=1.0)
out = self._bill(state, ShadowVerdict(None, ("a", "b")))
assert out.shadow_seconds == 4.0
assert out.shadow_per_source == {"a": 2.5, "b": 1.5}
assert sum(out.shadow_per_source.values()) <= out.shadow_seconds + 1e-9
def test_the_delta_is_clamped_like_the_enforced_one(self) -> None:
out = shadow_accumulate(
_state(),
stamp=T0 + 3600.0,
interval=3.0,
verdict=ShadowVerdict(None, (FAILCLOSED_KEY,)),
)
assert out.shadow_seconds == 6.0
def test_never_touches_enforced_fields(self) -> None:
state = _state(per_game={"app:1": 5.0}, last_credited_key="app:1")
out = self._bill(state, ShadowVerdict(None, ("app:440",)))
assert (out.seconds, out.last_tick_at, out.per_game) == (
state.seconds,
state.last_tick_at,
state.per_game,
)
assert out.last_credited_key == "app:1"

181
steam_backlog_enforcer/tests/test_shadow_budget.py Normal file
View File

@ -0,0 +1,181 @@
"""Tests for the shadow leisure budget: resolution, the held day, the tick, views."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from steam_backlog_enforcer import _budget_view as budget_view
from steam_backlog_enforcer import _leetcode_bonus, _shadow, _workout_budget
from steam_backlog_enforcer import _mcp_gaming_time as mcp_gaming_time
from steam_backlog_enforcer._budget_resolve import resolve_budget
from steam_backlog_enforcer._gaming_days import held_shadow_today
from steam_backlog_enforcer._playtime import playtime_tick
from steam_backlog_enforcer._playtime_state import (
PlaytimeState,
load_state,
rules_for,
save_state,
)
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._budget_dates import AFTER_CUT, pin_today
from steam_backlog_enforcer.tests._fake_session import fake_session
if TYPE_CHECKING:
from collections.abc import Iterator
_WORKOUT = "steam_backlog_enforcer._workout_budget._fetch_workout_today"
_LEETCODE = "steam_backlog_enforcer._leetcode_bonus.read_ledger_solved_today"
_READING = "steam_backlog_enforcer._budget_resolve.read_today"
_HOUR = 3600.0
_LOCAL = timezone(timedelta(hours=2))
_NOW = datetime(2026, 7, 27, 20, 0, tzinfo=_LOCAL)
_TODAY = "2026-07-27"
_PKG = "steam_backlog_enforcer._playtime"
@pytest.fixture(autouse=True)
def _fresh_caches() -> Iterator[None]:
_workout_budget.reset_cache()
_leetcode_bonus.reset_cache()
with pin_today(AFTER_CUT):
yield
_workout_budget.reset_cache()
_leetcode_bonus.reset_cache()
def _resolve(*, done: bool) -> object:
with (
patch(_WORKOUT, return_value=done),
patch(_LEETCODE, return_value=done),
patch(_READING, return_value=done),
):
return resolve_budget(Config())
class TestShadowResolution:
def test_a_fully_done_day_prices_ten_hours_and_enforces_eight(self) -> None:
resolved = _resolve(done=True)
assert resolved.shadow_seconds == 10 * _HOUR
assert resolved.seconds == 8 * _HOUR
def test_the_leisure_base_is_two_hours_above_the_enforced_one(self) -> None:
resolved = _resolve(done=False)
assert resolved.shadow_seconds == resolved.seconds + 2 * _HOUR
class TestHeldShadow:
def test_same_day_holds_the_high_water(self) -> None:
stored = PlaytimeState(day_key=_TODAY, shadow_budget_seconds=9 * _HOUR)
assert held_shadow_today(stored, _NOW) == 9 * _HOUR
def test_another_day_or_no_state_holds_nothing(self) -> None:
stored = PlaytimeState(day_key="2026-07-26", shadow_budget_seconds=_HOUR)
assert held_shadow_today(stored, _NOW) == 0.0
assert held_shadow_today(None, _NOW) == 0.0
def test_rules_take_the_higher_of_live_and_held(self) -> None:
held = "steam_backlog_enforcer._playtime_state.held_shadow_today"
with patch(held, return_value=20 * _HOUR):
assert rules_for(Config(), demo=False).shadow_budget_seconds == 20 * (_HOUR)
with patch(held, return_value=0.0):
live = rules_for(Config(), demo=False).shadow_budget_seconds
assert 0.0 < live < 20 * _HOUR
def test_demo_has_no_shadow_budget(self) -> None:
assert rules_for(Config(), demo=True).shadow_budget_seconds == 0.0
class TestShadowTick:
def _tick(self, owners: dict[int, str], *, total_block: bool = False) -> None:
save_state(
PlaytimeState(day_key=_TODAY, last_tick_at=_NOW.timestamp() - 3.0),
demo=False,
)
cutoff = "steam_backlog_enforcer._playtime_cutoff"
with (
patch(f"{cutoff}.reconcile"),
patch(f"{cutoff}.notify_desktop_user"),
patch(f"{_PKG}.mounted_targets", return_value=set()),
patch(f"{_PKG}.is_total_block_active", return_value=total_block),
patch(f"{_PKG}.qualifying_pids", return_value=owners),
patch.object(_shadow, "read_now", return_value=None),
patch.object(_shadow, "paused_by", return_value=None),
patch(f"{_PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = _NOW
playtime_tick(Config(), interval=3.0, session=fake_session())
def test_bills_shadow_beside_the_enforced_count(self) -> None:
self._tick({7: "app:440"})
state = load_state(demo=False)
assert state is not None
assert (state.seconds, state.per_game) == (3.0, {"app:440": 3.0})
assert state.shadow_seconds == 3.0
assert state.shadow_per_source == {"app:440": 3.0}
assert state.shadow_budget_seconds > 0.0
def test_no_game_and_no_watcher_fails_closed(self) -> None:
self._tick({})
state = load_state(demo=False)
assert state is not None
assert state.seconds == 0.0
assert state.shadow_per_source == {_shadow.FAILCLOSED_KEY: 3.0}
def test_no_shadow_under_a_total_block(self) -> None:
self._tick({}, total_block=True)
state = load_state(demo=False)
assert state is not None
assert state.shadow_seconds == 0.0
assert state.shadow_budget_seconds > 0.0
def test_the_session_journals_the_shadow_verdict(self) -> None:
session = fake_session()
save_state(
PlaytimeState(day_key=_TODAY, last_tick_at=_NOW.timestamp() - 3.0),
demo=False,
)
with (
patch("steam_backlog_enforcer._playtime_cutoff.reconcile"),
patch(f"{_PKG}.mounted_targets", return_value=set()),
patch(f"{_PKG}.is_total_block_active", return_value=False),
patch(f"{_PKG}.qualifying_pids", return_value={}),
patch(f"{_PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = _NOW
playtime_tick(Config(), interval=3.0, session=session)
assert [v.state for v in session.shadow.observed] == ["leisure"]
class TestShadowViews:
def test_mcp_reports_the_shadow_block(self) -> None:
stored = PlaytimeState(
day_key=_TODAY,
seconds=10.0,
shadow_seconds=40.0,
shadow_budget_seconds=100.0,
shadow_per_source={"app:440": 40.0},
shadow_paused_seconds=5.0,
)
with (
patch.object(budget_view, "load_state", return_value=stored),
patch.object(budget_view, "mounted_targets", return_value=set()),
):
out = mcp_gaming_time.get_gaming_time()
assert out["shadow"] == {
"seconds_used": 40.0,
"budget_seconds": 100.0,
"seconds_remaining": 60.0,
"would_block": False,
"paused_seconds": 5.0,
"per_source": {"app:440": 40.0},
}
def test_would_block_at_zero(self) -> None:
stored = PlaytimeState(shadow_seconds=100.0, shadow_budget_seconds=100.0)
shadow = budget_view.build_today(stored, rules_for(Config(), demo=True))
assert shadow["shadow"]["would_block"] is True
assert shadow["shadow"]["seconds_remaining"] == 0.0

65
steam_backlog_enforcer/tests/test_shadow_log.py Normal file
View File

@ -0,0 +1,65 @@
"""Tests for the shadow verdict journal: one budget.jsonl record per change."""
from __future__ import annotations
import json
from steam_backlog_enforcer import _playtime_log as log_mod
from steam_backlog_enforcer._budget_log_tail import last_verdict
from steam_backlog_enforcer._playtime_log import BudgetLog
from steam_backlog_enforcer._playtime_state import PlaytimeState
from steam_backlog_enforcer._shadow import ShadowVerdict
from steam_backlog_enforcer._shadow_log import EVENT_SHADOW_CHANGE, ShadowJournal
_STATE = PlaytimeState(
day_key="2026-10-19",
shadow_seconds=30.0,
shadow_budget_seconds=36000.0,
shadow_per_source={"app:440": 30.0},
shadow_paused_seconds=6.0,
)
def _records() -> list[dict[str, object]]:
text = log_mod.BUDGET_LOG_FILE.read_text(encoding="utf-8")
return [json.loads(line) for line in text.splitlines()]
def _journal() -> ShadowJournal:
return ShadowJournal(BudgetLog(path=log_mod.BUDGET_LOG_FILE))
class TestShadowJournal:
def test_writes_on_change_only(self) -> None:
journal = _journal()
leisure = ShadowVerdict(paused_by=None, keys=("app:440",))
journal.observe(leisure, _STATE)
journal.observe(leisure, _STATE)
journal.observe(ShadowVerdict(paused_by="i3lock", keys=()), _STATE)
records = _records()
assert [r["shadow_state"] for r in records] == ["leisure", "paused"]
assert {r["event"] for r in records} == {EVENT_SHADOW_CHANGE}
def test_a_new_source_is_a_change(self) -> None:
journal = _journal()
journal.observe(ShadowVerdict(None, ("a",)), _STATE)
journal.observe(ShadowVerdict(None, ("b",)), _STATE)
assert [r["sources"] for r in _records()] == [["a"], ["b"]]
def test_carries_the_shadow_fields(self) -> None:
_journal().observe(ShadowVerdict(None, ("app:440",)), _STATE)
record = _records()[0]
assert record["day_key"] == "2026-10-19"
assert record["shadow_seconds"] == 30.0
assert record["shadow_budget_seconds"] == 36000.0
assert record["shadow_remaining_seconds"] == 35970.0
assert record["shadow_paused_seconds"] == 6.0
assert record["shadow_per_source"] == {"app:440": 30.0}
assert record["paused_by"] is None
def test_live_status_skips_shadow_records(self) -> None:
"""The web UI's live game view reads the last *game* verdict."""
log = BudgetLog(path=log_mod.BUDGET_LOG_FILE)
log.record("verdict_change", state="engaged", qualifying=[])
ShadowJournal(log).observe(ShadowVerdict(None, ("a",)), _STATE)
assert last_verdict(demo=False).state == "engaged"

3
steam_backlog_enforcer/tests/test_total_block_purge.py
View File

@ -4,7 +4,8 @@ from __future__ import annotations
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer._total_block_launchers import LAUNCHER_PROCESS_NAMES
from leisure_watch.procscan import LAUNCHER_PROCESS_NAMES
from steam_backlog_enforcer._total_block_purge import (
is_steam_installed,
kill_and_uninstall_launchers,