Bill gaming time only while the user is actually playing

A running game process was treated as proof of play. On 2026-08-28 the budget
billed ~2h23m while a gatelock screen locker covered the display and the
machine could not be touched at all -- the game was simply still resident.

A tick now bills only when all three hold, evaluated together rather than
short-circuited so the audit log can answer "was the screen ALSO locked?":

  * no gatelock guard owns the screen  (_screen_hold, via /proc/locks)
  * the focused window belongs to a qualifying process or its ancestry
  * some input source has been used within idle_grace_seconds (default 300)

Any probe that cannot answer BILLS the tick and is recorded in `degraded`.
A broken detector must never become an unlimited-gaming exploit.

Three findings from measuring this machine rather than assuming:

  * /dev/input/js0 here is an *ASRock LED Controller* -- the kernel exposes
    the motherboard RGB device as a joystick. A presence check would have
    disabled the idle criterion permanently, so _controller_input watches for
    real events instead (skipping joydev's synthetic open burst).
  * Aseprite's window carries no _NET_WM_PID at all, so focus attribution
    falls back to XRes, which asks the server who owns the window rather than
    trusting the client to have announced itself.
  * /proc/locks reports gatelock's hold without taking it; gatelock's own
    read_claim_if_held detects held-ness by acquiring, which would race a
    guard every 3 seconds.

Idle pauses also refund the grace period once per walk-away, on the
engaged->paused edge only, never across a day rollover, and never on the first
tick after a restart -- the only place the counter moves backwards.

Everything is recorded to /var/log/steam-backlog-enforcer/budget.jsonl, on
change plus a heartbeat rather than one line per 3s tick. Records name the
qualifying PIDs: reconstructing this incident from journald alone was only
possible by inferring accrual from the spacing of two warning notifications.

Also: refuse to bounce Steam while a game is running. ensure_steam_debug_port
relaunches Steam when the CDP port is closed, which kills the running game --
that is how restarting the daemon to deploy this change ended a live session.
It now defers exactly as it already did for in-flight game updates.

Talks to X through ctypes over one cached connection: at a 3s tick, forking
xprintidle and xprop would cost ~40 forks a minute.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019CEsG6BzHtYR9cHcdxPZqG
This commit is contained in:
Krzysztof Rudnicki 2026-08-28 19:24:14 +02:00
parent 95f472ea6e
commit b3e629e3f0
33 changed files with 3031 additions and 44 deletions

View File

@ -0,0 +1,153 @@
"""Noticing that a game controller is being used.
The X screensaver idle counter only tracks keyboards and pointers. A gamepad
session that goes five minutes without touching either would look idle, and
pausing the budget there would hand out free gaming — the single worst failure
mode of the engagement gate, because it rewards the behaviour it fails to see.
Presence is not a usable proxy for use. Measured on this machine, ``/dev/input/js0``
is an *ASRock LED Controller*: the kernel exposes the motherboard's RGB device as
a joystick. Treating "a js device exists" as "a controller is in play" would
disable the idle criterion permanently, on a machine with no gamepad attached.
So this watches for actual events instead. Each joystick device is opened
non-blocking and drained every tick; joydev gives every reader its own event
queue, so draining ours cannot steal input from the game. The synthetic burst
joydev replays on open is flagged ``JS_EVENT_INIT`` and is skipped — without that
filter, merely opening the device would look like a button press.
"""
from __future__ import annotations
import contextlib
import errno
import logging
import os
from pathlib import Path
from typing import Final
logger = logging.getLogger(__name__)
_DEVICE_DIR: Final = Path("/dev/input")
_DEVICE_PREFIX: Final = "js"
# struct js_event { __u32 time; __s16 value; __u8 type; __u8 number; }
_EVENT_SIZE: Final = 8
_TYPE_OFFSET: Final = 6
_JS_EVENT_INIT: Final = 0x80
# Cap per device per tick so a flood cannot stall the enforce loop.
_MAX_EVENTS_PER_POLL: Final = 64
class ControllerActivity:
"""Tracks when a joystick device last produced a real event."""
def __init__(self, *, device_dir: Path = _DEVICE_DIR) -> None:
"""Start with no devices open and no activity seen.
Args:
device_dir: Directory holding ``js*`` nodes.
"""
self._device_dir = device_dir
self._fds: dict[Path, int] = {}
self._last_event_at: float | None = None
def poll(self, *, now: float) -> None:
"""Drain pending controller events, recording any real activity.
Args:
now: Monotonic timestamp for this tick.
"""
self._sync_devices()
for path in list(self._fds):
if self._drain(path):
self._last_event_at = now
def seconds_since_activity(self, *, now: float) -> float | None:
"""Return seconds since the last real controller event.
Args:
now: Monotonic timestamp for this tick.
Returns:
The elapsed seconds, or ``None`` if no event has been seen at all —
which the caller must not read as "idle", only as "no evidence".
"""
if self._last_event_at is None:
return None
return now - self._last_event_at
def close(self) -> None:
"""Close every open device."""
for path in list(self._fds):
self._forget(path)
def _sync_devices(self) -> None:
"""Open newly appeared ``js*`` devices and forget vanished ones."""
try:
present = {
entry
for entry in self._device_dir.iterdir()
if entry.name.startswith(_DEVICE_PREFIX)
}
except OSError:
present = set()
for path in list(self._fds):
if path not in present:
self._forget(path)
for path in present - set(self._fds):
self._open(path)
def _open(self, path: Path) -> None:
"""Open *path* for reading, ignoring devices we may not read.
Args:
path: Device to open.
"""
try:
self._fds[path] = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
except OSError:
# A js node we cannot open tells us nothing either way; the idle
# criterion simply falls back to keyboard and pointer input.
logger.debug("Cannot open controller device %s", path)
def _drain(self, path: Path) -> bool:
"""Read pending events from *path*.
Args:
path: Device to drain.
Returns:
Whether at least one non-synthetic event was read.
"""
fd = self._fds[path]
seen = False
for _ in range(_MAX_EVENTS_PER_POLL):
try:
chunk = os.read(fd, _EVENT_SIZE)
except OSError as exc:
if exc.errno != errno.EAGAIN:
self._forget(path)
break
if len(chunk) < _EVENT_SIZE:
break
if not chunk[_TYPE_OFFSET] & _JS_EVENT_INIT:
seen = True
return seen
def _forget(self, path: Path) -> None:
"""Close and drop *path*.
Args:
path: Device to close.
"""
fd = self._fds.pop(path, None)
if fd is not None:
# The usual reason to forget a device is that reading it just
# failed, which often means the fd is already unusable; closing it
# again must not take the enforce loop down with it.
with contextlib.suppress(OSError):
os.close(fd)

View File

@ -17,6 +17,7 @@ from steam_backlog_enforcer._enforce_steps import (
# from here, so splitting it into _owned_apps_cache stays invisible to them.
from steam_backlog_enforcer._owned_apps_cache import get_all_owned_app_ids
from steam_backlog_enforcer._playtime import playtime_tick
from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session
from steam_backlog_enforcer._steam_launch import steam_is_installed
from steam_backlog_enforcer._total_block import (
end_total_block_cleanup,
@ -103,6 +104,7 @@ def _enforce_loop_iteration(
config: Config,
state: State,
*,
session: PlaytimeSession,
demo: bool = False,
) -> None:
"""Perform one iteration of the enforcement loop.
@ -110,13 +112,14 @@ def _enforce_loop_iteration(
Args:
config: Enforcer configuration.
state: Current enforcer state.
session: Cross-tick engagement and logging state.
demo: Run the gaming budget on a 60-second demo budget.
"""
# Daily gaming budget runs FIRST and unconditionally. Every guard below
# returns early in situations where gaming time is still being spent (total
# block active, Steam absent, nothing assigned) and where a 06:00 release
# still has to happen — gating this on any of them would strand the block.
playtime_tick(config, interval=ENFORCE_INTERVAL, demo=demo)
playtime_tick(config, interval=ENFORCE_INTERVAL, session=session, demo=demo)
# Total block takes priority over the assigned-game enforcement below -
# while active, don't fight ourselves (e.g. installing the assigned
@ -204,6 +207,9 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
_echo(f" Enforce loop: ACTIVE (every {ENFORCE_INTERVAL}s)")
_echo(" Guarding: processes + installs + store")
_echo(" Press Ctrl+C to stop.\n")
# One session for the whole daemon: the engagement backdate needs to see
# the previous tick's verdict, so it cannot be rebuilt per iteration.
session = new_session()
try:
while True:
# Reload state from disk so CLI changes (e.g. new game
@ -219,7 +225,7 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
state.current_game_name = fresh.current_game_name
state.finished_app_ids = fresh.finished_app_ids
_enforce_loop_iteration(config, state, demo=demo)
_enforce_loop_iteration(config, state, session=session, demo=demo)
time.sleep(ENFORCE_INTERVAL)
except KeyboardInterrupt:
_echo("\nEnforcer stopped.")

View File

@ -0,0 +1,168 @@
"""Turning each raw engagement signal into a pause cause.
Split from :mod:`_playtime_engagement`, which owns the per-tick decision and the
backdate; this owns the three probes and the rule each one applies. Every probe
follows the same contract: append to *causes* when the criterion says "not
playing", append to *degraded* when the probe could not answer, and never raise.
**A probe that fails does not pause the tick.** Failure means the tick is billed.
A detector that stopped working must never become an unlimited-gaming exploit,
so the burden of proof sits on "not playing", not on "playing".
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._engagement_types import (
CAUSE_FOCUS,
CAUSE_IDLE,
CAUSE_SCREEN_HELD,
PROBE_FOCUS,
PROBE_IDLE,
PROBE_SCREEN_HELD,
CauseTally,
)
from steam_backlog_enforcer._playtime_kill import _MAX_PROCESS_TREE_DEPTH, _read_ppid
from steam_backlog_enforcer._screen_hold import ScreenHoldError, screen_hold
from steam_backlog_enforcer._x_focus import focused_pid
from steam_backlog_enforcer._x_probe import XProbeError
if TYPE_CHECKING:
from pathlib import Path
from steam_backlog_enforcer._controller_input import ControllerActivity
from steam_backlog_enforcer._playtime_state import PlaytimeRules
from steam_backlog_enforcer._x_probe import XProbe
logger = logging.getLogger(__name__)
_INIT_PID = 1
def assess_screen(
holder_path: Path, tally: CauseTally
) -> tuple[bool | None, int | None]:
"""Add the screen-held cause if a gatelock guard owns the display.
Args:
holder_path: The gatelock holder lock.
tally: Accumulating causes and probe failures.
Returns:
The hold state and holding PID, both ``None`` when the probe failed.
"""
try:
hold = screen_hold(holder_path)
except ScreenHoldError as exc:
logger.warning("Screen-hold probe failed; billing this tick: %s", exc)
tally.degraded.append(PROBE_SCREEN_HELD)
return None, None
if hold.held:
tally.causes.append(CAUSE_SCREEN_HELD)
return hold.held, hold.holder_pid
def assess_focus(
probe: XProbe,
rules: PlaytimeRules,
qualifying: set[int],
tally: CauseTally,
) -> int | None:
"""Add the focus cause unless a qualifying process owns the focus.
Args:
probe: X connection.
rules: Policy for this tick.
qualifying: PIDs that would otherwise count.
tally: Accumulating causes and probe failures.
Returns:
The focused PID, or ``None``.
"""
if not rules.require_game_focus:
return None
try:
pid = focused_pid(probe.connect())
except XProbeError as exc:
logger.warning("Focus probe failed; billing this tick: %s", exc)
tally.degraded.append(PROBE_FOCUS)
probe.close()
return None
if not focus_qualifies(pid, qualifying):
tally.causes.append(CAUSE_FOCUS)
return pid
def assess_idle(
probe: XProbe,
controller: ControllerActivity,
rules: PlaytimeRules,
tally: CauseTally,
now_monotonic: float,
) -> tuple[float | None, float | None]:
"""Add the idle cause when every input source has gone quiet.
The controller is polled first and its age folded in with ``min``: the X
idle counter does not see a gamepad, and pausing during controller play
would be the gate's worst failure.
Args:
probe: X connection.
controller: Controller-activity watcher.
rules: Policy for this tick.
tally: Accumulating causes and probe failures.
now_monotonic: Monotonic timestamp for controller ages.
Returns:
The effective idle seconds and the controller idle seconds.
"""
controller.poll(now=now_monotonic)
controller_idle = controller.seconds_since_activity(now=now_monotonic)
try:
pointer_idle = probe.idle_seconds()
except XProbeError as exc:
logger.warning("Idle probe failed; billing this tick: %s", exc)
tally.degraded.append(PROBE_IDLE)
probe.close()
return None, controller_idle
idle = pointer_idle
if controller_idle is not None:
idle = min(idle, controller_idle)
if idle >= rules.idle_grace_seconds:
tally.causes.append(CAUSE_IDLE)
return idle, controller_idle
def focus_qualifies(focus_pid: int | None, qualifying: set[int]) -> bool:
"""Whether *focus_pid* is, or descends from, a qualifying process.
A game's visible window is often owned by a child of the process carrying
``SteamAppId`` — Proton in particular — so the ancestry is walked rather
than compared directly.
Args:
focus_pid: PID owning the focused window, or ``None``.
qualifying: PIDs that count as gaming.
Returns:
Whether the focused window belongs to the game.
"""
if focus_pid is None:
return False
pid = focus_pid
for _ in range(_MAX_PROCESS_TREE_DEPTH):
if pid in qualifying:
return True
if pid <= _INIT_PID:
return False
parent = _read_ppid(pid)
if parent is None:
return False
pid = parent
return False

View File

@ -0,0 +1,91 @@
"""The verdict describing whether this tick should bill the gaming budget.
Split from :mod:`_playtime_engagement` so the decision *record* stays a plain
frozen value with no probes attached: everything that writes it is testable
without an X server, and everything that reads it — the accumulator and the
audit log — sees the same shape.
The verdict deliberately records *every* cause rather than short-circuiting on
the first one. The idle backdate only fires when idle is the sole cause, and a
log that recorded just the first-matching reason could not answer "was the
screen also locked?" after the fact.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Final
STATE_ENGAGED: Final = "engaged"
STATE_PAUSED: Final = "paused"
STATE_NOT_APPLICABLE: Final = "not_applicable"
CAUSE_IDLE: Final = "idle"
CAUSE_SCREEN_HELD: Final = "screen_held"
CAUSE_FOCUS: Final = "focus"
PROBE_IDLE: Final = "idle"
PROBE_SCREEN_HELD: Final = "screen_held"
PROBE_FOCUS: Final = "focus"
@dataclass
class CauseTally:
"""Pause causes and probe failures accumulated across one tick's probes.
Passed to each probe instead of two parallel lists: it keeps the probe
signatures short, and it makes "this probe appends to both" a single idea
rather than a convention two arguments have to remember.
"""
causes: list[str] = field(default_factory=list)
degraded: list[str] = field(default_factory=list)
@dataclass(frozen=True)
class EngagementVerdict:
"""Why this tick did or did not count against the budget."""
state: str
causes: tuple[str, ...] = ()
degraded: tuple[str, ...] = ()
idle_seconds: float | None = None
controller_idle_seconds: float | None = None
screen_held: bool | None = None
holder_pid: int | None = None
focus_pid: int | None = None
qualifying: tuple[int, ...] = field(default_factory=tuple)
@property
def engaged(self) -> bool:
"""Whether this tick should be billed.
``not_applicable`` counts as engaged: nothing qualifies, so the
accumulator will credit nothing regardless, and calling it "paused"
would fill the audit log with a pause reason during ordinary desktop
use.
Returns:
Whether the tick bills.
"""
return self.state != STATE_PAUSED
@property
def reason(self) -> str:
"""A short human-readable summary of *causes*.
Returns:
The joined causes, or the state when there are none.
"""
return "+".join(self.causes) if self.causes else self.state
def idle_only(self) -> bool:
"""Whether idleness is the sole reason this tick was paused.
The backdate refunds the idle grace period, which is only correct when
nothing else would have paused the tick anyway.
Returns:
Whether *causes* is exactly ``(CAUSE_IDLE,)``.
"""
return self.causes == (CAUSE_IDLE,)

View File

@ -30,6 +30,7 @@ from __future__ import annotations
from dataclasses import replace
from datetime import datetime, timezone
import logging
import time
from typing import TYPE_CHECKING
from steam_backlog_enforcer._playtime_block import (
@ -57,6 +58,7 @@ from steam_backlog_enforcer._playtime_state import (
from steam_backlog_enforcer._total_block import is_total_block_active
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_session import PlaytimeSession
from steam_backlog_enforcer.config import Config
logger = logging.getLogger(__name__)
@ -66,16 +68,27 @@ _SECONDS_PER_MINUTE = 60
_MINUTES_PER_HOUR = 60
def playtime_tick(config: Config, *, interval: float, demo: bool = False) -> None:
def playtime_tick(
config: Config,
*,
interval: float,
session: PlaytimeSession,
demo: bool = False,
) -> None:
"""Account for this tick's gaming time and enforce the daily budget.
Called first in every enforce-loop iteration, before any other guard, so
that time is still counted (and a 06:00 release still happens) in the
situations where the rest of the loop early-returns.
Time only accrues while the user is actually engaged: a resident game
process behind a locked screen, an unfocused window or five minutes of
silence is not play. See :mod:`_playtime_engagement`.
Args:
config: Loaded user configuration.
interval: Nominal seconds between enforce-loop ticks.
session: Cross-tick engagement and logging state.
demo: Whether this is a short-budget demo run.
"""
now = datetime.now(timezone.utc).astimezone()
@ -90,13 +103,20 @@ def playtime_tick(config: Config, *, interval: float, demo: bool = False) -> Non
save_state(state, demo=demo)
return
pids = qualifying_pids(rules)
monotonic = time.monotonic()
verdict = session.tracker.assess(rules, qualifying=pids, now_monotonic=monotonic)
state = accumulate(
state,
now=now,
qualifying=qualifying_pids(rules),
qualifying=pids if verdict.engaged else set(),
interval=interval,
)
save_state(_policy(state, rules, now=now), demo=demo)
state = session.tracker.backdate(state, verdict, rules=rules)
state = _policy(state, rules, now=now)
session.journal.observe(verdict, state, rules=rules, now_monotonic=monotonic)
save_state(state, demo=demo)
def _state_or_recover(rules: PlaytimeRules, *, now: datetime) -> PlaytimeState:

View File

@ -0,0 +1,164 @@
"""Deciding whether the user is actually playing, not just leaving a game open.
A running game process is not evidence of play. The budget was previously billed
for two hours during which a screen locker covered the display and the machine
could not be touched at all — the game was simply still resident.
Three signals gate a tick, evaluated together rather than short-circuited, so
that the log can answer "was the screen *also* locked?" after the fact and the
backdate can tell an idle-only pause from a compound one. The probes themselves
live in :mod:`_engagement_probes`.
This module owns the per-tick verdict and the one piece of state a pure function
cannot hold: the previous verdict, which the idle backdate needs to recognise an
engaged-to-paused edge.
"""
from __future__ import annotations
from dataclasses import replace
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._controller_input import ControllerActivity
from steam_backlog_enforcer._engagement_probes import (
assess_focus,
assess_idle,
assess_screen,
)
from steam_backlog_enforcer._engagement_types import (
STATE_ENGAGED,
STATE_NOT_APPLICABLE,
STATE_PAUSED,
CauseTally,
EngagementVerdict,
)
from steam_backlog_enforcer._screen_hold import holder_lock_path
from steam_backlog_enforcer._x_probe import XProbe
if TYPE_CHECKING:
from pathlib import Path
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
logger = logging.getLogger(__name__)
class EngagementTracker:
"""Assesses engagement per tick and owns the state the backdate needs."""
def __init__(
self,
*,
uid: int,
probe: XProbe | None = None,
controller: ControllerActivity | None = None,
holder_path: Path | None = None,
) -> None:
"""Build a tracker over the desktop session owned by *uid*.
Args:
uid: Owner of the desktop session.
probe: X probe, injected by tests.
controller: Controller-activity watcher, injected by tests.
holder_path: gatelock holder lock, injected by tests.
"""
self._probe = probe if probe is not None else XProbe()
self._controller = (
controller if controller is not None else ControllerActivity()
)
self._holder_path = (
holder_path if holder_path is not None else holder_lock_path(uid)
)
self._previous: EngagementVerdict | None = None
self._previous_day: str | None = None
def close(self) -> None:
"""Release the X connection and controller devices."""
self._probe.close()
self._controller.close()
def assess(
self,
rules: PlaytimeRules,
*,
qualifying: set[int],
now_monotonic: float,
) -> EngagementVerdict:
"""Judge whether this tick counts against the budget.
Args:
rules: Policy for this tick.
qualifying: PIDs that would otherwise count.
now_monotonic: Monotonic timestamp, for controller-activity ages.
Returns:
The verdict for this tick.
"""
if not qualifying:
return EngagementVerdict(state=STATE_NOT_APPLICABLE)
pids = tuple(sorted(qualifying))
if not rules.engagement_gate:
return EngagementVerdict(state=STATE_ENGAGED, qualifying=pids)
tally = CauseTally()
held, holder_pid = assess_screen(self._holder_path, tally)
focus_pid = assess_focus(self._probe, rules, qualifying, tally)
idle, controller_idle = assess_idle(
self._probe, self._controller, rules, tally, now_monotonic
)
return EngagementVerdict(
state=STATE_PAUSED if tally.causes else STATE_ENGAGED,
causes=tuple(tally.causes),
degraded=tuple(tally.degraded),
idle_seconds=idle,
controller_idle_seconds=controller_idle,
screen_held=held,
holder_pid=holder_pid,
focus_pid=focus_pid,
qualifying=pids,
)
def backdate(
self,
state: PlaytimeState,
verdict: EngagementVerdict,
*,
rules: PlaytimeRules,
) -> PlaytimeState:
"""Refund the idle grace period once per walk-away.
Time billed *before* the idle threshold tripped was not play either, so
the first paused tick of an idle-only spell gives it back. Deliberately
conservative: it fires only on the engaged-to-paused edge, only when
idleness is the sole cause, and never across a day rollover — and after
a daemon restart there is no previous verdict, so nothing is refunded.
Args:
state: Accounting state after this tick's accumulation.
verdict: This tick's verdict.
rules: Policy for this tick.
Returns:
The state, with the grace period deducted when all of that holds.
"""
previous, previous_day = self._previous, self._previous_day
self._previous, self._previous_day = verdict, state.day_key
if previous is None or previous.state != STATE_ENGAGED:
return state
if previous_day != state.day_key:
return state
if not verdict.idle_only():
return state
refunded = max(0.0, state.seconds - rules.idle_grace_seconds)
logger.info(
"Idle for %.0fs; refunding the %.0fs grace period.",
verdict.idle_seconds or 0.0,
state.seconds - refunded,
)
return replace(state, seconds=refunded)

View File

@ -0,0 +1,209 @@
"""A structured audit trail for every decision the gaming budget makes.
Reconstructing the incident that prompted this module — two hours billed while
a screen locker covered the display — took cross-referencing journald against
the screen locker's own JSONL, and even then the accrual could only be inferred
from the spacing of two warning notifications. That is not an audit trail.
So every budget decision is recorded here as one JSON object per line, readable
without ``sudo`` (unlike the state file, whose ``PermissionError`` is swallowed
on load and silently reported as "no state recorded yet").
Volume is controlled by only writing when something *changes*. At a three-second
tick, one line per tick would be 28 800 lines a day of almost entirely identical
records; instead a verdict is written when it differs from the last one, with a
periodic heartbeat so a long unchanging session still leaves a trace.
Nothing here may raise. A logging failure must not take down the enforcer, so
the sink degrades to a warning on the daemon's own logger and carries on.
"""
from __future__ import annotations
from datetime import datetime, timezone
import json
import logging
import logging.handlers
from pathlib import Path
from typing import TYPE_CHECKING, Final, cast
from steam_backlog_enforcer._engagement_types import STATE_NOT_APPLICABLE
if TYPE_CHECKING:
from steam_backlog_enforcer._engagement_types import EngagementVerdict
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
logger = logging.getLogger(__name__)
BUDGET_LOG_FILE: Final = Path("/var/log/steam-backlog-enforcer/budget.jsonl")
_MAX_BYTES: Final = 5 * 1024 * 1024
_BACKUP_COUNT: Final = 5
_HEARTBEAT_SECONDS: Final = 300.0
EVENT_VERDICT_CHANGE: Final = "verdict_change"
EVENT_HEARTBEAT: Final = "heartbeat"
EVENT_DETECTOR_FAILURE: Final = "detector_failure"
EVENT_MANUAL_ADJUSTMENT: Final = "manual_adjustment"
class BudgetLog:
"""Append-only JSONL sink, rotated by size."""
def __init__(self, *, path: Path = BUDGET_LOG_FILE) -> None:
"""Prepare a sink; the file is opened on first write.
Args:
path: Destination JSONL file.
"""
self._path = path
self._sink: logging.Logger | None = None
def _open(self) -> logging.Logger | None:
"""Return the rotating sink, creating it on first use.
Returns:
The sink, or ``None`` if it could not be created.
"""
if self._sink is not None:
return self._sink
try:
self._path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
self._path, maxBytes=_MAX_BYTES, backupCount=_BACKUP_COUNT
)
except OSError as exc:
logger.warning("Budget log unavailable at %s: %s", self._path, exc)
return None
handler.setFormatter(logging.Formatter("%(message)s"))
# Named per destination: a single shared logger would let two sinks
# with different paths silently evict each other's handler.
sink = logging.getLogger(f"{__name__}.sink.{self._path}")
sink.setLevel(logging.INFO)
sink.propagate = False
for stale in sink.handlers:
stale.close()
sink.handlers = [handler]
self._sink = sink
return self._sink
def close(self) -> None:
"""Close the underlying file handler."""
if self._sink is None:
return
for handler in self._sink.handlers:
handler.close()
self._sink.handlers = []
self._sink = None
def record(self, event: str, **fields: object) -> None:
"""Append one record.
Args:
event: Event name.
fields: Arbitrary JSON-serialisable payload.
"""
sink = self._open()
if sink is None:
return
payload = {
"timestamp": datetime.now(timezone.utc).astimezone().isoformat(),
"event": event,
**fields,
}
try:
sink.info(json.dumps(payload, default=str))
except (OSError, ValueError) as exc:
logger.warning("Could not write budget log record %r: %s", event, exc)
class TickJournal:
"""Decides which ticks are worth recording, and records them."""
def __init__(
self, log: BudgetLog, *, heartbeat: float = _HEARTBEAT_SECONDS
) -> None:
"""Start with nothing observed.
Args:
log: Sink to write through.
heartbeat: Seconds between heartbeats during an unchanged verdict.
"""
self._log = log
self._heartbeat = heartbeat
self._last_reason: str | None = None
self._last_written_at: float | None = None
def observe(
self,
verdict: EngagementVerdict,
state: PlaytimeState,
*,
rules: PlaytimeRules,
now_monotonic: float,
) -> None:
"""Record *verdict* if it changed, failed, or is due a heartbeat.
Args:
verdict: This tick's verdict.
state: Accounting state after this tick.
rules: Policy for this tick.
now_monotonic: Monotonic timestamp for heartbeat spacing.
"""
if verdict.degraded:
self._log.record(
EVENT_DETECTOR_FAILURE,
degraded=list(verdict.degraded),
billed=True,
**_snapshot(verdict, state, rules),
)
if verdict.reason != self._last_reason:
self._last_reason = verdict.reason
self._last_written_at = now_monotonic
self._log.record(EVENT_VERDICT_CHANGE, **_snapshot(verdict, state, rules))
return
if verdict.state == STATE_NOT_APPLICABLE:
return
# The change branch above always fires on the first observation and
# sets both fields together, so by here this is never None.
last = cast("float", self._last_written_at)
if now_monotonic - last < self._heartbeat:
return
self._last_written_at = now_monotonic
self._log.record(EVENT_HEARTBEAT, **_snapshot(verdict, state, rules))
def _snapshot(
verdict: EngagementVerdict, state: PlaytimeState, rules: PlaytimeRules
) -> dict[str, object]:
"""Flatten a tick into the fields every record carries.
The qualifying PIDs are included deliberately: the accrual that continued
after the last game exited could not be attributed without them.
Args:
verdict: This tick's verdict.
state: Accounting state after this tick.
rules: Policy for this tick.
Returns:
The record body.
"""
return {
"state": verdict.state,
"reason": verdict.reason,
"causes": list(verdict.causes),
"idle_seconds": verdict.idle_seconds,
"controller_idle_seconds": verdict.controller_idle_seconds,
"screen_held": verdict.screen_held,
"holder_pid": verdict.holder_pid,
"focus_pid": verdict.focus_pid,
"qualifying": list(verdict.qualifying),
"day_key": state.day_key,
"billed_seconds": round(state.seconds, 3),
"remaining_seconds": round(rules.budget_seconds - state.seconds, 3),
}

View File

@ -0,0 +1,83 @@
"""The cross-tick state the daily budget carries, and how it is built.
Two things cannot be recomputed from scratch each tick. The engagement backdate
must recognise an engaged-to-paused *edge*, so it needs the previous verdict;
and the audit journal only writes on change, so it needs the previous record.
Both live on a session that the enforce loop creates once and threads through,
rather than in module globals.
The session's fields are typed by :class:`Protocol` so a test can supply an
honest stub. A ``MagicMock`` would satisfy every attribute lookup — including
the ones a refactor forgot to update — which is exactly the failure a type
would otherwise have caught.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Protocol
from steam_backlog_enforcer._desktop_env import desktop_uid, resolve_desktop_user
from steam_backlog_enforcer._playtime_engagement import EngagementTracker
from steam_backlog_enforcer._playtime_log import BudgetLog, TickJournal
if TYPE_CHECKING:
from steam_backlog_enforcer._engagement_types import EngagementVerdict
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
class EngagementSource(Protocol):
"""What the budget tick needs from an engagement tracker."""
def assess(
self,
rules: PlaytimeRules,
*,
qualifying: set[int],
now_monotonic: float,
) -> EngagementVerdict:
"""Judge whether this tick counts against the budget."""
def backdate(
self,
state: PlaytimeState,
verdict: EngagementVerdict,
*,
rules: PlaytimeRules,
) -> PlaytimeState:
"""Refund the idle grace period on an engaged-to-idle edge."""
class TickRecorder(Protocol):
"""What the budget tick needs from the audit journal."""
def observe(
self,
verdict: EngagementVerdict,
state: PlaytimeState,
*,
rules: PlaytimeRules,
now_monotonic: float,
) -> None:
"""Record this tick if it is worth recording."""
@dataclass(frozen=True)
class PlaytimeSession:
"""Per-daemon engagement and logging state."""
tracker: EngagementSource
journal: TickRecorder
def new_session() -> PlaytimeSession:
"""Build the session the enforce loop threads through every tick.
Returns:
A session bound to the desktop user's X display and runtime dir.
"""
uid = desktop_uid(resolve_desktop_user())
return PlaytimeSession(
tracker=EngagementTracker(uid=uid),
journal=TickJournal(BudgetLog()),
)

View File

@ -90,6 +90,9 @@ class PlaytimeRules:
count_launchers: bool
enforcement: bool
demo: bool
engagement_gate: bool = True
idle_grace_seconds: float = 300.0
require_game_focus: bool = True
def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
@ -115,6 +118,9 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
count_launchers=config.count_launcher_processes,
enforcement=config.playtime_enforcement,
demo=demo,
engagement_gate=config.engagement_gate,
idle_grace_seconds=float(config.idle_grace_seconds),
require_game_focus=config.require_game_focus,
)

View File

@ -0,0 +1,124 @@
"""Detecting that a gatelock guard currently owns the screen.
The screen lockers (``screen_locker``, ``leetcode_guard``) coordinate through a
single advisory ``flock`` on ``$XDG_RUNTIME_DIR/gatelock/holder.lock``. The file's
*contents* cannot answer "is it held right now" — a stale record left by a dead
process is byte-identical to a live one — and gatelock's own
``read_claim_if_held`` answers the question by *taking* the lock.
The enforcer must not do that. Taking the lock every three seconds, even for
microseconds, would race a guard trying to claim the screen. ``/proc/locks``
reports the same fact by observation, so this reads that instead and never
touches the file.
Line format is ``idx: FLOCK ADVISORY WRITE <pid> <maj:min:inode> <start> <end>``,
with major and minor in hex. Lines containing ``->`` describe processes *waiting*
for a lock, not holding one, and are skipped.
"""
from __future__ import annotations
from dataclasses import dataclass
import os
from pathlib import Path
import re
from typing import Final
_PROC_LOCKS: Final = Path("/proc/locks")
_HOLDER_RELATIVE: Final = Path("gatelock/holder.lock")
# `maj:min:inode`, major and minor hex, inode decimal.
_DEVICE_FIELD: Final = re.compile(r"^[0-9a-f]+:[0-9a-f]+:\d+$")
_WAITER_MARKER: Final = "->"
_PID_FIELD_INDEX: Final = 4
class ScreenHoldError(RuntimeError):
"""Whether the screen is held could not be determined."""
@dataclass(frozen=True)
class ScreenHold:
"""Whether a guard owns the screen, and which process does."""
held: bool
holder_pid: int | None = None
def holder_lock_path(uid: int) -> Path:
"""Return the gatelock holder-lock path for *uid*.
Args:
uid: Owner of the desktop session.
Returns:
Path to ``holder.lock``.
"""
return Path(f"/run/user/{uid}") / _HOLDER_RELATIVE
def screen_hold(path: Path) -> ScreenHold:
"""Report whether *path*'s advisory lock is currently held.
Args:
path: The gatelock holder-lock file.
Returns:
The hold state, with the holding PID when there is one.
Raises:
ScreenHoldError: If ``/proc/locks`` cannot be read. A missing *path* is
not an error — it means no guard has ever run this boot.
"""
try:
stat = path.stat()
except OSError:
return ScreenHold(held=False)
target = f"{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}:{stat.st_ino}"
try:
raw = _PROC_LOCKS.read_text(encoding="utf-8")
except OSError as exc:
msg = f"cannot read {_PROC_LOCKS}: {exc}"
raise ScreenHoldError(msg) from exc
return _find_holder(raw, target)
def _find_holder(raw: str, target: str) -> ScreenHold:
"""Scan ``/proc/locks`` text for a live lock on *target*.
Args:
raw: Contents of ``/proc/locks``.
target: ``maj:min:inode`` string to match.
Returns:
The hold state.
"""
for line in raw.splitlines():
fields = line.split()
if _WAITER_MARKER in fields:
continue
if not any(field == target for field in fields if _DEVICE_FIELD.match(field)):
continue
return ScreenHold(held=True, holder_pid=_pid_of(fields))
return ScreenHold(held=False)
def _pid_of(fields: list[str]) -> int | None:
"""Return the PID column of a ``/proc/locks`` row.
Args:
fields: Whitespace-split row.
Returns:
The PID, or ``None`` if the row is shaped unexpectedly.
"""
if len(fields) <= _PID_FIELD_INDEX:
return None
try:
return int(fields[_PID_FIELD_INDEX])
except ValueError:
return None

View File

@ -38,3 +38,15 @@ class DesktopSessionNotReadyError(SteamUnavailableError):
that way for the whole session — the Kingdom Come: Deliverance II startup
crash. Waiting a pass costs 3s; getting it wrong costs a reboot.
"""
class GameInProgressError(SteamUnavailableError):
"""Raised to defer a Steam restart while a game is actually being played.
Subclasses :class:`SteamUnavailableError` so existing callers already
degrade gracefully (skip this pass, retry next loop). Bouncing Steam to
open the CDP port takes every running game down with it, losing unsaved
progress — observed on 2026-08-28, when restarting the daemon to deploy an
unrelated change killed a live session. Library hiding can wait; a player's
afternoon cannot be given back.
"""

View File

@ -28,9 +28,12 @@ from steam_backlog_enforcer._desktop_env import (
from steam_backlog_enforcer._steam_errors import (
DesktopSessionNotReadyError,
SteamUnavailableError,
SteamUpdateInProgressError,
)
from steam_backlog_enforcer._steam_process import _run_as_user
from steam_backlog_enforcer._steam_restart_guard import (
assert_safe_to_restart,
game_is_running,
)
from steam_backlog_enforcer._steam_state import steam_update_in_progress
logger = logging.getLogger(__name__)
@ -193,18 +196,7 @@ def ensure_steam_debug_port() -> None:
logger.info("Steam CDP port not available — (re)starting Steam...")
if _is_steam_running():
# Never bounce a running Steam while a game update is downloading or
# committing: the shutdown suspends it and can leave a partially
# written install (the root cause of the AoE2 launch crash). Defer and
# retry on the next enforce pass.
if steam_update_in_progress():
msg = (
"Deferring Steam restart: a game update is in progress. "
"Restarting now would interrupt and can corrupt it; will "
"retry once the update settles."
)
logger.info(msg)
raise SteamUpdateInProgressError(msg)
assert_safe_to_restart()
_shutdown_steam()
_launch_steam_with_debug()
@ -228,9 +220,17 @@ def ensure_steam_debug_port() -> None:
def restart_steam() -> None:
"""Gracefully restart the Steam client with CEF debugging enabled.
Skips the restart if a game update is downloading or committing, so the
update is not interrupted (interrupting it can corrupt the install).
Skips the restart while a game is running (the restart would kill it) or
while a game update is downloading or committing (interrupting it can
corrupt the install).
"""
if game_is_running():
logger.warning(
"Skipping Steam restart — a game is running; restarting now "
"would kill it and lose unsaved progress.",
)
return
if steam_update_in_progress():
logger.warning(
"Skipping Steam restart — a game update is in progress; "

View File

@ -0,0 +1,66 @@
"""When it is *not* safe to bounce a running Steam.
Opening the CDP debug port requires restarting Steam, and a restart is never
free: it takes down whatever the client is currently doing. Two situations make
that unacceptable, and both defer rather than abort — the enforce loop retries
every few seconds, so waiting costs a pass and nothing else.
Split out of :mod:`_steam_launch` to keep it under the 250-line cap, and because
"is a restart safe right now" is a question worth being able to ask on its own.
"""
from __future__ import annotations
import logging
from steam_backlog_enforcer._steam_errors import (
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__)
def game_is_running() -> bool:
"""Whether a Steam game is currently running.
Uses the same ``SteamAppId != 0`` predicate as the playtime budget, which
excludes the Steam client's own process tree — the client being up is not a
reason to defer, only an actual game is.
Returns:
Whether any game process is live.
"""
return any(app_id != 0 for app_id in get_running_steam_game_pids().values())
def assert_safe_to_restart() -> None:
"""Refuse to restart Steam while doing so would destroy something.
Raises:
GameInProgressError: If a game is running. Bouncing Steam kills it and
any unsaved progress with it — observed on 2026-08-28, when
restarting the daemon to deploy an unrelated change ended a live
session. Library hiding can wait; an afternoon cannot be returned.
SteamUpdateInProgressError: If a game update is downloading or
committing. The shutdown suspends it and can leave a partially
written install (the root cause of the AoE2 launch crash).
"""
if game_is_running():
msg = (
"Deferring Steam restart: a game is running. Restarting now would "
"kill it and lose unsaved progress; will retry once it exits."
)
logger.info(msg)
raise GameInProgressError(msg)
if steam_update_in_progress():
msg = (
"Deferring Steam restart: a game update is in progress. Restarting "
"now would interrupt and can corrupt it; will retry once the "
"update settles."
)
logger.info(msg)
raise SteamUpdateInProgressError(msg)

View File

@ -0,0 +1,176 @@
"""Resolving which process owns the focused X window.
``_NET_WM_PID`` is the obvious answer and the one every window-manager guide
names, but it is a *convention* toolkits opt into, not a guarantee. Measured on
this machine: Aseprite's window carries no ``_NET_WM_PID`` at all. Treating that
as "cannot tell" and billing the tick would make the focus criterion useless for
every such application — including, plausibly, games.
So the absent property falls through to the X-Resource extension, which asks the
*server* which client owns the window rather than trusting the client to have
announced itself. Verified against the same Aseprite window: ``XResQueryClientIds``
returned the correct PID where ``_NET_WM_PID`` returned nothing.
"""
from __future__ import annotations
import ctypes
from typing import Final
from steam_backlog_enforcer._x_probe import Conn, XProbeError, root_window
# XGetWindowProperty's `req_type`: accept whatever type the property carries.
_ANY_PROPERTY_TYPE: Final = 0
_SUCCESS: Final = 0
# XInternAtom's `only_if_exists`: never create an atom just by asking for it.
# An int, not a bool: it is a C ``Bool`` argument, not a Python flag.
_ONLY_IF_EXISTS: Final = 1
# XRes client-id spec masks, from <X11/extensions/XRes.h>.
_XRES_PID_MASK: Final = 1 << 1
_XRES_PID_BYTES: Final = 4
class _XResSpec(ctypes.Structure):
"""Mirror of libXRes's ``XResClientIdSpec``."""
_fields_ = (("client", ctypes.c_ulong), ("mask", ctypes.c_uint))
class _XResValue(ctypes.Structure):
"""Mirror of libXRes's ``XResClientIdValue``."""
_fields_ = (
("spec", _XResSpec),
("length", ctypes.c_long),
("value", ctypes.c_void_p),
)
def focused_pid(conn: Conn) -> int | None:
"""Return the PID owning the focused window.
Args:
conn: Live connection.
Returns:
The PID, or ``None`` when nothing is focused — a real answer, not a
failure.
Raises:
XProbeError: If a window is focused but cannot be attributed to a
process, which the caller must treat as "cannot tell" and bill.
"""
active = read_cardinal(conn, root_window(conn), b"_NET_ACTIVE_WINDOW")
if not active:
return None
declared = read_cardinal(conn, active, b"_NET_WM_PID")
if declared is not None:
return declared
resolved = _xres_pid(conn, active)
if resolved is None:
msg = f"window {active:#x} has no _NET_WM_PID and XRes could not name it"
raise XProbeError(msg)
return resolved
def _xres_pid(conn: Conn, window: int) -> int | None:
"""Ask the X server which process owns *window*.
Args:
conn: Live connection.
window: Window to attribute.
Returns:
The owning PID, or ``None`` if XRes is unavailable or has no answer.
"""
if conn.xres is None:
return None
spec = _XResSpec(client=window, mask=_XRES_PID_MASK)
count = ctypes.c_long()
values = ctypes.POINTER(_XResValue)()
status = conn.xres.XResQueryClientIds(
conn.display,
ctypes.c_long(1),
ctypes.byref(spec),
ctypes.byref(count),
ctypes.byref(values),
)
if status != _SUCCESS or not values:
return None
try:
return _first_pid(values, count.value)
finally:
conn.xres.XResClientIdsDestroy(count, values)
def _first_pid(values: ctypes._Pointer[_XResValue], count: int) -> int | None:
"""Return the first PID among *count* XRes client-id values.
Args:
values: Array returned by ``XResQueryClientIds``.
count: Number of entries in *values*.
Returns:
The PID, or ``None`` if no entry carries one.
"""
for index in range(count):
entry = values[index]
if entry.spec.mask != _XRES_PID_MASK or entry.length < _XRES_PID_BYTES:
continue
if not entry.value:
continue
return int(ctypes.cast(entry.value, ctypes.POINTER(ctypes.c_uint32))[0])
return None
def read_cardinal(conn: Conn, window: int, name: bytes) -> int | None:
"""Read the first 32-bit value of window property *name*.
Args:
conn: Live connection.
window: Window to query.
name: Property name, e.g. ``b"_NET_WM_PID"``.
Returns:
The value, or ``None`` if the property is absent or empty.
"""
conn.x11.XInternAtom.restype = ctypes.c_ulong
atom = conn.x11.XInternAtom(conn.display, name, _ONLY_IF_EXISTS)
if not atom:
return None
actual_type = ctypes.c_ulong()
actual_format = ctypes.c_int()
nitems = ctypes.c_ulong()
bytes_after = ctypes.c_ulong()
prop = ctypes.POINTER(ctypes.c_ulong)()
status = conn.x11.XGetWindowProperty(
conn.display,
ctypes.c_ulong(window),
ctypes.c_ulong(atom),
ctypes.c_long(0),
ctypes.c_long(1),
ctypes.c_int(0),
ctypes.c_ulong(_ANY_PROPERTY_TYPE),
ctypes.byref(actual_type),
ctypes.byref(actual_format),
ctypes.byref(nitems),
ctypes.byref(bytes_after),
ctypes.byref(prop),
)
if status != _SUCCESS or not prop:
return None
try:
if nitems.value == 0:
return None
return int(prop[0])
finally:
conn.x11.XFree(prop)

View File

@ -0,0 +1,185 @@
"""An X11 connection for reading input-idle time, without forking.
The enforce loop ticks every three seconds, so shelling out to ``xprintidle``
would cost roughly twenty forks a minute for a number the X server hands over
directly. This talks to libX11 and libXss through ``ctypes`` over one cached
connection instead. Focused-window resolution lives in :mod:`_x_focus`.
The daemon runs as root with ``DISPLAY=:0`` and ``HOME=/home/kuhy`` set by its
unit file, which is what lets it authenticate against the user's session.
Two details are load-bearing:
**A no-op X error handler is installed.** Xlib's default handler calls
``exit(1)``. Asking about a window that vanished mid-query is an ordinary race,
not a fatal condition, and the enforcer must not die of it.
**Every failure raises :class:`XProbeError`.** The caller bills the tick when a
probe cannot answer, so returning a bland ``None`` would be indistinguishable
from a real "not idle" and would quietly become free gaming.
"""
from __future__ import annotations
import ctypes
from dataclasses import dataclass
from typing import Final
_MILLISECONDS_PER_SECOND: Final = 1000.0
class XProbeError(RuntimeError):
"""The X server could not be reached or did not answer."""
class _ScreenSaverInfo(ctypes.Structure):
"""Mirror of libXss's ``XScreenSaverInfo``."""
_fields_ = (
("window", ctypes.c_ulong),
("state", ctypes.c_int),
("kind", ctypes.c_int),
("til_or_since", ctypes.c_ulong),
("idle", ctypes.c_ulong),
("event_mask", ctypes.c_ulong),
)
@dataclass(frozen=True)
class Conn:
"""An open display and the libraries it was opened with.
``xres`` is optional: the X-Resource extension is the fallback for windows
that advertise no ``_NET_WM_PID``, and a server without it simply means
those windows cannot be attributed.
"""
display: ctypes.c_void_p
x11: ctypes.CDLL
xss: ctypes.CDLL
xres: ctypes.CDLL | None
_ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)
def _ignore_x_error(_display: ctypes.c_void_p, _event: ctypes.c_void_p) -> int:
"""Swallow an X protocol error instead of letting Xlib call ``exit``."""
return 0
def load_libraries() -> tuple[ctypes.CDLL, ctypes.CDLL, ctypes.CDLL | None]:
"""Load libX11, libXss and — if present — libXRes.
Returns:
The ``(x11, xss, xres)`` handles; ``xres`` is ``None`` when absent.
Raises:
XProbeError: If libX11 or libXss is missing.
"""
try:
x11 = ctypes.CDLL("libX11.so.6")
xss = ctypes.CDLL("libXss.so.1")
except OSError as exc:
msg = f"X client libraries unavailable: {exc}"
raise XProbeError(msg) from exc
try:
xres: ctypes.CDLL | None = ctypes.CDLL("libXRes.so.1")
except OSError:
xres = None
return x11, xss, xres
Libraries = tuple[ctypes.CDLL, ctypes.CDLL, ctypes.CDLL | None]
class XProbe:
"""A cached X connection answering idle-time queries."""
def __init__(self, *, libraries: Libraries | None = None) -> None:
"""Store the library handles; the connection is opened lazily.
Args:
libraries: Pre-loaded ``(x11, xss, xres)`` handles, for tests.
"""
self._libraries = libraries
self._conn: Conn | None = None
# Xlib keeps only a borrowed pointer, so the trampoline must outlive it.
self._handler = _ERROR_HANDLER(_ignore_x_error)
def connect(self) -> Conn:
"""Return the open connection, opening it on first use.
Returns:
The live connection.
Raises:
XProbeError: If the display cannot be opened.
"""
if self._conn is not None:
return self._conn
if self._libraries is None:
self._libraries = load_libraries()
x11, xss, xres = self._libraries
x11.XSetErrorHandler(self._handler)
x11.XOpenDisplay.restype = ctypes.c_void_p
display = x11.XOpenDisplay(None)
if not display:
msg = "XOpenDisplay returned NULL; no reachable X session"
raise XProbeError(msg)
self._conn = Conn(display=ctypes.c_void_p(display), x11=x11, xss=xss, xres=xres)
return self._conn
def close(self) -> None:
"""Drop the connection so the next query reconnects."""
if self._conn is not None:
self._conn.x11.XCloseDisplay(self._conn.display)
self._conn = None
def idle_seconds(self) -> float:
"""Return seconds since the last keyboard or pointer input.
Note that a game controller does **not** reset this counter unless it
is presented to X as a keyboard or pointer, which is why the caller
drops the idle criterion while a controller is connected.
Returns:
The idle time in seconds.
Raises:
XProbeError: If the query fails.
"""
conn = self.connect()
conn.xss.XScreenSaverAllocInfo.restype = ctypes.POINTER(_ScreenSaverInfo)
info = conn.xss.XScreenSaverAllocInfo()
if not info:
msg = "XScreenSaverAllocInfo returned NULL"
raise XProbeError(msg)
try:
queried = conn.xss.XScreenSaverQueryInfo(
conn.display, ctypes.c_ulong(root_window(conn)), info
)
if not queried:
msg = "XScreenSaverQueryInfo failed; MIT-SCREEN-SAVER missing?"
raise XProbeError(msg)
return float(info.contents.idle) / _MILLISECONDS_PER_SECOND
finally:
conn.x11.XFree(info)
def root_window(conn: Conn) -> int:
"""Return the default root window id.
Args:
conn: Live connection.
Returns:
The root window id.
"""
conn.x11.XDefaultRootWindow.restype = ctypes.c_ulong
return int(conn.x11.XDefaultRootWindow(conn.display))

View File

@ -85,6 +85,14 @@ class Config:
Disabling stops the cutoff but never strands a live block — the release
path runs regardless, so "disabled" cannot come to mean "blocked forever".
"""
# Engagement gating: without it a resident game process bills the budget
# while the screen is locked. See :mod:`_playtime_engagement`.
engagement_gate: bool = True
"""Whether a tick must show active engagement before it bills."""
idle_grace_seconds: int = 300
"""Input silence before billing stops; controller input counts as input."""
require_game_focus: bool = True
"""Whether the focused window must belong to the game for a tick to bill."""
def save(self) -> None:
"""Persist config to disk."""

View File

@ -0,0 +1,123 @@
"""A :class:`PlaytimeSession` stub for tests that are not about engagement.
Most budget tests predate the engagement gate and care only about accounting,
warnings and the cutoff. They get a session that always says "engaged", which
reproduces the pre-gate behaviour exactly, so those tests keep asserting what
they were written to assert.
Deliberately hand-written rather than ``MagicMock``: a mock satisfies every
attribute lookup, including ones a later refactor renamed, so a stub that must
actually match the protocol is the point.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from steam_backlog_enforcer._engagement_types import (
STATE_ENGAGED,
STATE_PAUSED,
EngagementVerdict,
)
from steam_backlog_enforcer._playtime_session import PlaytimeSession
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
class StubTracker:
"""An engagement tracker with a fixed answer and no probes."""
def __init__(self, *, engaged: bool = True) -> None:
"""Record the answer this tracker always gives.
Args:
engaged: Whether every tick should count.
"""
self.engaged = engaged
self.verdicts: list[EngagementVerdict] = []
def assess(
self,
rules: PlaytimeRules,
*,
qualifying: set[int],
now_monotonic: float,
) -> EngagementVerdict:
"""Return the fixed verdict.
Args:
rules: Ignored.
qualifying: PIDs recorded on the verdict.
now_monotonic: Ignored.
Returns:
The verdict.
"""
del rules, now_monotonic
verdict = EngagementVerdict(
state=STATE_ENGAGED if self.engaged else STATE_PAUSED,
causes=() if self.engaged else ("focus",),
qualifying=tuple(sorted(qualifying)),
)
self.verdicts.append(verdict)
return verdict
def backdate(
self,
state: PlaytimeState,
verdict: EngagementVerdict,
*,
rules: PlaytimeRules,
) -> PlaytimeState:
"""Return *state* untouched.
Args:
state: Accounting state.
verdict: Ignored.
rules: Ignored.
Returns:
The unchanged state.
"""
del verdict, rules
return state
class StubJournal:
"""A journal that keeps records in memory instead of on disk."""
def __init__(self) -> None:
"""Start with nothing recorded."""
self.observed: list[EngagementVerdict] = []
def observe(
self,
verdict: EngagementVerdict,
state: PlaytimeState,
*,
rules: PlaytimeRules,
now_monotonic: float,
) -> None:
"""Remember *verdict*.
Args:
verdict: This tick's verdict.
state: Ignored.
rules: Ignored.
now_monotonic: Ignored.
"""
del state, rules, now_monotonic
self.observed.append(verdict)
def fake_session(*, engaged: bool = True) -> PlaytimeSession:
"""Build a session that never touches X, /proc/locks or the log file.
Args:
engaged: Whether every tick should count against the budget.
Returns:
The session.
"""
return PlaytimeSession(tracker=StubTracker(engaged=engaged), journal=StubJournal())

View File

@ -0,0 +1,99 @@
"""Fake libX11 / libXss / libXRes handles so no test needs an X server.
The real modules talk to X through ``ctypes``, which means the seam worth
faking is the shared-library handle itself. These stand-ins return the same
kinds of values Xlib does — an integer display pointer, a struct pointer, a
status code — and, where the C API writes through a pointer argument, actually
write through it, so the calling code is exercised rather than bypassed.
"""
from __future__ import annotations
import ctypes
from typing import TYPE_CHECKING
from unittest.mock import MagicMock
if TYPE_CHECKING:
from collections.abc import Callable
def fake_x11(*, display: int = 0x1234, root: int = 0x50, atom: int = 7) -> MagicMock:
"""Build a libX11 stand-in.
Args:
display: Value ``XOpenDisplay`` returns; ``0`` means "cannot connect".
root: Value ``XDefaultRootWindow`` returns.
atom: Value ``XInternAtom`` returns; ``0`` means "no such atom".
Returns:
The fake handle.
"""
lib = MagicMock(name="libX11")
lib.XOpenDisplay.return_value = display
lib.XDefaultRootWindow.return_value = root
lib.XInternAtom.return_value = atom
return lib
def fake_xss(*, idle_ms: int = 0, allocated: bool = True, ok: bool = True) -> MagicMock:
"""Build a libXss stand-in.
Args:
idle_ms: Idle milliseconds reported by ``XScreenSaverQueryInfo``.
allocated: Whether ``XScreenSaverAllocInfo`` returns a struct.
ok: Whether ``XScreenSaverQueryInfo`` succeeds.
Returns:
The fake handle.
"""
lib = MagicMock(name="libXss")
if allocated:
info = MagicMock(name="XScreenSaverInfo")
info.contents.idle = idle_ms
lib.XScreenSaverAllocInfo.return_value = info
else:
lib.XScreenSaverAllocInfo.return_value = 0
lib.XScreenSaverQueryInfo.return_value = 1 if ok else 0
return lib
def property_writer(
*, status: int = 0, nitems: int = 1, value: int = 4321, null: bool = False
) -> Callable[..., int]:
"""Build an ``XGetWindowProperty`` stand-in that writes its out-params.
Args:
status: Return code; non-zero means failure.
nitems: Number of items to report.
value: The first 32-bit value to hand back.
null: Whether to leave the property pointer NULL.
Returns:
A callable with ``XGetWindowProperty``'s signature.
"""
# Kept alive for the lifetime of the fake: the pointer written into the
# caller's out-param must not dangle.
keepalive: list[ctypes.c_ulong] = []
def impl( # signature mirrors the C function
_display: object,
_window: object,
_atom: object,
_offset: object,
_length: object,
_delete: object,
_req_type: object,
_actual_type: object,
_actual_format: object,
nitems_ref: object,
_bytes_after: object,
prop_ref: object,
) -> int:
nitems_ref._obj.value = nitems # type-erased ctypes byref holder
if not null:
holder = ctypes.c_ulong(value)
keepalive.append(holder)
prop_ref._obj.contents = holder
return status
return impl

View File

@ -201,3 +201,24 @@ def total_block_paths(tmp_path: Path) -> Iterator[Paths]:
patch(f"{PURGE}._STEAM_REMNANT_PATHS", built.remnant_paths),
):
yield built
# ──────────────────────────────────────────────────────────────
# Live process-table isolation
#
# The Steam-restart guard asks whether a game is running by scanning
# /proc for SteamAppId. Unpatched, that reads the *developer's* real
# process table: running the suite while playing a game made three
# restart tests fail on 2026-08-28. Tests that care about the guard
# patch this themselves.
# ──────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def _no_live_games() -> Iterator[None]:
"""Report no running games unless a test says otherwise."""
with patch(
"steam_backlog_enforcer._steam_restart_guard.get_running_steam_game_pids",
return_value={},
):
yield

View File

@ -0,0 +1,137 @@
"""Tests for noticing real gamepad input.
The distinction that matters: this machine exposes an *ASRock LED Controller*
as ``/dev/input/js0``, so mere presence of a js device must not register as
play — only actual, non-synthetic events do.
"""
from __future__ import annotations
import errno
import os
from typing import TYPE_CHECKING
from unittest.mock import patch
from steam_backlog_enforcer._controller_input import ControllerActivity
if TYPE_CHECKING:
from pathlib import Path
CI = "steam_backlog_enforcer._controller_input"
def _event(*, init: bool) -> bytes:
"""Build one 8-byte ``struct js_event``; byte 6 is the type field."""
return bytes([0, 0, 0, 0, 0, 0, 0x81 if init else 0x01, 0])
class TestSyncDevices:
def test_missing_directory_is_not_an_error(self, tmp_path: Path) -> None:
activity = ControllerActivity(device_dir=tmp_path / "absent")
activity.poll(now=1.0)
assert activity.seconds_since_activity(now=1.0) is None
def test_non_js_entries_are_ignored(self, tmp_path: Path) -> None:
(tmp_path / "event0").write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=1.0)
assert activity.seconds_since_activity(now=1.0) is None
def test_a_vanished_device_is_forgotten(self, tmp_path: Path) -> None:
node = tmp_path / "js0"
node.write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=1.0)
node.unlink()
activity.poll(now=2.0)
activity.close()
def test_an_unopenable_device_is_skipped(self, tmp_path: Path) -> None:
node = tmp_path / "js0"
node.write_bytes(_event(init=False))
with patch(f"{CI}.os.open", side_effect=OSError("denied")):
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=1.0)
assert activity.seconds_since_activity(now=1.0) is None
class TestDrain:
def test_a_real_event_is_activity(self, tmp_path: Path) -> None:
(tmp_path / "js0").write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=12.5) == 2.5
activity.close()
def test_the_synthetic_open_burst_is_not_activity(self, tmp_path: Path) -> None:
# joydev replays current state on open; treating that as a button press
# would make simply having an LED device look like a gaming session.
(tmp_path / "js0").write_bytes(_event(init=True) * 3)
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=10.0) is None
activity.close()
def test_a_truncated_record_stops_the_drain(self, tmp_path: Path) -> None:
(tmp_path / "js0").write_bytes(b"\x00\x00\x00")
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=10.0) is None
activity.close()
def test_the_per_tick_cap_bounds_the_drain(self, tmp_path: Path) -> None:
(tmp_path / "js0").write_bytes(_event(init=False) * 200)
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=10.0) == 0.0
activity.close()
def test_eagain_leaves_the_device_open(self, tmp_path: Path) -> None:
# "No events pending" is the normal case on a quiet controller; it must
# not be mistaken for the device going away.
(tmp_path / "js0").write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
with patch(f"{CI}.os.read", side_effect=OSError(errno.EAGAIN, "again")):
activity.poll(now=10.0)
assert list(activity._fds) == [tmp_path / "js0"]
assert activity.seconds_since_activity(now=10.0) is None
activity.close()
def test_an_empty_device_yields_nothing(self, tmp_path: Path) -> None:
fifo = tmp_path / "js0"
os.mkfifo(fifo)
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=10.0) is None
activity.close()
def test_a_device_that_stays_is_kept_open_across_ticks(
self, tmp_path: Path
) -> None:
node = tmp_path / "js0"
node.write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
first = dict(activity._fds)
activity.poll(now=11.0)
assert activity._fds == first
activity.close()
def test_forgetting_an_unknown_device_is_a_no_op(self, tmp_path: Path) -> None:
activity = ControllerActivity(device_dir=tmp_path)
activity._forget(tmp_path / "js9")
assert activity._fds == {}
def test_a_hard_read_error_drops_the_device(self, tmp_path: Path) -> None:
(tmp_path / "js0").write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
with patch(f"{CI}.os.read", side_effect=OSError(errno.EIO, "gone")):
activity.poll(now=10.0)
assert activity.seconds_since_activity(now=10.0) is None
def test_forgetting_survives_an_already_closed_fd(self, tmp_path: Path) -> None:
(tmp_path / "js0").write_bytes(_event(init=False))
activity = ControllerActivity(device_dir=tmp_path)
activity.poll(now=10.0)
with patch(f"{CI}.os.close", side_effect=OSError("bad fd")):
activity.close()

View File

@ -11,6 +11,7 @@ from steam_backlog_enforcer._enforce_loop import (
_enforce_loop_iteration,
)
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.tests._fake_session import fake_session
if TYPE_CHECKING:
from collections.abc import Iterator
@ -48,7 +49,7 @@ class TestEnforceLoopIteration:
patch(f"{PKG}._echo"),
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
def test_no_kill(self) -> None:
config = Config(
@ -60,7 +61,7 @@ class TestEnforceLoopIteration:
patch(f"{PKG}.enforce_allowed_game") as mock_enforce,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_enforce.assert_not_called()
def test_guards_installed(self) -> None:
@ -74,7 +75,7 @@ class TestEnforceLoopIteration:
patch(f"{PKG}._echo"),
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
def test_guard_removes_zero(self) -> None:
config = Config(
@ -86,7 +87,7 @@ class TestEnforceLoopIteration:
patch(f"{PKG}._guard_installed_games", return_value=0),
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
def test_skips_when_steam_absent(self) -> None:
"""With Steam uninstalled the iteration must do nothing.
@ -106,7 +107,7 @@ class TestEnforceLoopIteration:
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed,
patch(f"{ENFORCE_STEPS_PKG}.install_game") as mock_install,
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_enforce.assert_not_called()
mock_guard.assert_not_called()
@ -123,7 +124,7 @@ class TestEnforceLoopIteration:
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=False),
patch(f"{ENFORCE_STEPS_PKG}.install_game") as mock_install,
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_install.assert_called_once()
def test_no_app_id_skip_reinstall(self) -> None:
@ -137,7 +138,7 @@ class TestEnforceLoopIteration:
patch(f"{PKG}._guard_installed_games") as mock_guard,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed,
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_enforce.assert_not_called()
mock_guard.assert_not_called()
mock_installed.assert_not_called()

View File

@ -9,6 +9,7 @@ from steam_backlog_enforcer._enforce_loop import (
do_enforce,
)
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.tests._fake_session import fake_session
PKG = "steam_backlog_enforcer._enforce_loop"
ENFORCE_STEPS_PKG = "steam_backlog_enforcer._enforce_steps"
@ -27,7 +28,7 @@ class TestEnforceLoopIterationTotalBlock:
patch(f"{PKG}._guard_installed_games") as mock_guard,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed,
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_tick.assert_called_once()
mock_enforce.assert_not_called()
mock_guard.assert_not_called()
@ -42,7 +43,7 @@ class TestEnforceLoopIterationTotalBlock:
patch(f"{PKG}.enforce_total_block_tick") as mock_tick,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_tick.assert_not_called()
def test_expired_lock_triggers_cleanup_once(self) -> None:
@ -54,7 +55,7 @@ class TestEnforceLoopIterationTotalBlock:
patch(f"{PKG}.end_total_block_cleanup") as mock_cleanup,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_cleanup.assert_called_once()
def test_no_lock_no_cleanup_call(self) -> None:
@ -66,7 +67,7 @@ class TestEnforceLoopIterationTotalBlock:
patch(f"{PKG}.end_total_block_cleanup") as mock_cleanup,
patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_cleanup.assert_not_called()
@ -145,7 +146,7 @@ class TestPlaytimeIsEnforcedFirst:
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
mock_playtime.assert_called_once()
def test_receives_the_loop_interval(self) -> None:
@ -156,7 +157,7 @@ class TestPlaytimeIsEnforcedFirst:
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state)
_enforce_loop_iteration(config, state, session=fake_session())
assert mock_playtime.call_args.kwargs["interval"] == 3
assert mock_playtime.call_args.kwargs["demo"] is False
@ -168,7 +169,7 @@ class TestPlaytimeIsEnforcedFirst:
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.enforce_total_block_tick"),
):
_enforce_loop_iteration(config, state, demo=True)
_enforce_loop_iteration(config, state, session=fake_session(), demo=True)
assert mock_playtime.call_args.kwargs["demo"] is True
def test_do_enforce_plumbs_demo_into_the_iteration(self) -> None:

View File

@ -0,0 +1,155 @@
"""Tests for turning each engagement signal into a pause cause.
The rule under test everywhere: a probe that *fails* must not pause the tick.
Failing closed keeps a broken detector from becoming free gaming.
"""
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer._engagement_probes import (
assess_focus,
assess_idle,
assess_screen,
focus_qualifies,
)
from steam_backlog_enforcer._engagement_types import CauseTally
from steam_backlog_enforcer._playtime_state import PlaytimeRules
from steam_backlog_enforcer._screen_hold import ScreenHold, ScreenHoldError
from steam_backlog_enforcer._x_probe import XProbeError
EP = "steam_backlog_enforcer._engagement_probes"
RULES = PlaytimeRules(
budget_seconds=28800.0,
warn_at=(3600,),
sigkill_after=30.0,
count_launchers=True,
enforcement=True,
demo=False,
engagement_gate=True,
idle_grace_seconds=300.0,
require_game_focus=True,
)
def _probe(idle: float = 0.0) -> MagicMock:
probe = MagicMock(name="XProbe")
probe.idle_seconds.return_value = idle
return probe
class TestAssessScreen:
def test_a_held_screen_pauses(self) -> None:
tally = CauseTally()
held = ScreenHold(held=True, holder_pid=99)
with patch(f"{EP}.screen_hold", return_value=held):
assert assess_screen(Path("x"), tally) == (True, 99)
assert tally.causes == ["screen_held"]
assert tally.degraded == []
def test_a_free_screen_does_not_pause(self) -> None:
tally = CauseTally()
with patch(f"{EP}.screen_hold", return_value=ScreenHold(held=False)):
assert assess_screen(Path("x"), tally) == (False, None)
assert tally.causes == []
def test_a_failed_probe_bills_and_is_recorded(self) -> None:
tally = CauseTally()
with patch(f"{EP}.screen_hold", side_effect=ScreenHoldError("boom")):
assert assess_screen(Path("x"), tally) == (None, None)
assert tally.causes == []
assert tally.degraded == ["screen_held"]
class TestAssessFocus:
def test_disabled_criterion_reports_nothing(self) -> None:
tally = CauseTally()
relaxed = replace(RULES, require_game_focus=False)
assert assess_focus(_probe(), relaxed, {1}, tally) is None
assert tally.causes == []
def test_the_game_owning_focus_does_not_pause(self) -> None:
tally = CauseTally()
with patch(f"{EP}.focused_pid", return_value=42):
assert assess_focus(_probe(), RULES, {42}, tally) == 42
assert tally.causes == []
def test_another_window_owning_focus_pauses(self) -> None:
tally = CauseTally()
with patch(f"{EP}.focused_pid", return_value=7):
assert assess_focus(_probe(), RULES, {42}, tally) == 7
assert tally.causes == ["focus"]
def test_a_failed_probe_bills_and_drops_the_connection(self) -> None:
tally = CauseTally()
probe = _probe()
with patch(f"{EP}.focused_pid", side_effect=XProbeError("no display")):
assert assess_focus(probe, RULES, {42}, tally) is None
assert tally.causes == []
assert tally.degraded == ["focus"]
probe.close.assert_called_once()
class TestAssessIdle:
def test_recent_input_does_not_pause(self) -> None:
tally = CauseTally()
controller = MagicMock()
controller.seconds_since_activity.return_value = None
assert assess_idle(_probe(1.0), controller, RULES, tally, 5.0) == (1.0, None)
assert tally.causes == []
def test_silence_past_the_grace_pauses(self) -> None:
tally = CauseTally()
controller = MagicMock()
controller.seconds_since_activity.return_value = None
assess_idle(_probe(400.0), controller, RULES, tally, 5.0)
assert tally.causes == ["idle"]
def test_recent_controller_input_keeps_the_tick_alive(self) -> None:
# The X idle counter cannot see a gamepad; without this, a controller
# session would pause the budget and hand out free gaming.
tally = CauseTally()
controller = MagicMock()
controller.seconds_since_activity.return_value = 2.0
assert assess_idle(_probe(9999.0), controller, RULES, tally, 5.0) == (2.0, 2.0)
assert tally.causes == []
def test_a_failed_probe_bills_and_drops_the_connection(self) -> None:
tally = CauseTally()
probe = _probe()
probe.idle_seconds.side_effect = XProbeError("no display")
controller = MagicMock()
controller.seconds_since_activity.return_value = 7.0
assert assess_idle(probe, controller, RULES, tally, 5.0) == (None, 7.0)
assert tally.causes == []
assert tally.degraded == ["idle"]
probe.close.assert_called_once()
class TestFocusQualifies:
def test_nothing_focused_does_not_qualify(self) -> None:
assert focus_qualifies(None, {1234}) is False
def test_the_process_itself_qualifies(self) -> None:
assert focus_qualifies(1234, {1234}) is True
def test_a_descendant_window_qualifies(self) -> None:
# Proton games commonly own their window from a child process.
with patch(f"{EP}._read_ppid", side_effect=[500, 42]):
assert focus_qualifies(9000, {42}) is True
def test_reaching_init_does_not_qualify(self) -> None:
with patch(f"{EP}._read_ppid", return_value=1):
assert focus_qualifies(9000, {42}) is False
def test_an_unreadable_parent_does_not_qualify(self) -> None:
with patch(f"{EP}._read_ppid", return_value=None):
assert focus_qualifies(9000, {42}) is False
def test_a_pathological_chain_terminates(self) -> None:
with patch(f"{EP}._read_ppid", side_effect=lambda pid: pid + 1):
assert focus_qualifies(9000, {42}) is False

View File

@ -0,0 +1,50 @@
"""Tests for the engagement verdict value object."""
from __future__ import annotations
from steam_backlog_enforcer._engagement_types import (
CAUSE_FOCUS,
CAUSE_IDLE,
STATE_ENGAGED,
STATE_NOT_APPLICABLE,
STATE_PAUSED,
EngagementVerdict,
)
class TestEngaged:
def test_engaged_state_bills(self) -> None:
assert EngagementVerdict(state=STATE_ENGAGED).engaged is True
def test_nothing_qualifying_still_bills(self) -> None:
# Nothing accrues anyway; calling it "paused" would fill the audit log
# with a pause reason during ordinary desktop use.
assert EngagementVerdict(state=STATE_NOT_APPLICABLE).engaged is True
def test_paused_state_does_not_bill(self) -> None:
assert EngagementVerdict(state=STATE_PAUSED).engaged is False
class TestReason:
def test_falls_back_to_the_state_when_there_are_no_causes(self) -> None:
assert EngagementVerdict(state=STATE_ENGAGED).reason == STATE_ENGAGED
def test_joins_every_cause(self) -> None:
verdict = EngagementVerdict(
state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS)
)
assert verdict.reason == "idle+focus"
class TestIdleOnly:
def test_true_for_a_sole_idle_cause(self) -> None:
assert EngagementVerdict(state=STATE_PAUSED, causes=(CAUSE_IDLE,)).idle_only()
def test_false_when_something_else_also_paused_the_tick(self) -> None:
verdict = EngagementVerdict(
state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS)
)
assert verdict.idle_only() is False
def test_false_when_engaged(self) -> None:
assert EngagementVerdict(state=STATE_ENGAGED).idle_only() is False

View File

@ -0,0 +1,201 @@
"""Tests for the per-tick engagement verdict and the idle backdate.
The backdate is deliberately hard to trigger: it is the only place the budget
counter ever moves backwards, so every guard around it is tested here.
"""
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer._engagement_types import (
CAUSE_FOCUS,
CAUSE_IDLE,
STATE_ENGAGED,
STATE_NOT_APPLICABLE,
STATE_PAUSED,
CauseTally,
EngagementVerdict,
)
from steam_backlog_enforcer._playtime_engagement import EngagementTracker
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
PE = "steam_backlog_enforcer._playtime_engagement"
RULES = PlaytimeRules(
budget_seconds=28800.0,
warn_at=(3600,),
sigkill_after=30.0,
count_launchers=True,
enforcement=True,
demo=False,
engagement_gate=True,
idle_grace_seconds=300.0,
require_game_focus=True,
)
def _tracker(**kwargs: object) -> EngagementTracker:
return EngagementTracker(
uid=1000,
probe=MagicMock(name="XProbe"),
controller=MagicMock(name="ControllerActivity"),
holder_path=Path("/nonexistent/holder.lock"),
**kwargs,
)
def _engaged() -> EngagementVerdict:
return EngagementVerdict(state=STATE_ENGAGED)
def _idle_paused() -> EngagementVerdict:
return EngagementVerdict(state=STATE_PAUSED, causes=(CAUSE_IDLE,))
class TestAssess:
def test_nothing_qualifying_is_not_applicable(self) -> None:
verdict = _tracker().assess(RULES, qualifying=set(), now_monotonic=1.0)
assert verdict.state == STATE_NOT_APPLICABLE
assert verdict.engaged is True
def test_a_disabled_gate_bills_without_probing(self) -> None:
tracker = _tracker()
with patch(f"{PE}.assess_screen") as screen:
verdict = tracker.assess(
replace(RULES, engagement_gate=False),
qualifying={7},
now_monotonic=1.0,
)
assert verdict.state == STATE_ENGAGED
assert verdict.qualifying == (7,)
screen.assert_not_called()
def test_every_cause_is_recorded_not_just_the_first(self) -> None:
# The log has to be able to answer "was the screen ALSO locked?", and
# the backdate must be able to tell an idle-only pause from a compound.
tracker = _tracker()
with (
patch(f"{PE}.assess_screen", side_effect=_add("screen_held", (True, 5))),
patch(f"{PE}.assess_focus", side_effect=_add_focus("focus", 9)),
patch(f"{PE}.assess_idle", side_effect=_add_idle("idle", (900.0, None))),
):
verdict = tracker.assess(RULES, qualifying={7}, now_monotonic=1.0)
assert verdict.state == STATE_PAUSED
assert set(verdict.causes) == {"screen_held", "focus", "idle"}
assert verdict.holder_pid == 5
assert verdict.focus_pid == 9
def test_no_causes_means_engaged(self) -> None:
tracker = _tracker()
with (
patch(f"{PE}.assess_screen", return_value=(False, None)),
patch(f"{PE}.assess_focus", return_value=7),
patch(f"{PE}.assess_idle", return_value=(1.0, None)),
):
verdict = tracker.assess(RULES, qualifying={7}, now_monotonic=1.0)
assert verdict.state == STATE_ENGAGED
class TestClose:
def test_releases_both_resources(self) -> None:
probe = MagicMock()
controller = MagicMock()
tracker = EngagementTracker(
uid=1000, probe=probe, controller=controller, holder_path=Path("x")
)
tracker.close()
probe.close.assert_called_once()
controller.close.assert_called_once()
class TestBackdate:
def _state(self, seconds: float = 1000.0) -> PlaytimeState:
return PlaytimeState(day_key="2026-08-28", seconds=seconds)
def test_refunds_the_grace_on_the_engaged_to_idle_edge(self) -> None:
tracker = _tracker()
state = tracker.backdate(self._state(), _engaged(), rules=RULES)
state = tracker.backdate(state, _idle_paused(), rules=RULES)
assert state.seconds == 700.0
def test_refunds_only_once_per_walk_away(self) -> None:
tracker = _tracker()
state = tracker.backdate(self._state(), _engaged(), rules=RULES)
state = tracker.backdate(state, _idle_paused(), rules=RULES)
state = tracker.backdate(state, _idle_paused(), rules=RULES)
assert state.seconds == 700.0
def test_never_refunds_below_zero(self) -> None:
tracker = _tracker()
state = tracker.backdate(self._state(seconds=10.0), _engaged(), rules=RULES)
state = tracker.backdate(state, _idle_paused(), rules=RULES)
assert state.seconds == 0.0
def test_no_refund_on_the_first_tick_after_a_restart(self) -> None:
# With no previous verdict there is no edge to recognise, and inventing
# one would hand out a free refund on every daemon restart.
tracker = _tracker()
state = tracker.backdate(self._state(), _idle_paused(), rules=RULES)
assert state.seconds == 1000.0
def test_no_refund_across_a_day_rollover(self) -> None:
tracker = _tracker()
tracker.backdate(self._state(), _engaged(), rules=RULES)
tomorrow = PlaytimeState(day_key="2026-08-29", seconds=1000.0)
assert tracker.backdate(tomorrow, _idle_paused(), rules=RULES).seconds == 1000.0
def test_no_refund_when_something_else_also_paused_the_tick(self) -> None:
tracker = _tracker()
state = tracker.backdate(self._state(), _engaged(), rules=RULES)
compound = EngagementVerdict(
state=STATE_PAUSED, causes=(CAUSE_IDLE, CAUSE_FOCUS)
)
assert tracker.backdate(state, compound, rules=RULES).seconds == 1000.0
def test_no_refund_while_still_engaged(self) -> None:
tracker = _tracker()
state = tracker.backdate(self._state(), _engaged(), rules=RULES)
assert tracker.backdate(state, _engaged(), rules=RULES).seconds == 1000.0
def test_reports_the_measured_idle_time(self) -> None:
tracker = _tracker()
tracker.backdate(self._state(), _engaged(), rules=RULES)
verdict = EngagementVerdict(
state=STATE_PAUSED, causes=(CAUSE_IDLE,), idle_seconds=301.0
)
assert tracker.backdate(self._state(), verdict, rules=RULES).seconds == 700.0
def _add(cause: str, result: tuple[bool | None, int | None]) -> object:
def impl(_path: object, tally: CauseTally) -> object:
tally.causes.append(cause)
return result
return impl
def _add_focus(cause: str, pid: int) -> object:
def impl(
_probe: object, _rules: object, _qualifying: object, tally: CauseTally
) -> int:
tally.causes.append(cause)
return pid
return impl
def _add_idle(cause: str, result: tuple[float | None, float | None]) -> object:
def impl(
_probe: object,
_controller: object,
_rules: object,
tally: CauseTally,
_now: float,
) -> object:
tally.causes.append(cause)
return result
return impl

View File

@ -0,0 +1,192 @@
"""Tests for the budget audit trail.
Two properties matter. It must never raise — a logging failure taking down the
enforcer would be worse than the missing record — and it must not write a line
per tick, which at a three-second interval would be 28 800 lines a day.
"""
from __future__ import annotations
import json
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._engagement_types import (
CAUSE_IDLE,
STATE_ENGAGED,
STATE_NOT_APPLICABLE,
STATE_PAUSED,
EngagementVerdict,
)
from steam_backlog_enforcer._playtime_log import (
EVENT_DETECTOR_FAILURE,
EVENT_HEARTBEAT,
EVENT_VERDICT_CHANGE,
BudgetLog,
TickJournal,
)
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
PL = "steam_backlog_enforcer._playtime_log"
RULES = PlaytimeRules(
budget_seconds=28800.0,
warn_at=(3600,),
sigkill_after=30.0,
count_launchers=True,
enforcement=True,
demo=False,
)
STATE = PlaytimeState(day_key="2026-08-28", seconds=100.0)
def _records(path: Path) -> list[dict]:
if not path.exists():
return []
return [json.loads(line) for line in path.read_text().splitlines() if line]
class TestBudgetLog:
def test_creates_its_directory_on_first_write(self, tmp_path: Path) -> None:
target = tmp_path / "deep" / "nested" / "budget.jsonl"
log = BudgetLog(path=target)
log.record("hello", answer=42)
log.close()
assert _records(target)[0]["answer"] == 42
def test_closing_an_unopened_sink_is_a_no_op(self, tmp_path: Path) -> None:
BudgetLog(path=tmp_path / "budget.jsonl").close()
def test_a_second_sink_on_the_same_path_replaces_the_first(
self, tmp_path: Path
) -> None:
# Two sinks share a logger name; the stale handler must be closed
# rather than silently dropped, or the file descriptor leaks.
target = tmp_path / "budget.jsonl"
first = BudgetLog(path=target)
first.record("one")
second = BudgetLog(path=target)
second.record("two")
second.close()
assert [r["event"] for r in _records(target)] == ["one", "two"]
def test_every_record_carries_a_timestamp_and_event(self, tmp_path: Path) -> None:
target = tmp_path / "budget.jsonl"
log = BudgetLog(path=target)
log.record("thing")
log.close()
record = _records(target)[0]
assert record["event"] == "thing"
assert record["timestamp"]
def test_reuses_the_sink_across_records(self, tmp_path: Path) -> None:
target = tmp_path / "budget.jsonl"
log = BudgetLog(path=target)
log.record("one")
log.record("two")
log.close()
assert [r["event"] for r in _records(target)] == ["one", "two"]
def test_an_uncreatable_path_does_not_raise(self, tmp_path: Path) -> None:
blocker = tmp_path / "file"
blocker.write_text("not a directory")
BudgetLog(path=blocker / "budget.jsonl").record("dropped")
def test_an_unserialisable_payload_does_not_raise(self, tmp_path: Path) -> None:
target = tmp_path / "budget.jsonl"
log = BudgetLog(path=target)
with patch(f"{PL}.json.dumps", side_effect=ValueError("nope")):
log.record("bad")
log.close()
def test_a_write_failure_does_not_raise(self, tmp_path: Path) -> None:
target = tmp_path / "budget.jsonl"
log = BudgetLog(path=target)
log.record("first")
with patch(f"{PL}.json.dumps", side_effect=OSError("disk gone")):
log.record("second")
log.close()
class TestTickJournal:
def _journal(self, tmp_path: Path, **kwargs: float) -> tuple[TickJournal, Path]:
target = tmp_path / "budget.jsonl"
log = BudgetLog(path=target)
self._logs.append(log)
return TickJournal(log, **kwargs), target
@pytest.fixture(autouse=True)
def _close_logs(self) -> Iterator[None]:
self._logs: list[BudgetLog] = []
yield
for log in self._logs:
log.close()
def _observe(
self, journal: TickJournal, verdict: EngagementVerdict, at: float
) -> None:
journal.observe(verdict, STATE, rules=RULES, now_monotonic=at)
def test_a_changed_verdict_is_recorded(self, tmp_path: Path) -> None:
journal, target = self._journal(tmp_path)
self._observe(journal, EngagementVerdict(state=STATE_ENGAGED), 0.0)
events = [r["event"] for r in _records(target)]
assert events == [EVENT_VERDICT_CHANGE]
def test_an_unchanged_verdict_is_not_repeated(self, tmp_path: Path) -> None:
journal, target = self._journal(tmp_path, heartbeat=300.0)
verdict = EngagementVerdict(state=STATE_ENGAGED)
for tick in range(5):
self._observe(journal, verdict, float(tick) * 3.0)
assert len(_records(target)) == 1
def test_a_heartbeat_is_written_once_the_interval_passes(
self, tmp_path: Path
) -> None:
journal, target = self._journal(tmp_path, heartbeat=10.0)
verdict = EngagementVerdict(state=STATE_ENGAGED)
self._observe(journal, verdict, 0.0)
self._observe(journal, verdict, 5.0)
self._observe(journal, verdict, 50.0)
assert [r["event"] for r in _records(target)] == [
EVENT_VERDICT_CHANGE,
EVENT_HEARTBEAT,
]
def test_ordinary_desktop_use_produces_no_heartbeat(self, tmp_path: Path) -> None:
journal, target = self._journal(tmp_path, heartbeat=0.0)
verdict = EngagementVerdict(state=STATE_NOT_APPLICABLE)
self._observe(journal, verdict, 0.0)
self._observe(journal, verdict, 1000.0)
assert len(_records(target)) == 1
def test_a_degraded_probe_is_always_recorded(self, tmp_path: Path) -> None:
journal, target = self._journal(tmp_path)
verdict = EngagementVerdict(state=STATE_ENGAGED, degraded=("idle",))
self._observe(journal, verdict, 0.0)
self._observe(journal, verdict, 3.0)
events = [r["event"] for r in _records(target)]
assert events.count(EVENT_DETECTOR_FAILURE) == 2
assert all(r.get("billed", True) for r in _records(target))
def test_the_snapshot_names_the_qualifying_processes(self, tmp_path: Path) -> None:
# Without these, accrual from an orphaned helper process cannot be
# attributed after the fact.
journal, target = self._journal(tmp_path)
verdict = EngagementVerdict(
state=STATE_PAUSED,
causes=(CAUSE_IDLE,),
idle_seconds=310.0,
qualifying=(11, 22),
)
self._observe(journal, verdict, 0.0)
record = _records(target)[0]
assert record["qualifying"] == [11, 22]
assert record["reason"] == "idle"
assert record["remaining_seconds"] == 28700.0

View File

@ -0,0 +1,33 @@
"""Tests for building the per-daemon budget session."""
from __future__ import annotations
from unittest.mock import patch
from steam_backlog_enforcer._playtime_engagement import EngagementTracker
from steam_backlog_enforcer._playtime_log import TickJournal
from steam_backlog_enforcer._playtime_session import new_session
PS = "steam_backlog_enforcer._playtime_session"
class TestNewSession:
def test_builds_a_tracker_and_a_journal(self) -> None:
with (
patch(f"{PS}.resolve_desktop_user", return_value="kuhy"),
patch(f"{PS}.desktop_uid", return_value=1000) as uid,
):
session = new_session()
assert isinstance(session.tracker, EngagementTracker)
assert isinstance(session.journal, TickJournal)
uid.assert_called_once_with("kuhy")
def test_binds_the_tracker_to_the_desktop_users_runtime_dir(self) -> None:
# Root's own /run/user/0 would be the wrong place to look for the
# gatelock holder file.
with (
patch(f"{PS}.resolve_desktop_user", return_value="kuhy"),
patch(f"{PS}.desktop_uid", return_value=1000),
):
session = new_session()
assert "/run/user/1000/" in str(session.tracker._holder_path)

View File

@ -6,9 +6,7 @@ from datetime import (
timedelta,
timezone,
)
from unittest.mock import (
patch,
)
from unittest.mock import patch
import pytest
@ -24,6 +22,7 @@ from steam_backlog_enforcer._playtime_state import (
save_state,
)
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._fake_session import fake_session
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
@ -164,7 +163,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
playtime_tick(Config(), interval=3.0, session=fake_session())
assert load_state(demo=False).seconds == 3.0
def test_total_block_releases_and_stops_accruing(self, quiet_tick: dict) -> None:
@ -181,7 +180,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
playtime_tick(Config(), interval=3.0, session=fake_session())
quiet_tick["reconcile"].assert_called_once_with(should_block=False)
assert load_state(demo=False).seconds == 5.0
@ -200,7 +199,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
playtime_tick(Config(), interval=3.0, session=fake_session())
stored = load_state(demo=False)
assert stored.day_key == TODAY
assert stored.seconds == 0.0
@ -222,7 +221,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, demo=True)
playtime_tick(Config(), interval=3.0, session=fake_session(), demo=True)
stored = load_state(demo=True)
assert stored.is_blocked() is True
quiet_tick["request_steam_shutdown"].assert_called_once()
@ -234,7 +233,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, demo=True)
playtime_tick(Config(), interval=3.0, session=fake_session(), demo=True)
assert load_state(demo=True) is not None
assert load_state(demo=False) is None
@ -244,7 +243,7 @@ class TestPlaytimeTick:
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0)
playtime_tick(Config(), interval=3.0, session=fake_session())
stored = load_state(demo=False)
assert stored is not None
assert stored.day_key == TODAY

View File

@ -0,0 +1,111 @@
"""Tests for observing a gatelock screen hold through ``/proc/locks``.
The probe must never take the lock itself, so every case here is driven by
synthetic ``/proc/locks`` text rather than by acquiring anything.
"""
from __future__ import annotations
import os
from pathlib import Path
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._screen_hold import (
ScreenHoldError,
holder_lock_path,
screen_hold,
)
SH = "steam_backlog_enforcer._screen_hold"
def _device_field(path: Path) -> str:
stat = path.stat()
return f"{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}:{stat.st_ino}"
class TestHolderLockPath:
def test_uses_the_runtime_dir_of_the_session_owner(self) -> None:
assert holder_lock_path(1000) == Path("/run/user/1000/gatelock/holder.lock")
def test_differs_per_uid(self) -> None:
assert holder_lock_path(1000) != holder_lock_path(1001)
class TestScreenHold:
def test_missing_file_is_not_held_and_not_an_error(self, tmp_path: Path) -> None:
result = screen_hold(tmp_path / "never-created.lock")
assert result.held is False
assert result.holder_pid is None
def test_reports_held_with_the_holding_pid(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text(
f"1: FLOCK ADVISORY WRITE 4242 {_device_field(lock)} 0 EOF\n"
)
with patch(f"{SH}._PROC_LOCKS", locks):
result = screen_hold(lock)
assert result.held is True
assert result.holder_pid == 4242
def test_reports_free_when_another_file_is_locked(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text("1: FLOCK ADVISORY WRITE 4242 00:19:999999 0 EOF\n")
with patch(f"{SH}._PROC_LOCKS", locks):
assert screen_hold(lock).held is False
def test_a_waiter_row_is_not_a_holder(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text(
f"1: -> FLOCK ADVISORY WRITE 4242 {_device_field(lock)} 0 EOF\n"
)
with patch(f"{SH}._PROC_LOCKS", locks):
assert screen_hold(lock).held is False
def test_rows_without_a_device_field_are_skipped(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text(
"garbage row with no device\n"
f"2: FLOCK ADVISORY READ 7 {_device_field(lock)} 0 EOF\n"
)
with patch(f"{SH}._PROC_LOCKS", locks):
assert screen_hold(lock).held is True
def test_short_row_yields_no_pid(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text(f"1: FLOCK {_device_field(lock)}\n")
with patch(f"{SH}._PROC_LOCKS", locks):
result = screen_hold(lock)
assert result.held is True
assert result.holder_pid is None
def test_non_numeric_pid_yields_no_pid(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
locks = tmp_path / "locks"
locks.write_text(
f"1: FLOCK ADVISORY WRITE nope {_device_field(lock)} 0 EOF\n"
)
with patch(f"{SH}._PROC_LOCKS", locks):
assert screen_hold(lock).holder_pid is None
def test_unreadable_proc_locks_raises(self, tmp_path: Path) -> None:
lock = tmp_path / "holder.lock"
lock.touch()
with (
patch(f"{SH}._PROC_LOCKS", tmp_path / "absent"),
pytest.raises(ScreenHoldError),
):
screen_hold(lock)

View File

@ -0,0 +1,71 @@
"""Tests for refusing to bounce Steam when the bounce would destroy something.
The game case is the one that cost a real session: restarting the daemon on
2026-08-28 relaunched Steam for its CDP port and killed a live game with it.
"""
from __future__ import annotations
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._steam_errors import (
GameInProgressError,
SteamUpdateInProgressError,
)
from steam_backlog_enforcer._steam_restart_guard import (
assert_safe_to_restart,
game_is_running,
)
GUARD = "steam_backlog_enforcer._steam_restart_guard"
class TestGameIsRunning:
def test_no_processes_means_no_game(self) -> None:
with patch(f"{GUARD}.get_running_steam_game_pids", return_value={}):
assert game_is_running() is False
def test_the_steam_client_itself_does_not_count(self) -> None:
with patch(f"{GUARD}.get_running_steam_game_pids", return_value={42: 0}):
assert game_is_running() is False
def test_a_real_app_id_counts(self) -> None:
with patch(
f"{GUARD}.get_running_steam_game_pids", return_value={42: 0, 43: 3164500}
):
assert game_is_running() is True
class TestAssertSafeToRestart:
def test_refuses_while_a_game_runs(self) -> None:
with (
patch(f"{GUARD}.get_running_steam_game_pids", return_value={7: 412830}),
patch(f"{GUARD}.steam_update_in_progress", return_value=False),
pytest.raises(GameInProgressError),
):
assert_safe_to_restart()
def test_a_running_game_outranks_an_update(self) -> None:
with (
patch(f"{GUARD}.get_running_steam_game_pids", return_value={7: 412830}),
patch(f"{GUARD}.steam_update_in_progress", return_value=True),
pytest.raises(GameInProgressError),
):
assert_safe_to_restart()
def test_refuses_during_an_update(self) -> None:
with (
patch(f"{GUARD}.get_running_steam_game_pids", return_value={}),
patch(f"{GUARD}.steam_update_in_progress", return_value=True),
pytest.raises(SteamUpdateInProgressError),
):
assert_safe_to_restart()
def test_allows_when_idle(self) -> None:
with (
patch(f"{GUARD}.get_running_steam_game_pids", return_value={}),
patch(f"{GUARD}.steam_update_in_progress", return_value=False),
):
assert_safe_to_restart()

View File

@ -24,6 +24,9 @@ if TYPE_CHECKING:
SS = "steam_backlog_enforcer._steam_state"
LH = "steam_backlog_enforcer._steam_launch"
# ensure_steam_debug_port defers through the shared restart guard, which
# imports steam_update_in_progress into its own namespace.
GUARD = "steam_backlog_enforcer._steam_restart_guard"
def _manifest(*, to_download: int, downloaded: int, to_stage: int, staged: int) -> str:
@ -106,7 +109,7 @@ class TestEnsureSteamDebugPortDefersUpdate:
patch(f"{LH}.steam_is_installed", return_value=True),
patch(f"{LH}._steam_has_debug_port", return_value=False),
patch(f"{LH}._is_steam_running", return_value=True),
patch(f"{LH}.steam_update_in_progress", return_value=True),
patch(f"{GUARD}.steam_update_in_progress", return_value=True),
patch(f"{LH}._shutdown_steam") as shutdown,
patch(f"{LH}._launch_steam_with_debug") as launch,
):
@ -117,6 +120,18 @@ class TestEnsureSteamDebugPortDefersUpdate:
class TestRestartSteamGuard:
def test_skips_while_a_game_is_running(self) -> None:
# The interactive restart path has the same teeth as the daemon's:
# bouncing Steam kills the game and any unsaved progress with it.
with (
patch(f"{LH}.game_is_running", return_value=True),
patch(f"{LH}._shutdown_steam") as shutdown,
patch(f"{LH}._launch_steam_with_debug") as launch,
):
restart_steam()
shutdown.assert_not_called()
launch.assert_not_called()
def test_skips_when_update_in_progress(self) -> None:
with (
patch(f"{LH}.steam_update_in_progress", return_value=True),

View File

@ -0,0 +1,187 @@
"""Tests for attributing the focused window to a process.
The XRes fallback is not a nicety: measured on this machine, Aseprite's window
carries no ``_NET_WM_PID`` at all, and treating that as "cannot tell" would
make the focus criterion useless for every application that behaves the same.
"""
from __future__ import annotations
import ctypes
from typing import TYPE_CHECKING, cast
from unittest.mock import MagicMock
import pytest
from steam_backlog_enforcer._x_focus import _XResValue, focused_pid, read_cardinal
from steam_backlog_enforcer._x_probe import Conn, XProbeError
from steam_backlog_enforcer.tests._fake_xlib import fake_x11, property_writer
if TYPE_CHECKING:
from collections.abc import Callable
def _conn(x11: MagicMock, xres: MagicMock | None = None) -> Conn:
"""Wrap fake handles in a Conn, casting only at this one seam."""
return Conn(
display=ctypes.c_void_p(1),
x11=cast("ctypes.CDLL", x11),
xss=cast("ctypes.CDLL", MagicMock()),
xres=None if xres is None else cast("ctypes.CDLL", xres),
)
def _xres(
*,
pids: list[int] | None = None,
entries: list[tuple[int, int, int | None]] | None = None,
status: int = 0,
null: bool = False,
) -> MagicMock:
"""Build a libXRes stand-in.
Args:
pids: Shorthand for well-formed PID entries.
entries: Raw ``(mask, length, pid)`` rows; ``pid=None`` means a NULL
value pointer, which the server can legitimately return.
status: Return code; non-zero means the query failed.
null: Whether to leave the results pointer NULL.
"""
rows = (
entries
if entries is not None
else [(1 << 1, 4, pid) for pid in (pids if pids is not None else [])]
)
lib = MagicMock(name="libXRes")
array = (_XResValue * max(len(rows), 1))()
keepalive: list[ctypes.c_uint32] = []
for index, (mask, length, pid) in enumerate(rows):
array[index].spec.mask = mask
array[index].length = length
if pid is None:
array[index].value = None
else:
holder = ctypes.c_uint32(pid)
keepalive.append(holder)
array[index].value = ctypes.cast(
ctypes.byref(holder), ctypes.c_void_p
).value
def impl(
_display: object,
_n: object,
_spec: object,
count_ref: object,
values_ref: object,
) -> int:
count_ref._obj.value = len(rows)
if not null:
values_ref._obj.contents = array[0]
return status
lib.XResQueryClientIds.side_effect = impl
lib._keepalive = (array, keepalive)
return lib
class TestReadCardinal:
def test_returns_the_first_value(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(value=99)
assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") == 99
def test_a_missing_atom_is_none(self) -> None:
assert read_cardinal(_conn(fake_x11(atom=0)), 0x1, b"_NOPE") is None
def test_a_failed_request_is_none(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(status=1)
assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None
def test_a_null_property_is_none(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(null=True)
assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None
def test_an_empty_property_is_none_and_still_freed(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(nitems=0)
assert read_cardinal(_conn(x11), 0x1, b"_NET_WM_PID") is None
x11.XFree.assert_called_once()
class TestFocusedPid:
def test_no_focused_window_is_none(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(value=0)
assert focused_pid(_conn(x11)) is None
def test_uses_net_wm_pid_when_present(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = property_writer(value=555)
assert focused_pid(_conn(x11)) == 555
def test_falls_back_to_xres(self) -> None:
x11 = fake_x11()
# The active window resolves, then _NET_WM_PID comes back absent.
x11.XGetWindowProperty.side_effect = _sequence(
property_writer(value=0x400001), property_writer(nitems=0)
)
assert focused_pid(_conn(x11, _xres(pids=[8080]))) == 8080
def test_raises_when_neither_source_can_attribute_the_window(self) -> None:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = _sequence(
property_writer(value=0x400001), property_writer(nitems=0)
)
with pytest.raises(XProbeError, match="XRes"):
focused_pid(_conn(x11, None))
class TestXResFallback:
def _focus_without_wm_pid(self) -> MagicMock:
x11 = fake_x11()
x11.XGetWindowProperty.side_effect = _sequence(
property_writer(value=0x400001), property_writer(nitems=0)
)
return x11
def test_a_failed_query_gives_no_pid(self) -> None:
with pytest.raises(XProbeError):
focused_pid(_conn(self._focus_without_wm_pid(), _xres(status=1)))
def test_a_null_result_gives_no_pid(self) -> None:
with pytest.raises(XProbeError):
focused_pid(_conn(self._focus_without_wm_pid(), _xres(null=True)))
def test_no_entries_gives_no_pid(self) -> None:
with pytest.raises(XProbeError):
focused_pid(_conn(self._focus_without_wm_pid(), _xres(pids=[])))
def test_a_non_pid_entry_is_skipped(self) -> None:
# XRes can return XID rows alongside PID rows; only the PID row counts.
xres = _xres(entries=[(1 << 0, 4, 111), (1 << 1, 4, 222)])
assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 222
def test_a_short_entry_is_skipped(self) -> None:
xres = _xres(entries=[(1 << 1, 0, 111), (1 << 1, 4, 333)])
assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 333
def test_a_null_value_is_skipped(self) -> None:
xres = _xres(entries=[(1 << 1, 4, None), (1 << 1, 4, 444)])
assert focused_pid(_conn(self._focus_without_wm_pid(), xres)) == 444
def test_the_result_is_destroyed(self) -> None:
xres = _xres(pids=[4242])
focused_pid(_conn(self._focus_without_wm_pid(), xres))
xres.XResClientIdsDestroy.assert_called_once()
def _sequence(*writers: Callable[..., int]) -> Callable[..., int]:
"""Chain property writers so successive calls use successive fakes."""
calls = iter(writers)
def impl(*args: object) -> int:
return next(calls)(*args)
return impl

View File

@ -0,0 +1,120 @@
"""Tests for the cached X connection and the input-idle query.
Every case runs against fake library handles, so the suite never needs a
display and never depends on whether the developer happens to be at the
keyboard.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
from steam_backlog_enforcer._x_probe import (
XProbe,
XProbeError,
_ignore_x_error,
load_libraries,
root_window,
)
from steam_backlog_enforcer.tests._fake_xlib import fake_x11, fake_xss
XP = "steam_backlog_enforcer._x_probe"
class TestLoadLibraries:
def test_returns_all_three_when_present(self) -> None:
x11, xss, xres = MagicMock(), MagicMock(), MagicMock()
with patch(f"{XP}.ctypes.CDLL", side_effect=[x11, xss, xres]):
loaded = load_libraries()
assert loaded[0] is x11
assert loaded[1] is xss
assert loaded[2] is xres
def test_missing_xres_is_tolerated(self) -> None:
x11, xss = MagicMock(), MagicMock()
with patch(f"{XP}.ctypes.CDLL", side_effect=[x11, xss, OSError("no")]):
loaded = load_libraries()
assert loaded[0] is x11
assert loaded[2] is None
def test_missing_core_libraries_raise(self) -> None:
with (
patch(f"{XP}.ctypes.CDLL", side_effect=OSError("no libX11")),
pytest.raises(XProbeError),
):
load_libraries()
class TestConnect:
def test_opens_once_and_caches(self) -> None:
x11 = fake_x11()
probe = XProbe(libraries=(x11, fake_xss(), None))
first = probe.connect()
assert probe.connect() is first
assert x11.XOpenDisplay.call_count == 1
def test_loads_libraries_when_none_were_injected(self) -> None:
libs = (fake_x11(), fake_xss(), None)
with patch(f"{XP}.load_libraries", return_value=libs) as loader:
XProbe().connect()
loader.assert_called_once_with()
def test_a_null_display_raises(self) -> None:
probe = XProbe(libraries=(fake_x11(display=0), fake_xss(), None))
with pytest.raises(XProbeError, match="XOpenDisplay"):
probe.connect()
def test_an_error_handler_is_installed(self) -> None:
# Xlib's default handler calls exit(1); a window vanishing mid-query
# must not be able to take the daemon down.
x11 = fake_x11()
XProbe(libraries=(x11, fake_xss(), None)).connect()
x11.XSetErrorHandler.assert_called_once()
class TestClose:
def test_closing_an_open_connection_releases_it(self) -> None:
x11 = fake_x11()
probe = XProbe(libraries=(x11, fake_xss(), None))
probe.connect()
probe.close()
x11.XCloseDisplay.assert_called_once()
probe.connect()
assert x11.XOpenDisplay.call_count == 2
def test_closing_an_unopened_connection_is_a_no_op(self) -> None:
x11 = fake_x11()
XProbe(libraries=(x11, fake_xss(), None)).close()
x11.XCloseDisplay.assert_not_called()
class TestIdleSeconds:
def test_converts_milliseconds_to_seconds(self) -> None:
probe = XProbe(libraries=(fake_x11(), fake_xss(idle_ms=4500), None))
assert probe.idle_seconds() == 4.5
def test_a_failed_allocation_raises(self) -> None:
probe = XProbe(libraries=(fake_x11(), fake_xss(allocated=False), None))
with pytest.raises(XProbeError, match="AllocInfo"):
probe.idle_seconds()
def test_a_failed_query_raises_and_still_frees(self) -> None:
x11 = fake_x11()
probe = XProbe(libraries=(x11, fake_xss(ok=False), None))
with pytest.raises(XProbeError, match="QueryInfo"):
probe.idle_seconds()
x11.XFree.assert_called_once()
class TestErrorHandler:
def test_swallows_the_error_and_reports_success(self) -> None:
assert _ignore_x_error(None, None) == 0
class TestRootWindow:
def test_returns_the_default_root(self) -> None:
conn = MagicMock()
conn.x11.XDefaultRootWindow.return_value = 0x99
assert root_window(conn) == 0x99