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

Show the gaming budget in the web UI

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-08-29 07:17:00 +02:00
parent 39de18042f
commit fcae46faf1
49 changed files with 2736 additions and 311 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
steam_backlog_enforcer/_budget_log_tail.py Normal file
View File

@ -0,0 +1,138 @@
"""Reading the last recorded verdict back out of the budget audit log.
``_playtime_log`` writes the daemon's own view of why time is or is not being
billed — engagement state, the causes behind it, which PIDs qualified. That is
exactly what a live status display wants, and reading it back beats having the
web server run its own engagement probes: a second opinion that disagreed with
the daemon's would be worse than no opinion at all.
The log is append-only and can reach 5 MB before rotating, so only its tail is
read. One record is all anyone needs and a status endpoint polled every few
seconds must not parse megabytes to find it.
Callers must treat what they get as a *recent* verdict, not a current one:
``TickJournal`` only writes on a change of verdict or a five-minute heartbeat,
so this record can be up to that old. ``observed_at`` is carried through so the
display can say how stale it is.
"""
from __future__ import annotations
from dataclasses import dataclass, field
import json
import logging
from typing import Any, Final
from steam_backlog_enforcer._playtime_log import budget_log_path
from steam_backlog_enforcer._playtime_procs import process_name
logger = logging.getLogger(__name__)
# Enough for several records at any plausible size, small enough to read on
# every poll without thinking about it.
_TAIL_BYTES: Final = 64 * 1024
@dataclass(frozen=True)
class RunningGame:
"""A qualifying process that is still alive."""
pid: int
name: str
@dataclass(frozen=True)
class SessionView:
"""The most recently logged engagement verdict."""
observed_at: str = ""
"""ISO-8601 timestamp the daemon recorded, ``""`` if unknown."""
state: str = ""
reason: str = ""
causes: list[str] = field(default_factory=list)
idle_seconds: float | None = None
screen_held: bool | None = None
games: list[RunningGame] = field(default_factory=list)
available: bool = False
"""Whether a record was found at all."""
def _last_record(*, demo: bool) -> dict[str, Any] | None:
"""Return the last parseable JSON object in the audit log.
Args:
demo: Whether to read the demo log.
Returns:
The decoded record, or ``None`` if the log is missing, unreadable or
holds no complete record in its tail.
"""
path = budget_log_path(demo=demo)
try:
with path.open("rb") as handle:
handle.seek(0, 2)
handle.seek(max(0, handle.tell() - _TAIL_BYTES))
chunk = handle.read()
except OSError:
return None
# The first line of the tail is usually a fragment of a longer record; it is
# discarded naturally by walking backwards to the first line that parses.
for line in reversed(chunk.decode("utf-8", errors="replace").splitlines()):
try:
record = json.loads(line)
except ValueError:
continue
if isinstance(record, dict):
return record
return None
def _live_games(record: dict[str, Any]) -> list[RunningGame]:
"""Resolve the record's qualifying PIDs to names, dropping dead ones.
Args:
record: A decoded audit-log record.
Returns:
One entry per PID that still exists.
"""
qualifying = record.get("qualifying")
if not isinstance(qualifying, list):
return []
games = []
for pid in qualifying:
if not isinstance(pid, int):
continue
name = process_name(pid)
if name is not None:
games.append(RunningGame(pid=pid, name=name))
return games
def last_verdict(*, demo: bool) -> SessionView:
"""Return the most recently logged engagement verdict.
Args:
demo: Whether to read the demo log.
Returns:
The verdict, or an empty view with ``available=False``.
"""
record = _last_record(demo=demo)
if record is None:
return SessionView()
causes = record.get("causes")
idle = record.get("idle_seconds")
held = record.get("screen_held")
return SessionView(
observed_at=str(record.get("timestamp") or ""),
state=str(record.get("state") or ""),
reason=str(record.get("reason") or ""),
causes=[str(cause) for cause in causes] if isinstance(causes, list) else [],
idle_seconds=float(idle) if isinstance(idle, (int, float)) else None,
screen_held=held if isinstance(held, bool) else None,
games=_live_games(record),
available=True,
)

205
steam_backlog_enforcer/_budget_view.py Normal file
View File

@ -0,0 +1,205 @@
"""One read-only view of the gaming budget, shared by MCP and the web API.
Both transports answer the same question — "how much have I played today, and
is the cutoff on?" — so both compose it here. The MCP tool used to build its
reply inline; a second copy in the HTTP handler would have been a second thing
to keep in step with ``PlaytimeRules``.
Everything here is a read. Changing the budget, resetting it and blocking stay
on the privileged paths, which is why the web server can serve this without
gaining a single mutating verb.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Final
from steam_backlog_enforcer._budget_log_tail import last_verdict
from steam_backlog_enforcer._playtime_block import mounted_targets
from steam_backlog_enforcer._playtime_history import load_history
from steam_backlog_enforcer._playtime_state import (
load_state,
rules_for,
state_path,
)
from steam_backlog_enforcer.config import Config, State
if TYPE_CHECKING:
from pathlib import Path
from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
READABLE: Final = "ok"
MISSING: Final = "missing"
DENIED: Final = "denied"
CORRUPT: Final = "corrupt"
_STATUS_MESSAGES: Final[dict[str, str]] = {
DENIED: (
"Cannot read the budget state file. It is owned by root, and only "
"becomes world-readable once the enforcer has rewritten it — restart "
"steam-backlog-enforcer.service."
),
MISSING: "No gaming time has been recorded yet today.",
CORRUPT: (
"The budget state file could not be parsed. The enforcer treats that "
"as tampering and will start the day fail-closed."
),
}
_HISTORY_DAYS: Final = 14
def state_access(path: Path) -> str:
"""Classify why the state file did or did not open.
``load_state`` deliberately collapses "missing", "corrupt" and "unreadable"
into ``None`` — its caller is the enforcer, whose recovery is the same for
all three. A display needs to tell them apart, because "you have not gamed
today" and "I am not allowed to look" are opposite messages.
The read is attempted rather than pre-checked with ``os.access``: the
kernel's answer to the actual open is the only one that cannot disagree
with what ``load_state`` just experienced. The file is a couple of hundred
bytes, so reading it twice costs nothing.
Args:
path: The state file.
Returns:
``READABLE``, ``MISSING`` or ``DENIED``.
"""
try:
path.read_bytes()
except FileNotFoundError:
return MISSING
except OSError:
return DENIED
return READABLE
def next_warning(state: PlaytimeState, rules: PlaytimeRules) -> int | None:
"""Return the next warning threshold that has yet to be reached.
Not ``_playtime_budget.pending_warning``: that returns a threshold already
*crossed* and awaiting its notification, which the daemon fires within one
tick — so an external reader sees ``None`` virtually always. This is the
forward-looking one, the threshold the countdown is heading towards.
Args:
state: Current accounting state.
rules: Policy in force.
Returns:
Seconds-remaining at which the next warning fires, or ``None`` when
every threshold has been passed.
"""
remaining = rules.budget_seconds - state.seconds
upcoming = [
threshold
for threshold in rules.warn_at
if threshold < remaining and threshold not in state.warned_seconds
]
return max(upcoming) if upcoming else None
def build_today(state: PlaytimeState, rules: PlaytimeRules) -> dict[str, Any]:
"""Compose the "today" block from loaded state.
Args:
state: Loaded accounting state.
rules: Policy in force.
Returns:
Serialisable today-block.
"""
remaining = max(0.0, rules.budget_seconds - state.seconds)
fraction = state.seconds / rules.budget_seconds if rules.budget_seconds else 1.0
return {
"gaming_day": state.day_key,
"day_starts_at": "06:00 local",
"seconds_used": round(state.seconds, 1),
"budget_seconds": rules.budget_seconds,
"seconds_remaining": round(remaining, 1),
"fraction_used": round(min(1.0, max(0.0, fraction)), 4),
"blocked": state.is_blocked(),
"blocked_at": state.blocked_at,
"next_warning_seconds": next_warning(state, rules),
"warned_seconds": list(state.warned_seconds),
}
def build_rules(rules: PlaytimeRules) -> dict[str, Any]:
"""Compose the rules block.
An explicit allowlist, never ``Config.__dict__``: that carries
``steam_api_key``, and this payload is served over HTTP.
Args:
rules: Policy in force.
Returns:
Serialisable rules block.
"""
return {
"budget_seconds": rules.budget_seconds,
"enforcement": rules.enforcement,
"counts_launchers": rules.count_launchers,
"engagement_gate": rules.engagement_gate,
"idle_grace_seconds": rules.idle_grace_seconds,
"require_game_focus": rules.require_game_focus,
"warn_at": list(rules.warn_at),
"demo": rules.demo,
"masked_launchers": sorted(str(path) for path in mounted_targets()),
}
def build_budget_snapshot(*, demo: bool = False) -> dict[str, Any]:
"""Compose the whole budget view: today, live session, history, rules.
Args:
demo: Whether to read the demo run's state and log.
Returns:
The serialisable payload behind ``GET /api/budget``.
"""
config = Config.load()
rules = rules_for(config, demo=demo)
stored = load_state(demo=demo)
access = state_access(state_path(demo=demo))
# A file that opened but did not parse is the remaining case: the enforcer
# treats that as tampering, and the display should say so rather than
# implying an empty day.
status = access if access != READABLE else (READABLE if stored else CORRUPT)
session = last_verdict(demo=demo)
return {
"ok": True,
"readable": status == READABLE,
"state_status": status,
"error": _STATUS_MESSAGES.get(status),
"today": build_today(stored, rules) if stored is not None else None,
"session": {
"available": session.available,
"observed_at": session.observed_at,
"state": session.state,
"reason": session.reason,
"causes": session.causes,
"idle_seconds": session.idle_seconds,
"screen_held": session.screen_held,
# A single Proton title qualifies a dozen processes — reaper,
# srt-bwrap, wineserver, plugplay.exe. Listing those as "games
# running" would be nonsense, so the headline is the game the
# enforcer actually assigned and the processes are the detail.
"game_name": State.load().current_game_name,
"qualifying_count": len(session.games),
"processes": [
{"pid": game.pid, "name": game.name} for game in session.games
],
},
"history": [
{"day": day.day, "seconds": round(day.seconds, 1)}
for day in load_history(_HISTORY_DAYS)
],
"rules": build_rules(rules),
}

33
steam_backlog_enforcer/_config_setup.py Normal file
View File

@ -0,0 +1,33 @@
"""First-run interactive configuration.
Split out of ``config.py``: prompting a human for credentials is a different
concern from the data model those credentials live in, and only the ``setup``
command ever reaches it. The existing ``test_config_interactive_setup.py``
already treated it as its own unit.
"""
from __future__ import annotations
import sys
from steam_backlog_enforcer import config as config_mod
from steam_backlog_enforcer.config import Config
def interactive_setup() -> Config:
"""Run first-time interactive setup."""
api_key = input("Enter your Steam Web API key: ").strip()
if not api_key:
sys.exit(1)
steam_id = input("Enter your Steam64 ID: ").strip()
if not steam_id:
sys.exit(1)
config = Config(steam_api_key=api_key, steam_id=steam_id)
config.save()
# Read through the module, not an import-time binding: the path is
# redirected in tests, and the API key lives here — it must not become
# world-readable, nor be written to the real config during a test run.
config_mod.CONFIG_FILE.chmod(0o600)
return config

39
steam_backlog_enforcer/_mcp_gaming_time.py
View File

@ -11,6 +11,7 @@ from datetime import datetime, timezone
import os
from typing import Any
from steam_backlog_enforcer._budget_view import build_budget_snapshot
from steam_backlog_enforcer._mcp_server import (
logger,
mcp,
@ -18,12 +19,9 @@ from steam_backlog_enforcer._mcp_server import (
from steam_backlog_enforcer._playtime import (
PlaytimeState,
gaming_day_key,
load_state,
rules_for,
save_state,
)
from steam_backlog_enforcer._playtime_block import mounted_targets, release_block
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer._playtime_block import release_block
@mcp.tool()
@ -34,31 +32,34 @@ def get_gaming_time() -> dict[str, Any]:
midnight still counts against the day it began in. Reads on-disk state and
the live mount table only — no network, no Steam API key, no config secrets.
"""
config = Config.load()
rules = rules_for(config, demo=False)
stored = load_state(demo=False)
masked = sorted(str(path) for path in mounted_targets())
snapshot = build_budget_snapshot(demo=False)
rules = snapshot["rules"]
today = snapshot["today"]
masked = rules["masked_launchers"]
if stored is None:
if today is None:
return {
"ok": True,
"recorded": False,
"budget_seconds": rules.budget_seconds,
"enforcement": rules.enforcement,
"state_status": snapshot["state_status"],
"budget_seconds": rules["budget_seconds"],
"enforcement": rules["enforcement"],
"masked_launchers": masked,
}
return {
"ok": True,
"recorded": True,
"gaming_day": stored.day_key,
"day_starts_at": "06:00 local",
"seconds_used": round(stored.seconds, 1),
"budget_seconds": rules.budget_seconds,
"seconds_remaining": round(max(0.0, rules.budget_seconds - stored.seconds), 1),
"blocked": stored.is_blocked(),
"enforcement": rules.enforcement,
"counts_launchers": rules.count_launchers,
"state_status": snapshot["state_status"],
"gaming_day": today["gaming_day"],
"day_starts_at": today["day_starts_at"],
"seconds_used": today["seconds_used"],
"budget_seconds": today["budget_seconds"],
"seconds_remaining": today["seconds_remaining"],
"next_warning_seconds": today["next_warning_seconds"],
"blocked": today["blocked"],
"enforcement": rules["enforcement"],
"counts_launchers": rules["counts_launchers"],
"masked_launchers": masked,
}

2
steam_backlog_enforcer/_playtime.py
View File

@ -101,6 +101,7 @@ def playtime_tick(
# way, and stop accruing against a budget nobody can spend.
reconcile(should_block=False)
save_state(state, demo=demo)
session.history.observe(state, demo=demo)
return
pids = qualifying_pids(rules)
@ -117,6 +118,7 @@ def playtime_tick(
state = _policy(state, rules, now=now)
session.journal.observe(verdict, state, rules=rules, now_monotonic=monotonic)
save_state(state, demo=demo)
session.history.observe(state, demo=demo)
def _state_or_recover(rules: PlaytimeRules, *, now: datetime) -> PlaytimeState:

161
steam_backlog_enforcer/_playtime_history.py Normal file
View File

@ -0,0 +1,161 @@
"""A durable per-day record of how much gaming time was actually spent.
``PlaytimeState`` only ever describes *today*: ``roll_over`` resets the counter
at the 06:00 boundary and the previous day is gone. The audit log in
``_playtime_log`` does retain per-day figures, but it is a rotating handler —
five 5 MB files — so a busy fortnight can push a day out of the window entirely.
Neither is a series you can plot.
So this module keeps one small file whose only job is the history: a
``day_key -> seconds`` mapping, trimmed to the last month, written world-readable
so an unprivileged reader (the web UI) can plot it.
Writes are throttled rather than tied to roll-over. Upserting today's figure on a
cadence means the outgoing day is already recorded to within ``_FLUSH_DELTA``
seconds of its final value by the time the boundary arrives, which avoids
threading a "previous state" argument through the tick's hot path just to catch
one transition a day.
Like ``BudgetLog``, nothing here may raise: losing a history point is a cosmetic
failure and must never take down the enforcer.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
import logging
from typing import TYPE_CHECKING, Final
from steam_backlog_enforcer._playtime_state import STATE_MODE
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
if TYPE_CHECKING:
from steam_backlog_enforcer._playtime_state import PlaytimeState
logger = logging.getLogger(__name__)
HISTORY_FILE: Final = CONFIG_DIR / "playtime_history.json"
_SCHEMA_VERSION: Final = 1
# Days retained. Comfortably more than the fortnight the UI plots, so a chart
# that later grows a longer window has data waiting for it.
_MAX_DAYS: Final = 30
# Rewrite only once the stored figure is this far behind. At a three-second tick
# an unthrottled upsert would rewrite the file 1200 times an hour to move a bar
# by a pixel.
_FLUSH_DELTA: Final = 60.0
@dataclass(frozen=True)
class HistoryDay:
"""One gaming day's total, as plotted."""
day: str
"""Gaming day, ``YYYY-MM-DD``, 06:00-shifted."""
seconds: float
"""Qualifying seconds billed during ``day``."""
def _read_days() -> dict[str, float]:
"""Return the stored ``day -> seconds`` mapping, or empty if unusable.
Unlike the state file this one is neither root-owned nor immutable, and it
lives in a user-owned directory — so an unprivileged process can replace it
with anything at all, and the web server parses whatever it finds. Every
field is therefore checked rather than trusted.
Returns:
The mapping, or ``{}`` when the file is missing, unreadable, corrupt or
of an unknown schema.
"""
try:
raw = json.loads(HISTORY_FILE.read_text(encoding="utf-8"))
except FileNotFoundError:
return {}
except (OSError, ValueError):
logger.warning(
"Playtime history at %s is unreadable; starting over.", HISTORY_FILE
)
return {}
if not isinstance(raw, dict) or raw.get("schema_version") != _SCHEMA_VERSION:
logger.warning("Playtime history at %s has an unknown schema.", HISTORY_FILE)
return {}
days = raw.get("days")
if not isinstance(days, dict):
return {}
return {
key: float(value)
for key, value in days.items()
if isinstance(key, str) and isinstance(value, (int, float))
}
def load_history(limit: int = 14) -> list[HistoryDay]:
"""Return the most recent *limit* days, oldest first.
Args:
limit: How many days to return.
Returns:
Recorded days in ascending date order, at most *limit* of them.
"""
days = _read_days()
ordered = sorted(days.items())[-limit:] if limit > 0 else []
return [HistoryDay(day=day, seconds=seconds) for day, seconds in ordered]
def record_day(day_key: str, seconds: float) -> None:
"""Upsert one day's total and trim the file to ``_MAX_DAYS``.
Args:
day_key: Gaming day to record.
seconds: Total billed for that day so far.
"""
days = _read_days()
days[day_key] = seconds
trimmed = dict(sorted(days.items())[-_MAX_DAYS:])
_atomic_write(
HISTORY_FILE,
json.dumps({"schema_version": _SCHEMA_VERSION, "days": trimmed}, indent=2)
+ "\n",
mode=STATE_MODE,
)
class HistoryWriter:
"""Throttled writer for the per-day history, held across ticks.
Mirrors how ``TickJournal`` carries its heartbeat state on the session
rather than in a module global, which is what keeps the throttle testable
without reaching into module internals between cases.
"""
def __init__(self) -> None:
self._last_written: dict[str, float] = {}
def observe(self, state: PlaytimeState, *, demo: bool) -> None:
"""Record *state*'s day if it is new or has moved far enough.
Demo runs are skipped for the same reason they get their own state file
and their own audit log: a 60-second demo must not leave a record
describing a budget nobody was ever charged.
Args:
state: Current accounting state.
demo: Whether this is a demo run.
"""
if demo or not state.day_key:
return
previous = self._last_written.get(state.day_key)
if previous is not None and abs(state.seconds - previous) < _FLUSH_DELTA:
return
try:
record_day(state.day_key, state.seconds)
except OSError as exc:
# A history point is cosmetic; the enforcer keeps enforcing.
logger.warning("Could not record playtime history: %s", exc)
return
self._last_written = {state.day_key: state.seconds}

19
steam_backlog_enforcer/_playtime_procs.py
View File

@ -91,6 +91,25 @@ def _match_cmdline(entry: Path, names: frozenset[str]) -> str | None:
return second if second in names else None
def process_name(pid: int) -> str | None:
"""Return *pid*'s program name, or ``None`` if it is no longer running.
Existence is the point as much as the name: callers resolve PIDs recorded
earlier, and a PID that has since been recycled would otherwise be reported
under whatever process now holds the number.
Args:
pid: Process id to look up.
Returns:
The contents of ``/proc/<pid>/comm``, or ``None``.
"""
try:
return (_PROC / str(pid) / "comm").read_text(encoding="utf-8").strip() or None
except (OSError, ValueError):
return None
def qualifying_pids(rules: PlaytimeRules) -> set[int]:
"""Return PIDs whose runtime counts against the daily budget.

10
steam_backlog_enforcer/_playtime_session.py
View File

@ -19,6 +19,7 @@ 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_history import HistoryWriter
from steam_backlog_enforcer._playtime_log import (
BudgetLog,
TickJournal,
@ -66,12 +67,20 @@ class TickRecorder(Protocol):
"""Record this tick if it is worth recording."""
class HistoryRecorder(Protocol):
"""What the budget tick needs from the per-day history writer."""
def observe(self, state: PlaytimeState, *, demo: bool) -> None:
"""Record today's running total if it has moved far enough."""
@dataclass(frozen=True)
class PlaytimeSession:
"""Per-daemon engagement and logging state."""
tracker: EngagementSource
journal: TickRecorder
history: HistoryRecorder
def new_session(*, demo: bool = False) -> PlaytimeSession:
@ -89,4 +98,5 @@ def new_session(*, demo: bool = False) -> PlaytimeSession:
return PlaytimeSession(
tracker=EngagementTracker(uid=uid),
journal=TickJournal(BudgetLog(path=budget_log_path(demo=demo))),
history=HistoryWriter(),
)

11
steam_backlog_enforcer/_playtime_state.py
View File

@ -28,6 +28,15 @@ PLAYTIME_DEMO_STATE_FILE = CONFIG_DIR / "playtime_demo_state.json"
_SCHEMA_VERSION = 1
# The daemon runs as root, and ``mkstemp`` would leave this at 0600 — which
# silently broke every unprivileged reader: ``load_state`` swallows the
# resulting ``PermissionError`` and reports "no state recorded yet", so both the
# MCP ``get_gaming_time`` tool and the web UI showed an empty budget while the
# file on disk held a live count. World-*readable* costs nothing here: the file
# holds no secrets, and root ownership plus the immutable flag are what stop it
# being rewritten.
STATE_MODE = 0o644
# The gaming day starts here, local time. 05:59 belongs to the previous day.
_DAY_BOUNDARY_HOURS = 6
@ -197,6 +206,6 @@ def save_state(state: PlaytimeState, *, demo: bool) -> None:
path = state_path(demo=demo)
if not demo:
unlock_for_write(path)
_atomic_write(path, json.dumps(state.__dict__, indent=2) + "\n")
_atomic_write(path, json.dumps(state.__dict__, indent=2) + "\n", mode=STATE_MODE)
if not demo:
_try_set_immutable(path, immutable=True)

25
steam_backlog_enforcer/_web_server.py
View File

@ -19,6 +19,7 @@ import mimetypes
from pathlib import Path
from urllib.parse import urlsplit
from steam_backlog_enforcer._budget_view import build_budget_snapshot
from steam_backlog_enforcer._web_dataset import build_web_dataset, dataset_to_payload
from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.game_install import _echo
@ -31,6 +32,7 @@ WEB_DIST = (Path(__file__).resolve().parent.parent / "web" / "dist").resolve()
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8000
_API_DATASET = "/api/dataset"
_API_BUDGET = "/api/budget"
# Content types that are text but not under the ``text/`` prefix.
_EXTRA_TEXT_TYPES = frozenset(
@ -47,12 +49,27 @@ class _Handler(BaseHTTPRequestHandler):
logger.debug("%s - %s", self.address_string(), fmt % args)
def do_GET(self) -> None:
"""Dispatch a GET to the dataset API or to a static file."""
path = urlsplit(self.path).path
if path == _API_DATASET:
"""Dispatch a GET to one of the APIs or to a static file."""
split = urlsplit(self.path)
if split.path == _API_DATASET:
self._serve_dataset()
elif split.path == _API_BUDGET:
# ``?demo=1`` reads the demo run's state and log, which is how the
# 60-second demo can be watched hitting its cutoff in the browser
# without spending a real day's budget to see it.
self._serve_budget(demo="demo=1" in split.query)
else:
self._serve_static(path)
self._serve_static(split.path)
def _serve_budget(self, *, demo: bool) -> None:
"""Build and send the gaming-budget snapshot as JSON."""
try:
body = json.dumps(build_budget_snapshot(demo=demo)).encode("utf-8")
except (OSError, ValueError, KeyError):
logger.exception("Failed to build budget snapshot")
self._send(HTTPStatus.INTERNAL_SERVER_ERROR, b"budget error", "text/plain")
return
self._send(HTTPStatus.OK, body, "application/json")
def _serve_dataset(self) -> None:
"""Build and send the projected dataset as JSON."""

34
steam_backlog_enforcer/config.py
View File

@ -9,7 +9,6 @@ import json
import logging
import os
from pathlib import Path
import sys
import tempfile
from typing import Any
@ -33,13 +32,26 @@ HOSTS_FILE = Path("/etc/hosts")
logger = logging.getLogger(__name__)
def _atomic_write(path: Path, data: str) -> None:
"""Write data to a file atomically via a temporary file + rename."""
def _atomic_write(path: Path, data: str, *, mode: int | None = None) -> None:
"""Write data to a file atomically via a temporary file + rename.
Args:
path: Destination file.
data: Text to write.
mode: Permission bits for the result. ``None`` keeps ``mkstemp``'s
0600, which is what files holding secrets (``config.json`` carries
``steam_api_key``) must stay at. Callers whose file has to be
readable by an unprivileged reader pass it explicitly; it is set on
the temporary file so the mode lands with the rename rather than
leaving a window where the destination exists at 0600.
"""
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
tmp_path = Path(tmp)
try:
os.write(fd, data.encode("utf-8"))
if mode is not None:
os.fchmod(fd, mode)
os.close(fd)
tmp_path.replace(path)
except BaseException:
@ -232,19 +244,3 @@ def load_snapshot() -> list[dict[str, Any]] | None:
)
return result
return None
def interactive_setup() -> Config:
"""Run first-time interactive setup."""
api_key = input("Enter your Steam Web API key: ").strip()
if not api_key:
sys.exit(1)
steam_id = input("Enter your Steam64 ID: ").strip()
if not steam_id:
sys.exit(1)
config = Config(steam_api_key=api_key, steam_id=steam_id)
config.save()
CONFIG_FILE.chmod(0o600)
return config

2
steam_backlog_enforcer/main/misc.py
View File

@ -9,6 +9,7 @@ from __future__ import annotations
import sys
from typing import TYPE_CHECKING
from steam_backlog_enforcer._config_setup import interactive_setup
from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
from steam_backlog_enforcer._total_block import start_total_block
from steam_backlog_enforcer._web_server import serve
@ -16,7 +17,6 @@ from steam_backlog_enforcer._whitelist import (
add_pending_exception,
validate_reason,
)
from steam_backlog_enforcer.config import interactive_setup
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.library_hider import restart_steam, unhide_all_games
from steam_backlog_enforcer.store_blocker import block_store, unblock_store

23
steam_backlog_enforcer/tests/_fake_session.py
View File

@ -111,6 +111,23 @@ class StubJournal:
self.observed.append(verdict)
class StubHistory:
"""A per-day history writer that records in memory instead of on disk."""
def __init__(self) -> None:
self.recorded: list[tuple[str, float]] = []
def observe(self, state: PlaytimeState, *, demo: bool) -> None:
"""Remember the day and total.
Args:
state: This tick's state.
demo: Ignored; the real writer skips demo runs.
"""
del demo
self.recorded.append((state.day_key, state.seconds))
def fake_session(*, engaged: bool = True) -> PlaytimeSession:
"""Build a session that never touches X, /proc/locks or the log file.
@ -120,4 +137,8 @@ def fake_session(*, engaged: bool = True) -> PlaytimeSession:
Returns:
The session.
"""
return PlaytimeSession(tracker=StubTracker(engaged=engaged), journal=StubJournal())
return PlaytimeSession(
tracker=StubTracker(engaged=engaged),
journal=StubJournal(),
history=StubHistory(),
)

17
steam_backlog_enforcer/tests/_isolate_playtime.py
View File

@ -49,6 +49,23 @@ def _isolate_playtime(tmp_path: Path) -> Iterator[None]:
"steam_backlog_enforcer._playtime_state.PLAYTIME_DEMO_STATE_FILE",
fake_config / "playtime_demo_state.json",
),
# Per-day history. Derived from CONFIG_DIR at import time, so patching
# CONFIG_DIR alone does not redirect it.
patch(
"steam_backlog_enforcer._playtime_history.HISTORY_FILE",
fake_config / "playtime_history.json",
),
# The audit log lives under /var/log, outside CONFIG_DIR entirely, and
# is the record used to reconstruct real incidents — a test must never
# append to it, nor read the host's.
patch(
"steam_backlog_enforcer._playtime_log.BUDGET_LOG_FILE",
tmp_path / "budget.jsonl",
),
patch(
"steam_backlog_enforcer._playtime_log.BUDGET_DEMO_LOG_FILE",
tmp_path / "budget-demo.jsonl",
),
patch("steam_backlog_enforcer._playtime_procs._PROC", fake_proc),
patch("steam_backlog_enforcer._playtime_kill._PROC", fake_proc),
# Never let a test chattr +i a file inside tmp_path — pytest could

77
steam_backlog_enforcer/tests/test_atomic_write.py Normal file
View File

@ -0,0 +1,77 @@
"""Tests for _atomic_write — the shared atomic file-write primitive.
Split out of ``test_config.py`` to hold both files inside the 250-line cap;
``_atomic_write`` is a file-writing primitive rather than part of the config
data model, and its permission behaviour is what several other modules rely on.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from steam_backlog_enforcer.config import _atomic_write
if TYPE_CHECKING:
from pathlib import Path
class TestAtomicWrite:
"""Tests for _atomic_write."""
def test_writes_file(self, tmp_path: Path) -> None:
"""Test writes file."""
target = tmp_path / "out.json"
_atomic_write(target, '{"key": "value"}\n')
assert target.read_text(encoding="utf-8") == '{"key": "value"}\n'
def test_defaults_to_owner_only(self, tmp_path: Path) -> None:
"""Files that may hold secrets keep mkstemp's 0600."""
target = tmp_path / "out.json"
_atomic_write(target, "data")
assert target.stat().st_mode & 0o777 == 0o600
def test_honours_an_explicit_mode(self, tmp_path: Path) -> None:
"""The mode lands on the result, not just on the temp file."""
target = tmp_path / "out.json"
_atomic_write(target, "data", mode=0o644)
assert target.stat().st_mode & 0o777 == 0o644
def test_creates_parent_dirs(self, tmp_path: Path) -> None:
"""Test creates parent dirs."""
target = tmp_path / "sub" / "deep" / "out.json"
_atomic_write(target, "data")
assert target.read_text(encoding="utf-8") == "data"
def test_cleanup_on_write_error(self, tmp_path: Path) -> None:
"""Test cleanup on write error."""
target = tmp_path / "out.json"
with (
patch(
"steam_backlog_enforcer.config.os.write",
side_effect=OSError("disk full"),
),
pytest.raises(OSError, match="disk full"),
):
_atomic_write(target, "data")
assert not target.exists()
tmp_files = list(tmp_path.glob("*.tmp"))
assert tmp_files == []
def test_cleanup_on_replace_error(self, tmp_path: Path) -> None:
"""Test cleanup on replace error."""
target = tmp_path / "out.json"
with (
patch.object(
type(target),
"replace",
side_effect=OSError("no perm"),
),
pytest.raises(OSError, match="no perm"),
):
_atomic_write(target, "data")
assert not target.exists()
tmp_files = list(tmp_path.glob("*.tmp"))
assert tmp_files == []

128
steam_backlog_enforcer/tests/test_budget_log_tail.py Normal file
View File

@ -0,0 +1,128 @@
"""Tests for _budget_log_tail module — 100% branch coverage."""
from __future__ import annotations
import json
from unittest.mock import patch
from steam_backlog_enforcer import _playtime_log as log_mod
from steam_backlog_enforcer._budget_log_tail import last_verdict
_PKG = "steam_backlog_enforcer._budget_log_tail"
def _write(lines: list[str]) -> None:
log_mod.BUDGET_LOG_FILE.write_text("\n".join(lines) + "\n", encoding="utf-8")
def _record(**over: object) -> str:
body: dict[str, object] = {
"timestamp": "2026-08-28T20:42:19+02:00",
"event": "verdict_change",
"state": "engaged",
"reason": "engaged",
"causes": [],
"idle_seconds": 1.5,
"screen_held": False,
"qualifying": [],
"day_key": "2026-08-28",
"billed_seconds": 100.0,
}
body.update(over)
return json.dumps(body)
class TestLastVerdict:
"""Tests for reading the tail of the audit log."""
def test_missing_log_is_unavailable(self) -> None:
view = last_verdict(demo=False)
assert view.available is False
assert view.state == ""
def test_reads_the_last_record(self) -> None:
_write([_record(state="paused"), _record(state="engaged", reason="ok")])
view = last_verdict(demo=False)
assert view.available is True
assert view.state == "engaged"
assert view.reason == "ok"
assert view.idle_seconds == 1.5
assert view.screen_held is False
assert view.observed_at.startswith("2026-08-28")
def test_skips_a_malformed_last_line(self) -> None:
_write([_record(state="engaged"), "{ this is not json"])
assert last_verdict(demo=False).state == "engaged"
def test_empty_log_is_unavailable(self) -> None:
log_mod.BUDGET_LOG_FILE.write_text("", encoding="utf-8")
assert last_verdict(demo=False).available is False
def test_a_json_scalar_is_not_a_record(self) -> None:
_write(["42"])
assert last_verdict(demo=False).available is False
def test_unreadable_log_is_unavailable(self) -> None:
_write([_record()])
with patch("pathlib.Path.open", side_effect=PermissionError):
assert last_verdict(demo=False).available is False
def test_reads_only_the_tail(self) -> None:
# Seeking back a fixed number of bytes lands mid-record; that leading
# fragment must be skipped rather than crashing the read.
newest = _record(state="newest")
_write([_record(state="old")] * 50 + [newest])
with patch(f"{_PKG}._TAIL_BYTES", len(newest) + 20):
assert last_verdict(demo=False).state == "newest"
def test_a_tail_shorter_than_one_record_finds_nothing(self) -> None:
_write([_record(state="newest")])
with patch(f"{_PKG}._TAIL_BYTES", 10):
assert last_verdict(demo=False).available is False
def test_demo_log_is_read_separately(self) -> None:
_write([_record(state="production")])
log_mod.BUDGET_DEMO_LOG_FILE.write_text(
_record(state="demo") + "\n", encoding="utf-8"
)
assert last_verdict(demo=True).state == "demo"
assert last_verdict(demo=False).state == "production"
def test_missing_and_odd_fields_default(self) -> None:
_write([json.dumps({"event": "heartbeat"})])
view = last_verdict(demo=False)
assert view.available is True
assert view.state == ""
assert view.causes == []
assert view.idle_seconds is None
assert view.screen_held is None
assert view.observed_at == ""
def test_non_list_causes_are_dropped(self) -> None:
_write([_record(causes="focus")])
assert last_verdict(demo=False).causes == []
class TestLiveGames:
"""Tests for resolving qualifying PIDs to live processes."""
def test_resolves_running_pids(self) -> None:
_write([_record(qualifying=[1, 2])])
with patch(f"{_PKG}.process_name", side_effect=["alpha", "beta"]):
games = last_verdict(demo=False).games
assert [(g.pid, g.name) for g in games] == [(1, "alpha"), (2, "beta")]
def test_drops_pids_that_have_exited(self) -> None:
# A recycled PID would otherwise be reported under the wrong process.
_write([_record(qualifying=[1, 2])])
with patch(f"{_PKG}.process_name", side_effect=["alpha", None]):
games = last_verdict(demo=False).games
assert [g.name for g in games] == ["alpha"]
def test_ignores_non_integer_pids(self) -> None:
_write([_record(qualifying=["nope"])])
assert last_verdict(demo=False).games == []
def test_non_list_qualifying_is_dropped(self) -> None:
_write([_record(qualifying=7)])
assert last_verdict(demo=False).games == []

198
steam_backlog_enforcer/tests/test_budget_view.py Normal file
View File

@ -0,0 +1,198 @@
"""Tests for _budget_view module — 100% branch coverage."""
from __future__ import annotations
import json
from unittest.mock import patch
from steam_backlog_enforcer import _budget_view as view
from steam_backlog_enforcer import _playtime_state as state_mod
from steam_backlog_enforcer._budget_log_tail import RunningGame, SessionView
from steam_backlog_enforcer._budget_view import (
CORRUPT,
DENIED,
MISSING,
READABLE,
build_budget_snapshot,
build_rules,
build_today,
next_warning,
state_access,
)
from steam_backlog_enforcer._playtime_history import HistoryDay
from steam_backlog_enforcer._playtime_state import PlaytimeState, rules_for
from steam_backlog_enforcer.config import Config
RULES = rules_for(Config(), demo=False)
class TestStateAccess:
"""Tests for telling apart the reasons state fails to load."""
def test_missing_file(self) -> None:
assert state_access(state_mod.PLAYTIME_STATE_FILE) == MISSING
def test_readable_file(self) -> None:
state_mod.PLAYTIME_STATE_FILE.write_text("{}", encoding="utf-8")
assert state_access(state_mod.PLAYTIME_STATE_FILE) == READABLE
def test_unreadable_file(self) -> None:
state_mod.PLAYTIME_STATE_FILE.write_text("{}", encoding="utf-8")
with patch("pathlib.Path.read_bytes", side_effect=PermissionError):
assert state_access(state_mod.PLAYTIME_STATE_FILE) == DENIED
class TestNextWarning:
"""Tests for the forward-looking warning threshold."""
def test_largest_threshold_below_remaining(self) -> None:
state = PlaytimeState(day_key="d", seconds=0.0)
assert next_warning(state, RULES) == 3600
def test_skips_thresholds_already_fired(self) -> None:
state = PlaytimeState(day_key="d", seconds=0.0, warned_seconds=[3600])
assert next_warning(state, RULES) == 1800
def test_none_once_every_threshold_is_passed(self) -> None:
state = PlaytimeState(day_key="d", seconds=RULES.budget_seconds)
assert next_warning(state, RULES) is None
def test_is_not_the_already_crossed_threshold(self) -> None:
# pending_warning() would return 3600 here; the display wants the next
# one the countdown is heading towards.
state = PlaytimeState(day_key="d", seconds=RULES.budget_seconds - 3000)
assert next_warning(state, RULES) == 1800
class TestBuildToday:
"""Tests for the today block."""
def test_reports_usage_and_remaining(self) -> None:
today = build_today(PlaytimeState(day_key="2026-08-28", seconds=100.0), RULES)
assert today["seconds_used"] == 100.0
assert today["seconds_remaining"] == RULES.budget_seconds - 100.0
assert today["blocked"] is False
assert today["fraction_used"] < 0.01
def test_never_reports_negative_remaining(self) -> None:
today = build_today(PlaytimeState(day_key="d", seconds=10**9), RULES)
assert today["seconds_remaining"] == 0.0
assert today["fraction_used"] == 1.0
def test_reports_the_cutoff(self) -> None:
today = build_today(
PlaytimeState(day_key="d", seconds=1.0, blocked_at=2.0), RULES
)
assert today["blocked"] is True
assert today["blocked_at"] == 2.0
def test_a_zero_budget_reads_as_fully_spent(self) -> None:
rules = rules_for(Config(daily_gaming_seconds=0), demo=False)
today = build_today(PlaytimeState(day_key="d", seconds=0.0), rules)
assert today["fraction_used"] == 1.0
class TestBuildRules:
"""Tests for the rules block."""
def test_lists_the_policy(self) -> None:
with patch.object(view, "mounted_targets", return_value=set()):
rules = build_rules(RULES)
assert rules["budget_seconds"] == 8 * 3600
assert rules["warn_at"] == [3600, 1800, 600, 300]
assert rules["masked_launchers"] == []
def test_lists_masked_launchers(self) -> None:
with patch.object(view, "mounted_targets", return_value={"/usr/bin/steam"}):
assert build_rules(RULES)["masked_launchers"] == ["/usr/bin/steam"]
class TestBuildBudgetSnapshot:
"""Tests for the whole payload."""
def _snapshot(self, **over: object) -> dict:
kwargs: dict = {
"stored": PlaytimeState(day_key="2026-08-28", seconds=100.0),
"access": READABLE,
"session": SessionView(available=False),
}
kwargs.update(over)
with (
patch.object(view, "load_state", return_value=kwargs["stored"]),
patch.object(view, "state_access", return_value=kwargs["access"]),
patch.object(view, "last_verdict", return_value=kwargs["session"]),
patch.object(view, "mounted_targets", return_value=set()),
):
return build_budget_snapshot()
def test_reports_a_readable_day(self) -> None:
snapshot = self._snapshot()
assert snapshot["readable"] is True
assert snapshot["state_status"] == READABLE
assert snapshot["error"] is None
assert snapshot["today"]["gaming_day"] == "2026-08-28"
def test_reports_denied_state(self) -> None:
snapshot = self._snapshot(stored=None, access=DENIED)
assert snapshot["readable"] is False
assert snapshot["state_status"] == DENIED
assert "restart" in snapshot["error"]
assert snapshot["today"] is None
# The rest of the payload still comes through.
assert snapshot["rules"]["budget_seconds"] == 8 * 3600
def test_reports_missing_state(self) -> None:
snapshot = self._snapshot(stored=None, access=MISSING)
assert snapshot["state_status"] == MISSING
assert "No gaming time" in snapshot["error"]
def test_a_file_that_opens_but_will_not_parse_is_corrupt(self) -> None:
snapshot = self._snapshot(stored=None, access=READABLE)
assert snapshot["state_status"] == CORRUPT
assert "tampering" in snapshot["error"]
def test_includes_the_live_session(self) -> None:
session = SessionView(
observed_at="2026-08-28T20:42:19+02:00",
state="engaged",
reason="engaged",
causes=["focus"],
idle_seconds=2.0,
screen_held=False,
games=[RunningGame(pid=7, name="hollow_knight")],
available=True,
)
with patch.object(
view.State, "load", return_value=view.State(current_game_name="HK")
):
block = self._snapshot(session=session)["session"]
assert block["available"] is True
assert block["game_name"] == "HK"
assert block["qualifying_count"] == 1
assert block["processes"] == [{"pid": 7, "name": "hollow_knight"}]
def test_includes_recorded_history(self) -> None:
with patch.object(
view,
"load_history",
return_value=[HistoryDay(day="2026-08-27", seconds=1.5)],
):
snapshot = self._snapshot()
assert snapshot["history"] == [{"day": "2026-08-27", "seconds": 1.5}]
def test_leaks_no_secret(self) -> None:
"""The API key must never cross the HTTP boundary."""
with patch.object(Config, "load", return_value=Config(steam_api_key="SEKRET")):
snapshot = self._snapshot()
assert "SEKRET" not in json.dumps(snapshot)
assert set(snapshot["rules"]) == {
"budget_seconds",
"enforcement",
"counts_launchers",
"engagement_gate",
"idle_grace_seconds",
"require_game_focus",
"warn_at",
"demo",
"masked_launchers",
}

50
steam_backlog_enforcer/tests/test_config.py
View File

@ -6,12 +6,9 @@ import json
from typing import TYPE_CHECKING, Any
from unittest.mock import patch
import pytest
from steam_backlog_enforcer.config import (
Config,
State,
_atomic_write,
load_snapshot,
save_snapshot,
)
@ -20,53 +17,6 @@ if TYPE_CHECKING:
from pathlib import Path
class TestAtomicWrite:
"""Tests for _atomic_write."""
def test_writes_file(self, tmp_path: Path) -> None:
"""Test writes file."""
target = tmp_path / "out.json"
_atomic_write(target, '{"key": "value"}\n')
assert target.read_text(encoding="utf-8") == '{"key": "value"}\n'
def test_creates_parent_dirs(self, tmp_path: Path) -> None:
"""Test creates parent dirs."""
target = tmp_path / "sub" / "deep" / "out.json"
_atomic_write(target, "data")
assert target.read_text(encoding="utf-8") == "data"
def test_cleanup_on_write_error(self, tmp_path: Path) -> None:
"""Test cleanup on write error."""
target = tmp_path / "out.json"
with (
patch(
"steam_backlog_enforcer.config.os.write",
side_effect=OSError("disk full"),
),
pytest.raises(OSError, match="disk full"),
):
_atomic_write(target, "data")
assert not target.exists()
tmp_files = list(tmp_path.glob("*.tmp"))
assert tmp_files == []
def test_cleanup_on_replace_error(self, tmp_path: Path) -> None:
"""Test cleanup on replace error."""
target = tmp_path / "out.json"
with (
patch.object(
type(target),
"replace",
side_effect=OSError("no perm"),
),
pytest.raises(OSError, match="no perm"),
):
_atomic_write(target, "data")
assert not target.exists()
tmp_files = list(tmp_path.glob("*.tmp"))
assert tmp_files == []
class TestConfig:
"""Tests for Config dataclass."""

4
steam_backlog_enforcer/tests/test_config_interactive_setup.py
View File

@ -7,9 +7,7 @@ from unittest.mock import patch
import pytest
from steam_backlog_enforcer.config import (
interactive_setup,
)
from steam_backlog_enforcer._config_setup import interactive_setup
if TYPE_CHECKING:
from pathlib import Path

28
steam_backlog_enforcer/tests/test_mcp_gaming_time.py
View File

@ -4,26 +4,28 @@ from __future__ import annotations
from unittest.mock import patch
from steam_backlog_enforcer import _budget_view as budget_view
from steam_backlog_enforcer import _mcp
from steam_backlog_enforcer import _mcp_gaming_time as mcp_gaming_time
from steam_backlog_enforcer import _mcp_server as mcp_server
from steam_backlog_enforcer._playtime_state import PlaytimeState
class TestGetGamingTime:
def test_reports_unrecorded_state(self) -> None:
with (
patch.object(mcp_gaming_time, "load_state", return_value=None),
patch.object(mcp_gaming_time, "mounted_targets", return_value=set()),
patch.object(budget_view, "load_state", return_value=None),
patch.object(budget_view, "mounted_targets", return_value=set()),
):
out = mcp_gaming_time.get_gaming_time()
assert out["recorded"] is False
assert out["budget_seconds"] == 8 * 3600
def test_reports_usage(self) -> None:
stored = mcp_gaming_time.PlaytimeState(day_key="2026-07-27", seconds=100.0)
stored = PlaytimeState(day_key="2026-07-27", seconds=100.0)
with (
patch.object(mcp_gaming_time, "load_state", return_value=stored),
patch.object(mcp_gaming_time, "mounted_targets", return_value=set()),
patch.object(budget_view, "load_state", return_value=stored),
patch.object(budget_view, "mounted_targets", return_value=set()),
):
out = mcp_gaming_time.get_gaming_time()
assert out["recorded"] is True
@ -33,20 +35,20 @@ class TestGetGamingTime:
assert out["blocked"] is False
def test_remaining_never_goes_negative(self) -> None:
stored = mcp_gaming_time.PlaytimeState(day_key="d", seconds=10**9)
stored = PlaytimeState(day_key="d", seconds=10**9)
with (
patch.object(mcp_gaming_time, "load_state", return_value=stored),
patch.object(mcp_gaming_time, "mounted_targets", return_value=set()),
patch.object(budget_view, "load_state", return_value=stored),
patch.object(budget_view, "mounted_targets", return_value=set()),
):
out = mcp_gaming_time.get_gaming_time()
assert out["seconds_remaining"] == 0.0
def test_lists_masked_launchers(self) -> None:
stored = mcp_gaming_time.PlaytimeState(day_key="d", seconds=1.0, blocked_at=2.0)
stored = PlaytimeState(day_key="d", seconds=1.0, blocked_at=2.0)
with (
patch.object(mcp_gaming_time, "load_state", return_value=stored),
patch.object(budget_view, "load_state", return_value=stored),
patch.object(
mcp_gaming_time, "mounted_targets", return_value={"/usr/bin/steam"}
budget_view, "mounted_targets", return_value={"/usr/bin/steam"}
),
):
out = mcp_gaming_time.get_gaming_time()
@ -56,8 +58,8 @@ class TestGetGamingTime:
def test_leaks_no_secret(self) -> None:
"""No Config secret may cross the MCP boundary."""
with (
patch.object(mcp_gaming_time, "load_state", return_value=None),
patch.object(mcp_gaming_time, "mounted_targets", return_value=set()),
patch.object(budget_view, "load_state", return_value=None),
patch.object(budget_view, "mounted_targets", return_value=set()),
):
out = mcp_gaming_time.get_gaming_time()
assert "steam_api_key" not in out

6
steam_backlog_enforcer/tests/test_playtime.py
View File

@ -87,6 +87,12 @@ class TestSaveLoadRoundTrip:
loaded = load_state(demo=False)
assert loaded == original
def test_written_world_readable(self) -> None:
# The daemon runs as root and mkstemp would leave this 0600, which made
# every unprivileged reader see "no state recorded yet".
save_state(PlaytimeState(day_key="d", seconds=1.0), demo=False)
assert state_path(demo=False).stat().st_mode & 0o777 == 0o644
def test_demo_and_production_are_independent(self) -> None:
save_state(PlaytimeState(day_key="prod", seconds=10.0), demo=False)
save_state(PlaytimeState(day_key="demo", seconds=20.0), demo=True)

155
steam_backlog_enforcer/tests/test_playtime_history.py Normal file
View File

@ -0,0 +1,155 @@
"""Tests for _playtime_history module — 100% branch coverage."""
from __future__ import annotations
import json
from unittest.mock import patch
from steam_backlog_enforcer import _playtime_history as history_mod
from steam_backlog_enforcer._playtime_history import (
HistoryWriter,
load_history,
record_day,
)
from steam_backlog_enforcer._playtime_state import PlaytimeState
_PKG = "steam_backlog_enforcer._playtime_history"
def _stored() -> dict[str, float]:
return json.loads(history_mod.HISTORY_FILE.read_text(encoding="utf-8"))["days"]
class TestRecordDay:
"""Tests for record_day."""
def test_creates_and_upserts(self) -> None:
record_day("2026-08-27", 100.0)
record_day("2026-08-28", 200.0)
record_day("2026-08-28", 250.0)
assert _stored() == {"2026-08-27": 100.0, "2026-08-28": 250.0}
def test_written_world_readable(self) -> None:
record_day("2026-08-28", 1.0)
assert history_mod.HISTORY_FILE.stat().st_mode & 0o777 == 0o644
def test_trims_to_the_retention_window(self) -> None:
with patch(f"{_PKG}._MAX_DAYS", 2):
for day in ("2026-08-26", "2026-08-27", "2026-08-28"):
record_day(day, 1.0)
assert sorted(_stored()) == ["2026-08-27", "2026-08-28"]
class TestLoadHistory:
"""Tests for load_history and its guards."""
def test_missing_file_is_empty(self) -> None:
assert load_history() == []
def test_returns_oldest_first_and_limited(self) -> None:
for day, seconds in (
("2026-08-26", 1.0),
("2026-08-27", 2.0),
("2026-08-28", 3.0),
):
record_day(day, seconds)
days = load_history(2)
assert [d.day for d in days] == ["2026-08-27", "2026-08-28"]
assert days[0].seconds == 2.0
def test_zero_limit_returns_nothing(self) -> None:
record_day("2026-08-28", 1.0)
assert load_history(0) == []
def test_unparseable_file_is_ignored(self) -> None:
history_mod.HISTORY_FILE.write_text("{not json", encoding="utf-8")
assert load_history() == []
def test_unreadable_file_is_ignored(self) -> None:
record_day("2026-08-28", 1.0)
with patch("pathlib.Path.read_text", side_effect=PermissionError):
assert load_history() == []
def test_unknown_schema_is_ignored(self) -> None:
history_mod.HISTORY_FILE.write_text(
json.dumps({"schema_version": 99, "days": {}}), encoding="utf-8"
)
assert load_history() == []
def test_non_mapping_payload_is_ignored(self) -> None:
history_mod.HISTORY_FILE.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
assert load_history() == []
def test_non_mapping_days_is_ignored(self) -> None:
history_mod.HISTORY_FILE.write_text(
json.dumps({"schema_version": 1, "days": ["nope"]}), encoding="utf-8"
)
assert load_history() == []
def test_bad_entries_are_dropped(self) -> None:
# The file is neither root-owned nor immutable, so anything can be in it.
history_mod.HISTORY_FILE.write_text(
json.dumps(
{
"schema_version": 1,
"days": {"2026-08-28": 5, "2026-08-27": "x", "9": None},
}
),
encoding="utf-8",
)
assert [(d.day, d.seconds) for d in load_history()] == [("2026-08-28", 5.0)]
class TestHistoryWriter:
"""Tests for the throttled writer."""
def test_records_a_new_day_immediately(self) -> None:
writer = HistoryWriter()
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False)
assert _stored() == {"2026-08-28": 10.0}
def test_skips_a_small_move(self) -> None:
writer = HistoryWriter()
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False)
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=12.0), demo=False)
assert _stored() == {"2026-08-28": 10.0}
def test_records_a_large_move(self) -> None:
writer = HistoryWriter()
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False)
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=90.0), demo=False)
assert _stored() == {"2026-08-28": 90.0}
def test_records_a_refund(self) -> None:
# backdate() can move the counter *down* by the idle grace.
writer = HistoryWriter()
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=400.0), demo=False)
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=100.0), demo=False)
assert _stored() == {"2026-08-28": 100.0}
def test_records_the_new_day_after_rollover(self) -> None:
writer = HistoryWriter()
writer.observe(PlaytimeState(day_key="2026-08-27", seconds=500.0), demo=False)
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=0.0), demo=False)
assert _stored() == {"2026-08-27": 500.0, "2026-08-28": 0.0}
def test_demo_runs_are_not_recorded(self) -> None:
HistoryWriter().observe(
PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=True
)
assert not history_mod.HISTORY_FILE.exists()
def test_stateless_record_is_not_written(self) -> None:
HistoryWriter().observe(PlaytimeState(day_key="", seconds=10.0), demo=False)
assert not history_mod.HISTORY_FILE.exists()
def test_write_failure_does_not_raise(self) -> None:
writer = HistoryWriter()
with patch(f"{_PKG}.record_day", side_effect=OSError("read-only")):
writer.observe(
PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False
)
assert not history_mod.HISTORY_FILE.exists()
# Not remembered as written, so the next attempt tries again.
writer.observe(PlaytimeState(day_key="2026-08-28", seconds=10.0), demo=False)
assert _stored() == {"2026-08-28": 10.0}

9
steam_backlog_enforcer/tests/test_playtime_log.py
View File

@ -13,6 +13,7 @@ from unittest.mock import patch
import pytest
from steam_backlog_enforcer import _playtime_log as log_mod
from steam_backlog_enforcer._engagement_types import (
CAUSE_IDLE,
STATE_ENGAGED,
@ -21,8 +22,6 @@ from steam_backlog_enforcer._engagement_types import (
EngagementVerdict,
)
from steam_backlog_enforcer._playtime_log import (
BUDGET_DEMO_LOG_FILE,
BUDGET_LOG_FILE,
EVENT_DETECTOR_FAILURE,
EVENT_HEARTBEAT,
EVENT_VERDICT_CHANGE,
@ -57,14 +56,14 @@ def _records(path: Path) -> list[dict]:
class TestBudgetLogPath:
def test_production_run_journals_to_the_real_trail(self) -> None:
assert budget_log_path(demo=False) == BUDGET_LOG_FILE
assert budget_log_path(demo=False) == log_mod.BUDGET_LOG_FILE
def test_demo_run_cannot_reach_the_production_trail(self) -> None:
# A demo bills against a 60-second budget. Letting those records land
# in the production log makes the log unusable for the one job it has:
# reconstructing what the real budget actually did.
assert budget_log_path(demo=True) == BUDGET_DEMO_LOG_FILE
assert budget_log_path(demo=True) != BUDGET_LOG_FILE
assert budget_log_path(demo=True) == log_mod.BUDGET_DEMO_LOG_FILE
assert budget_log_path(demo=True) != log_mod.BUDGET_LOG_FILE
class TestBudgetLog:

121
steam_backlog_enforcer/tests/test_playtime_policy.py Normal file
View File

@ -0,0 +1,121 @@
"""Tests for the budget policy decision — warn, cut off, sustain.
Split out of ``test_playtime_tick.py`` to hold both files inside the
250-line cap. ``_policy`` is a pure decision over state and rules; the tick
tests exercise the loop that feeds it.
"""
from contextlib import ExitStack
from datetime import (
datetime,
timedelta,
timezone,
)
from unittest.mock import patch
import pytest
from steam_backlog_enforcer._playtime import (
_policy,
)
from steam_backlog_enforcer._playtime_state import (
PlaytimeState,
rules_for,
)
from steam_backlog_enforcer.config import Config
PKG = "steam_backlog_enforcer._playtime"
LOCAL = timezone(timedelta(hours=2))
NOW = datetime(2026, 7, 27, 20, 0, tzinfo=LOCAL)
TODAY = "2026-07-27"
@pytest.fixture
def quiet_tick() -> object:
"""Neutralise every side effect the tick can produce."""
with ExitStack() as stack:
# Each name is patched in the module that *resolves* it, not in the
# package that re-exports it -- otherwise the patch never bites.
cutoff = "steam_backlog_enforcer._playtime_cutoff"
mocks = {
name: stack.enter_context(patch(f"{where}.{name}"))
for name, where in (
("reconcile", PKG),
("request_steam_shutdown", cutoff),
("kill_gaming_processes", cutoff),
("notify_desktop_user", cutoff),
("mounted_targets", PKG),
)
}
mocks["mounted_targets"].return_value = set()
stack.enter_context(patch(f"{PKG}.is_total_block_active", return_value=False))
yield mocks
def _rules(*, demo: bool = False, **cfg: object) -> object:
return rules_for(Config(**cfg), demo=demo)
class TestPolicyBelowBudget:
def test_releases_and_warns(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=8 * 3600 - 3600)
with (
patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec,
patch(
"steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
) as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_notify.assert_called_once()
assert out.warned_seconds == [3600]
def test_no_warning_when_far_from_budget(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=0.0)
with (
patch("steam_backlog_enforcer._playtime.reconcile"),
patch(
"steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
) as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_notify.assert_not_called()
assert out.warned_seconds == []
class TestPolicyAtOrOverBudget:
def test_engages_the_cutoff_when_not_yet_blocked(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch("steam_backlog_enforcer._playtime._begin_cutoff") as mock_begin,
patch("steam_backlog_enforcer._playtime._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_called_once()
mock_sustain.assert_not_called()
def test_sustains_the_block_once_engaged(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6, blocked_at=1.0)
with (
patch("steam_backlog_enforcer._playtime._begin_cutoff") as mock_begin,
patch("steam_backlog_enforcer._playtime._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_not_called()
mock_sustain.assert_called_once()
class TestPolicyEnforcementDisabled:
def test_releases_but_does_not_block(self) -> None:
"""Disabling must never come to mean 'blocked forever'."""
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec,
patch(
"steam_backlog_enforcer._playtime_cutoff.request_steam_shutdown"
) as mock_shutdown,
):
out = _policy(state, _rules(playtime_enforcement=False), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_shutdown.assert_not_called()
assert out.is_blocked() is False

23
steam_backlog_enforcer/tests/test_playtime_procs.py
View File

@ -11,6 +11,7 @@ from steam_backlog_enforcer._playtime_notify import (
from steam_backlog_enforcer._playtime_procs import (
_match_cmdline,
get_pids_by_cmdline_names,
process_name,
qualifying_pids,
)
from steam_backlog_enforcer._playtime_state import (
@ -181,3 +182,25 @@ class TestHumanise:
def test_thirty_minutes(self) -> None:
assert _humanise(1800) == "30 minutes"
class TestProcessName:
"""Tests for resolving a PID to its program name."""
def _comm(self, tmp_path: Path, pid: int, name: str) -> None:
entry = tmp_path / "playtime_proc" / str(pid)
entry.mkdir(parents=True, exist_ok=True)
(entry / "comm").write_text(name, encoding="utf-8")
def test_reads_comm(self, tmp_path: Path) -> None:
self._comm(tmp_path, 42, "hollow_knight\n")
assert process_name(42) == "hollow_knight"
def test_missing_pid_is_none(self) -> None:
# A PID recorded earlier may have exited, and the number may since have
# been recycled — "gone" is the only safe answer.
assert process_name(999999) is None
def test_empty_comm_is_none(self, tmp_path: Path) -> None:
self._comm(tmp_path, 43, "\n")
assert process_name(43) is None

7
steam_backlog_enforcer/tests/test_playtime_session.py
View File

@ -4,10 +4,9 @@ from __future__ import annotations
from unittest.mock import patch
from steam_backlog_enforcer import _playtime_log as log_mod
from steam_backlog_enforcer._playtime_engagement import EngagementTracker
from steam_backlog_enforcer._playtime_log import (
BUDGET_DEMO_LOG_FILE,
BUDGET_LOG_FILE,
TickJournal,
)
from steam_backlog_enforcer._playtime_session import new_session
@ -43,5 +42,5 @@ class TestNewSession:
):
live = new_session()
demo = new_session(demo=True)
assert demo.journal._log._path == BUDGET_DEMO_LOG_FILE
assert live.journal._log._path == BUDGET_LOG_FILE
assert demo.journal._log._path == log_mod.BUDGET_DEMO_LOG_FILE
assert live.journal._log._path == log_mod.BUDGET_LOG_FILE

100
steam_backlog_enforcer/tests/test_playtime_tick.py
View File

@ -11,7 +11,6 @@ from unittest.mock import patch
import pytest
from steam_backlog_enforcer._playtime import (
_policy,
_state_or_recover,
playtime_tick,
)
@ -87,71 +86,6 @@ class TestStateOrRecover:
assert out.is_blocked() is True
class TestPolicyBelowBudget:
def test_releases_and_warns(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=8 * 3600 - 3600)
with (
patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec,
patch(
"steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
) as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_notify.assert_called_once()
assert out.warned_seconds == [3600]
def test_no_warning_when_far_from_budget(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=0.0)
with (
patch("steam_backlog_enforcer._playtime.reconcile"),
patch(
"steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
) as mock_notify,
):
out = _policy(state, _rules(), now=NOW)
mock_notify.assert_not_called()
assert out.warned_seconds == []
class TestPolicyAtOrOverBudget:
def test_engages_the_cutoff_when_not_yet_blocked(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch("steam_backlog_enforcer._playtime._begin_cutoff") as mock_begin,
patch("steam_backlog_enforcer._playtime._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_called_once()
mock_sustain.assert_not_called()
def test_sustains_the_block_once_engaged(self) -> None:
state = PlaytimeState(day_key=TODAY, seconds=10**6, blocked_at=1.0)
with (
patch("steam_backlog_enforcer._playtime._begin_cutoff") as mock_begin,
patch("steam_backlog_enforcer._playtime._sustain_block") as mock_sustain,
):
_policy(state, _rules(), now=NOW)
mock_begin.assert_not_called()
mock_sustain.assert_called_once()
class TestPolicyEnforcementDisabled:
def test_releases_but_does_not_block(self) -> None:
"""Disabling must never come to mean 'blocked forever'."""
state = PlaytimeState(day_key=TODAY, seconds=10**6)
with (
patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec,
patch(
"steam_backlog_enforcer._playtime_cutoff.request_steam_shutdown"
) as mock_shutdown,
):
out = _policy(state, _rules(playtime_enforcement=False), now=NOW)
mock_rec.assert_called_once_with(should_block=False)
mock_shutdown.assert_not_called()
assert out.is_blocked() is False
class TestPlaytimeTick:
def test_accumulates_while_a_game_runs(self, quiet_tick: dict) -> None:
save_state(
@ -166,6 +100,40 @@ class TestPlaytimeTick:
playtime_tick(Config(), interval=3.0, session=fake_session())
assert load_state(demo=False).seconds == 3.0
def test_records_the_day_in_the_history(self, quiet_tick: dict) -> None:
del quiet_tick
save_state(
PlaytimeState(day_key=TODAY, last_tick_at=NOW.timestamp() - 3.0),
demo=False,
)
session = fake_session()
with (
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, session=session)
assert session.history.recorded == [(TODAY, 3.0)]
def test_records_the_day_even_under_a_total_block(self, quiet_tick: dict) -> None:
"""The counter still needs a history point while the block is up."""
del quiet_tick
save_state(
PlaytimeState(
day_key=TODAY, seconds=5.0, last_tick_at=NOW.timestamp() - 3.0
),
demo=False,
)
session = fake_session()
with (
patch(f"{PKG}.is_total_block_active", return_value=True),
patch(f"{PKG}.qualifying_pids", return_value={7}),
patch(f"{PKG}.datetime") as mock_dt,
):
mock_dt.now.return_value = NOW
playtime_tick(Config(), interval=3.0, session=session)
assert session.history.recorded == [(TODAY, 5.0)]
def test_total_block_releases_and_stops_accruing(self, quiet_tick: dict) -> None:
"""Our mounts would make the total block's `pacman -R steam` fail."""
save_state(

33
steam_backlog_enforcer/tests/test_web_server.py
View File

@ -65,6 +65,39 @@ def _make_dist(
return dist
class TestBudgetEndpoint:
"""Tests for the /api/budget route."""
def test_budget_ok(self) -> None:
with (
patch(f"{_PKG}.build_budget_snapshot", return_value={"ok": True}) as build,
_running() as port,
):
status, body, ctype = _get(port, "/api/budget")
assert status == 200
assert "application/json" in ctype
assert json.loads(body) == {"ok": True}
# Production by default, and nothing to do with the dataset projection.
assert build.call_args.kwargs == {"demo": False}
def test_demo_query_reads_the_demo_run(self) -> None:
with (
patch(f"{_PKG}.build_budget_snapshot", return_value={"ok": True}) as build,
_running() as port,
):
_get(port, "/api/budget?demo=1")
assert build.call_args.kwargs == {"demo": True}
def test_budget_error(self) -> None:
with (
patch(f"{_PKG}.build_budget_snapshot", side_effect=OSError("boom")),
_running() as port,
):
status, body, _ = _get(port, "/api/budget")
assert status == 500
assert body == b"budget error"
class TestDatasetEndpoint:
"""Tests for the /api/dataset route."""

36
web/src/App.test.tsx
View File

@ -2,7 +2,7 @@ import { render, screen, waitFor, within } from '@testing-library/react'
import userEvent from '@testing-library/user-event'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import App from './App'
import { makeDataset, makeGame, makeState } from './test/factories'
import { makeBudget, makeDataset, makeGame, makeState } from './test/factories'
describe('App', () => {
beforeEach(() => {
@ -93,4 +93,38 @@ describe('App', () => {
await user.click(screen.getByRole('button', { name: /Reset to CLI defaults/i }))
expect(document.querySelector('.big')?.textContent).toBe('2')
})
it('switches to the budget tab and back', async () => {
const ds = makeDataset([makeGame({ app_id: 1, name: 'Alpha' })])
vi.stubGlobal(
'fetch',
vi.fn().mockImplementation((url: string) =>
Promise.resolve(
url.startsWith('/api/budget')
? { ok: true, json: async () => makeBudget() }
: { ok: true, json: async () => ds },
),
),
)
const user = userEvent.setup()
render(<App />)
await waitFor(() =>
expect(
screen.getByRole('heading', { name: 'Backlog Completion Planner' }),
).toBeInTheDocument(),
)
await user.click(screen.getByRole('tab', { name: 'Gaming budget' }))
await waitFor(() =>
expect(screen.getByRole('heading', { name: 'Today' })).toBeInTheDocument(),
)
expect(screen.queryByRole('table')).toBeNull()
await user.click(screen.getByRole('tab', { name: 'Backlog planner' }))
await waitFor(() =>
expect(
screen.getByRole('heading', { name: 'Backlog Completion Planner' }),
).toBeInTheDocument(),
)
})
})

151
web/src/App.tsx
View File

@ -1,128 +1,43 @@
import { useEffect, useMemo, useState } from 'react'
import { fetchDataset } from './api'
import { FilterPanel } from './components/FilterPanel'
import { GameTable } from './components/GameTable'
import { SummaryCards } from './components/SummaryCards'
import { TimelineChart } from './components/TimelineChart'
import { applyFilters } from './estimate'
import type { Filters, WebDataset, WebDefaults } from './types'
import { useState } from 'react'
import { BudgetPanel } from './components/BudgetPanel'
import { PlannerView } from './components/PlannerView'
import type { TabId } from './types'
function defaultFilters(d: WebDefaults): Filters {
return {
minCountComp: d.min_count_comp,
minComp100: d.min_comp_100_polls,
minConfidenceSum: d.min_confidence_sum,
protonMode: 'playable',
protonMinTier: d.min_playable_tier,
protonTreatMissingAsPass: true,
dailyHours: 4,
basis: 'leisure',
maxHoursPerGame: 0,
playtimeMode: 'all',
includeNoData: false,
fallbackHours: 20,
excluded: new Set<number>(),
search: '',
targetDate: '',
}
/**
* `?demo=1` points the budget tab at the short-budget demo run, which is the
* only way to watch the cutoff engage without spending a real day's budget.
*/
function demoRequested(): boolean {
return new URLSearchParams(window.location.search).get('demo') === '1'
}
const TABS: { id: TabId; label: string }[] = [
{ id: 'planner', label: 'Backlog planner' },
{ id: 'budget', label: 'Gaming budget' },
]
function App() {
const [dataset, setDataset] = useState<WebDataset | null>(null)
const [filters, setFilters] = useState<Filters | null>(null)
const [error, setError] = useState<string | null>(null)
useEffect(() => {
fetchDataset()
.then((d) => {
setDataset(d)
setFilters(defaultFilters(d.defaults))
})
.catch((e: unknown) => setError(e instanceof Error ? e.message : String(e)))
}, [])
const result = useMemo(
() => (dataset && filters ? applyFilters(dataset, filters) : null),
[dataset, filters],
)
if (error) {
return (
<div className="status">
<h1>Steam Backlog Enforcer</h1>
<p className="error">Could not load data: {error}</p>
<p className="hint">
Is the backend running? Start it with <code>./run.sh serve</code>.
</p>
</div>
)
}
if (!dataset || !filters || !result) {
return (
<div className="status">
<h1>Steam Backlog Enforcer</h1>
<p className="hint">Loading your backlog…</p>
</div>
)
}
const update = (patch: Partial<Filters>) => setFilters({ ...filters, ...patch })
const toggleExclude = (appId: number) => {
const next = new Set(filters.excluded)
if (next.has(appId)) next.delete(appId)
else next.add(appId)
setFilters({ ...filters, excluded: next })
}
const tableRows = result.rows.filter((r) => r.passesFilters)
const [tab, setTab] = useState<TabId>('planner')
return (
<div className="app">
<header className="app-head">
<div>
<h1>Backlog Completion Planner</h1>
<p className="sub">
{dataset.state.current_game_name && (
<>
Currently playing <strong>{dataset.state.current_game_name}</strong>{' '}
·{' '}
</>
)}
{dataset.state.games_done_since_start} games finished since{' '}
{dataset.state.enforcement_started_at.slice(0, 10) || '—'} ·{' '}
{dataset.games.length} candidates
</p>
</div>
</header>
<div className="layout">
<FilterPanel
filters={filters}
defaults={dataset.defaults}
update={update}
onReset={() => setFilters(defaultFilters(dataset.defaults))}
/>
<main className="content">
<SummaryCards
result={result}
filters={filters}
state={dataset.state}
presets={dataset.defaults.hours_per_day_presets}
defaultQualifying={dataset.default_summary.qualifying}
paceVsHltb={dataset.pace_vs_hltb}
/>
<TimelineChart result={result} filters={filters} state={dataset.state} />
<GameTable
rows={tableRows}
search={filters.search}
onSearch={(s) => update({ search: s })}
onToggleExclude={toggleExclude}
/>
</main>
</div>
<nav className="tabs" role="tablist">
{TABS.map(({ id, label }) => (
<button
key={id}
type="button"
role="tab"
aria-selected={tab === id}
className={tab === id ? 'tab active' : 'tab'}
onClick={() => setTab(id)}
>
{label}
</button>
))}
</nav>
{/* The budget panel only mounts on its own tab, so the planner never
pays for its polling — and vice versa. */}
{tab === 'budget' ? <BudgetPanel demo={demoRequested()} /> : <PlannerView />}
</div>
)
}

25
web/src/api.ts
View File

@ -1,10 +1,25 @@
import type { WebDataset } from './types'
import type { BudgetSnapshot, WebDataset } from './types'
/** Fetch the projected dataset from the Python backend. */
export async function fetchDataset(): Promise<WebDataset> {
const resp = await fetch('/api/dataset')
/** Fetch and decode a JSON document from the Python backend. */
async function getJson<T>(path: string): Promise<T> {
const resp = await fetch(path)
if (!resp.ok) {
throw new Error(`API returned ${resp.status} ${resp.statusText}`)
}
return (await resp.json()) as WebDataset
return (await resp.json()) as T
}
/** Fetch the projected dataset from the Python backend. */
export function fetchDataset(): Promise<WebDataset> {
return getJson<WebDataset>('/api/dataset')
}
/**
* Fetch the current gaming-budget snapshot.
*
* `demo` reads the short-budget demo run instead of production, which is how
* the cutoff can be watched in the browser without spending a real day.
*/
export function fetchBudget(demo: boolean): Promise<BudgetSnapshot> {
return getJson<BudgetSnapshot>(`/api/budget${demo ? '?demo=1' : ''}`)
}

43
web/src/components/BudgetHistoryChart.test.tsx Normal file
View File

@ -0,0 +1,43 @@
import { render, screen } from '@testing-library/react'
import { describe, expect, it } from 'vitest'
import { BudgetHistoryChart } from './BudgetHistoryChart'
describe('BudgetHistoryChart', () => {
it('explains itself when nothing has been recorded', () => {
render(<BudgetHistoryChart days={[]} budgetSeconds={28800} />)
expect(screen.getByText(/No budget history recorded yet/)).toBeInTheDocument()
expect(document.querySelector('svg')).toBeNull()
})
it('draws a single day without filling the plot', () => {
render(
<BudgetHistoryChart days={[{ day: '2026-08-28', seconds: 14400 }]} budgetSeconds={28800} />,
)
const bars = document.querySelectorAll('rect.bar')
expect(bars).toHaveLength(1)
// Scaled against the budget, half a budget is half the plot height.
expect(Number(bars[0].getAttribute('height'))).toBeLessThan(
Number(document.querySelector('svg')?.getAttribute('viewBox')?.split(' ')[3]) / 2,
)
expect(screen.getByText('08-28')).toBeInTheDocument()
})
it('marks a day that reached the budget', () => {
render(
<BudgetHistoryChart
days={[
{ day: '2026-08-27', seconds: 3600 },
{ day: '2026-08-28', seconds: 28800 },
]}
budgetSeconds={28800}
/>,
)
expect(document.querySelectorAll('rect.bar')).toHaveLength(2)
expect(document.querySelectorAll('rect.bar.over')).toHaveLength(1)
})
it('survives a zero budget and zero seconds', () => {
render(<BudgetHistoryChart days={[{ day: '2026-08-28', seconds: 0 }]} budgetSeconds={0} />)
expect(document.querySelectorAll('rect')).toHaveLength(1)
})
})

86
web/src/components/BudgetHistoryChart.tsx Normal file
View File

@ -0,0 +1,86 @@
import { fmtDuration } from '../format'
import type { BudgetHistoryDay } from '../types'
interface Props {
days: BudgetHistoryDay[]
budgetSeconds: number
}
const W = 820
const H = 220
const PAD = { top: 16, right: 20, bottom: 36, left: 56 }
export function BudgetHistoryChart({ days, budgetSeconds }: Props) {
if (days.length === 0) {
return (
<div className="chart">
<h2>Last 14 days</h2>
<p className="hint">
No budget history recorded yet — a bar appears for each gaming day from now on.
</p>
</div>
)
}
// Scaled against the budget rather than the tallest bar, so a single day
// reads as "most of the allowance" instead of a lone full-height bar that
// means nothing on its own.
const maxSeconds =
Math.max(budgetSeconds, ...days.map((d) => d.seconds)) || 1
const plotW = W - PAD.left - PAD.right
const plotH = H - PAD.top - PAD.bottom
const sy = (seconds: number) => PAD.top + plotH - (seconds / maxSeconds) * plotH
const slot = plotW / days.length
const barW = slot * 0.7
const yTicks = [0, 0.5, 1].map((f) => f * maxSeconds)
return (
<div className="chart">
<h2>Last 14 days</h2>
<svg viewBox={`0 0 ${W} ${H}`} className="chart-svg" role="img">
{yTicks.map((s) => (
<g key={`y${s}`}>
<line x1={PAD.left} y1={sy(s)} x2={W - PAD.right} y2={sy(s)} className="grid" />
<text x={PAD.left - 8} y={sy(s) + 4} className="axis-label end">
{fmtDuration(s)}
</text>
</g>
))}
<line
x1={PAD.left}
y1={sy(budgetSeconds)}
x2={W - PAD.right}
y2={sy(budgetSeconds)}
className="budget-line"
/>
{days.map((d, i) => (
<rect
key={d.day}
x={PAD.left + i * slot + (slot - barW) / 2}
y={sy(d.seconds)}
width={barW}
height={PAD.top + plotH - sy(d.seconds)}
className={d.seconds >= budgetSeconds ? 'bar over' : 'bar'}
>
<title>{`${d.day}: ${fmtDuration(d.seconds)}`}</title>
</rect>
))}
{days.map((d, i) => (
<text
key={`x${d.day}`}
x={PAD.left + i * slot + slot / 2}
y={H - 12}
className="axis-label mid"
>
{d.day.slice(5)}
</text>
))}
</svg>
<p className="hint">
Billed gaming time per day. The dashed line is the daily budget; today&apos;s bar
can still move.
</p>
</div>
)
}

71
web/src/components/BudgetLiveCard.test.tsx Normal file
View File

@ -0,0 +1,71 @@
import { render, screen } from '@testing-library/react'
import { describe, expect, it } from 'vitest'
import { makeBudgetSession } from '../test/factories'
import { BudgetLiveCard } from './BudgetLiveCard'
describe('BudgetLiveCard', () => {
it('reports an engaged session with its game', () => {
render(<BudgetLiveCard session={makeBudgetSession()} />)
expect(screen.getByText(/Billing — you are playing/)).toBeInTheDocument()
expect(screen.getByText('Hollow Knight')).toBeInTheDocument()
expect(screen.getByText(/3 qualifying processes/)).toBeInTheDocument()
expect(screen.getByText(/Idle 2s/)).toBeInTheDocument()
expect(screen.getByText(/Screen held: no/)).toBeInTheDocument()
})
it('explains why a session is paused', () => {
render(
<BudgetLiveCard
session={makeBudgetSession({
state: 'paused',
reason: 'focus',
causes: ['focus', 'idle'],
screen_held: true,
})}
/>,
)
expect(screen.getByText(/Not billing — paused/)).toBeInTheDocument()
expect(
screen.getByText(/the game window is not focused, no input for longer/),
).toBeInTheDocument()
expect(screen.getByText(/Screen held: yes/)).toBeInTheDocument()
})
it('falls back to the raw cause and verdict when unrecognised', () => {
render(
<BudgetLiveCard
session={makeBudgetSession({ state: 'weird', causes: ['gremlins'] })}
/>,
)
expect(screen.getByText('weird')).toBeInTheDocument()
expect(screen.getByText(/gremlins/)).toBeInTheDocument()
})
it('omits the game line when nothing qualifies', () => {
render(
<BudgetLiveCard
session={makeBudgetSession({
state: 'not_applicable',
qualifying_count: 0,
processes: [],
idle_seconds: null,
screen_held: null,
})}
/>,
)
expect(screen.getByText(/no game running/)).toBeInTheDocument()
expect(screen.queryByText(/qualifying processes/)).toBeNull()
expect(screen.queryByText(/Idle/)).toBeNull()
})
it('handles a qualifying session with no assigned game name', () => {
render(<BudgetLiveCard session={makeBudgetSession({ game_name: '' })} />)
expect(screen.getByText(/3 qualifying processes/)).toBeInTheDocument()
expect(screen.queryByText('Hollow Knight')).toBeNull()
})
it('says when no verdict has been logged', () => {
render(<BudgetLiveCard session={makeBudgetSession({ available: false })} />)
expect(screen.getByText(/No verdict has been logged yet/)).toBeInTheDocument()
})
})

64
web/src/components/BudgetLiveCard.tsx Normal file
View File

@ -0,0 +1,64 @@
import { fmtAgo, fmtDuration } from '../format'
import type { BudgetSession } from '../types'
interface Props {
session: BudgetSession
}
const VERDICT_LABELS: Record<string, string> = {
engaged: 'Billing — you are playing',
paused: 'Not billing — paused',
not_applicable: 'Not billing — no game running',
}
const CAUSE_LABELS: Record<string, string> = {
idle: 'no input for longer than the idle grace',
focus: 'the game window is not focused',
screen_held: 'the screen is locked or held',
}
function describe(values: string[], labels: Record<string, string>): string[] {
return values.map((value) => labels[value] ?? value)
}
export function BudgetLiveCard({ session }: Props) {
if (!session.available) {
return (
<section className="budget-live">
<h2>Right now</h2>
<p className="hint">No verdict has been logged yet.</p>
</section>
)
}
return (
<section className="budget-live">
<h2>Right now</h2>
<p>
<span className={`budget-verdict budget-verdict--${session.state}`}>
{VERDICT_LABELS[session.state] ?? session.state}
</span>
</p>
{session.qualifying_count > 0 && (
<p className="hint">
{session.game_name ? (
<>
Assigned game <strong>{session.game_name}</strong> ·{' '}
</>
) : null}
{session.qualifying_count} qualifying processes
</p>
)}
{session.causes.length > 0 && (
<p className="hint">Paused because {describe(session.causes, CAUSE_LABELS).join(', ')}.</p>
)}
<p className="hint">
{session.idle_seconds !== null && <>Idle {fmtDuration(session.idle_seconds)} · </>}
{session.screen_held !== null && <>Screen held: {session.screen_held ? 'yes' : 'no'} · </>}
{/* The daemon logs on change or a 5-minute heartbeat, so this reading
can legitimately be minutes old. Saying so beats implying "now". */}
as of {fmtAgo(session.observed_at)}
</p>
</section>
)
}

72
web/src/components/BudgetPanel.test.tsx Normal file
View File

@ -0,0 +1,72 @@
import { render, screen, waitFor } from '@testing-library/react'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { makeBudget } from '../test/factories'
import { BudgetPanel } from './BudgetPanel'
function stubBudget(snapshot: unknown) {
vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: true, json: async () => snapshot }))
}
describe('BudgetPanel', () => {
beforeEach(() => {
vi.restoreAllMocks()
})
afterEach(() => {
vi.unstubAllGlobals()
})
it('shows a loading line before the first response', () => {
stubBudget(makeBudget())
render(<BudgetPanel demo={false} />)
expect(screen.getByText(/Reading budget state/)).toBeInTheDocument()
})
it('renders all four sections once loaded', async () => {
stubBudget(makeBudget())
render(<BudgetPanel demo={false} />)
await waitFor(() =>
expect(screen.getByRole('heading', { name: 'Today' })).toBeInTheDocument(),
)
expect(screen.getByRole('heading', { name: 'Right now' })).toBeInTheDocument()
expect(screen.getByRole('heading', { name: 'Last 14 days' })).toBeInTheDocument()
expect(screen.getByRole('heading', { name: 'Rules in effect' })).toBeInTheDocument()
})
it('surfaces a transport failure', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue({ ok: false, status: 500, statusText: 'Server Error' }),
)
render(<BudgetPanel demo={false} />)
await waitFor(() =>
expect(screen.getByText(/Could not read budget state/)).toBeInTheDocument(),
)
})
it('names the permission problem and still shows the rest', async () => {
stubBudget(
makeBudget({
readable: false,
state_status: 'denied',
error: 'Cannot read the budget state file.',
today: null,
}),
)
render(<BudgetPanel demo={false} />)
await waitFor(() =>
expect(screen.getByText(/Cannot read the budget state file/)).toBeInTheDocument(),
)
// The log, history and config are all still readable — the panel must not
// go blank just because one file is locked down.
expect(screen.getByRole('heading', { name: 'Right now' })).toBeInTheDocument()
expect(screen.getByRole('heading', { name: 'Rules in effect' })).toBeInTheDocument()
})
it('falls back to a generic message when the backend sends none', async () => {
stubBudget(makeBudget({ readable: false, state_status: 'denied', error: null, today: null }))
render(<BudgetPanel demo={false} />)
await waitFor(() =>
expect(screen.getByText(/Budget state unavailable/)).toBeInTheDocument(),
)
})
})

50
web/src/components/BudgetPanel.tsx Normal file
View File

@ -0,0 +1,50 @@
import { useBudgetPoll } from '../useBudgetPoll'
import { BudgetHistoryChart } from './BudgetHistoryChart'
import { BudgetLiveCard } from './BudgetLiveCard'
import { BudgetRulesCard } from './BudgetRulesCard'
import { BudgetTodayCard } from './BudgetTodayCard'
interface Props {
demo: boolean
}
export function BudgetPanel({ demo }: Props) {
const { data, error } = useBudgetPoll(demo)
if (error !== null) {
return (
<div className="status">
<p className="error">Could not read budget state: {error}</p>
<p className="hint">
Is the backend running? Start it with <code>./run.sh serve</code>.
</p>
</div>
)
}
if (data === null) {
return (
<div className="status">
<p className="hint">Reading budget state…</p>
</div>
)
}
return (
<main className="content budget">
{/* A state file we cannot read must say so and still show everything we
can read — the log, the history and the config are all reachable. */}
{data.today === null ? (
<section className="budget-today">
<h2>Today</h2>
<p className="error">{data.error ?? 'Budget state unavailable.'}</p>
</section>
) : (
<BudgetTodayCard today={data.today} />
)}
<BudgetLiveCard session={data.session} />
<BudgetHistoryChart days={data.history} budgetSeconds={data.rules.budget_seconds} />
<BudgetRulesCard rules={data.rules} />
</main>
)
}

37
web/src/components/BudgetRulesCard.test.tsx Normal file
View File

@ -0,0 +1,37 @@
import { render, screen } from '@testing-library/react'
import { describe, expect, it } from 'vitest'
import { makeBudgetRules } from '../test/factories'
import { BudgetRulesCard } from './BudgetRulesCard'
describe('BudgetRulesCard', () => {
it('lists the production policy', () => {
render(<BudgetRulesCard rules={makeBudgetRules()} />)
expect(screen.getByText('8h 0m')).toBeInTheDocument()
expect(screen.getByText('on')).toBeInTheDocument()
expect(screen.getByText(/idle grace 5m/)).toBeInTheDocument()
expect(screen.getByText(/game must be focused/)).toBeInTheDocument()
expect(screen.getByText('yes')).toBeInTheDocument()
expect(screen.getByText(/1h 0m, 30m, 10m, 5m remaining/)).toBeInTheDocument()
expect(screen.queryByText(/demo/)).toBeNull()
expect(screen.queryByText('Masked now')).toBeNull()
})
it('shows the demo mode and masked launchers when they apply', () => {
render(
<BudgetRulesCard
rules={makeBudgetRules({
enforcement: false,
counts_launchers: false,
engagement_gate: false,
require_game_focus: false,
demo: true,
masked_launchers: ['/usr/bin/steam'],
})}
/>,
)
expect(screen.getByText(/demo \(short budget/)).toBeInTheDocument()
expect(screen.getByText('/usr/bin/steam')).toBeInTheDocument()
expect(screen.getByText('no')).toBeInTheDocument()
expect(screen.queryByText(/game must be focused/)).toBeNull()
})
})

45
web/src/components/BudgetRulesCard.tsx Normal file
View File

@ -0,0 +1,45 @@
import { fmtDuration } from '../format'
import type { BudgetRules } from '../types'
interface Props {
rules: BudgetRules
}
function onOff(value: boolean): string {
return value ? 'on' : 'off'
}
export function BudgetRulesCard({ rules }: Props) {
return (
<section className="budget-rules">
<h2>Rules in effect</h2>
<dl className="budget-rules-list">
<dt>Daily budget</dt>
<dd>{fmtDuration(rules.budget_seconds)}</dd>
<dt>Enforcement</dt>
<dd>{onOff(rules.enforcement)}</dd>
<dt>Engagement gate</dt>
<dd>
{onOff(rules.engagement_gate)} · idle grace {fmtDuration(rules.idle_grace_seconds)}
{rules.require_game_focus ? ' · game must be focused' : ''}
</dd>
<dt>Counts launchers</dt>
<dd>{rules.counts_launchers ? 'yes' : 'no'}</dd>
<dt>Warnings at</dt>
<dd>{rules.warn_at.map((s) => fmtDuration(s)).join(', ')} remaining</dd>
{rules.demo && (
<>
<dt>Mode</dt>
<dd>demo (short budget, separate state)</dd>
</>
)}
{rules.masked_launchers.length > 0 && (
<>
<dt>Masked now</dt>
<dd>{rules.masked_launchers.join(', ')}</dd>
</>
)}
</dl>
</section>
)
}

53
web/src/components/BudgetTodayCard.test.tsx Normal file
View File

@ -0,0 +1,53 @@
import { render, screen } from '@testing-library/react'
import { describe, expect, it } from 'vitest'
import { makeBudgetToday } from '../test/factories'
import { BudgetTodayCard } from './BudgetTodayCard'
describe('BudgetTodayCard', () => {
it('shows remaining time, usage and the next warning', () => {
render(<BudgetTodayCard today={makeBudgetToday()} />)
expect(screen.getByText('7h 0m')).toBeInTheDocument()
expect(screen.getByText(/1h 0m of 8h 0m used/)).toBeInTheDocument()
expect(screen.getByText(/Next warning at 1h 0m remaining/)).toBeInTheDocument()
expect(document.querySelector('.budget-meter--ok')).not.toBeNull()
})
it('turns the meter amber once a warning has fired', () => {
render(
<BudgetTodayCard
today={makeBudgetToday({ seconds_remaining: 900, warned_seconds: [3600, 1800] })}
/>,
)
expect(document.querySelector('.budget-meter--warn')).not.toBeNull()
})
it('says so when no warnings remain', () => {
render(
<BudgetTodayCard
today={makeBudgetToday({
next_warning_seconds: null,
warned_seconds: [3600, 1800, 600, 300],
})}
/>,
)
expect(screen.getByText(/No further warnings today/)).toBeInTheDocument()
expect(document.querySelector('.budget-meter--warn')).not.toBeNull()
})
it('shows the cutoff badge when blocked', () => {
render(
<BudgetTodayCard
today={makeBudgetToday({
blocked: true,
blocked_at: 1,
seconds_used: 28800,
seconds_remaining: 0,
fraction_used: 1,
next_warning_seconds: null,
})}
/>,
)
expect(screen.getByText(/Cutoff engaged/)).toBeInTheDocument()
expect(document.querySelector('.budget-meter--blocked')).not.toBeNull()
})
})

52
web/src/components/BudgetTodayCard.tsx Normal file
View File

@ -0,0 +1,52 @@
import { fmtDuration } from '../format'
import type { BudgetToday } from '../types'
interface Props {
today: BudgetToday
}
/** Which colour the meter takes: normal, into the warning band, or spent.
*
* Keyed on warnings already fired rather than on `next_warning_seconds`: that
* field is by construction always *below* the remaining time, so comparing the
* two could never turn the meter amber. "The enforcer has warned you" is also
* the thing the colour is actually trying to say.
*/
function meterState(today: BudgetToday): string {
if (today.blocked) return 'blocked'
return today.warned_seconds.length > 0 ? 'warn' : 'ok'
}
export function BudgetTodayCard({ today }: Props) {
return (
<section className="budget-today">
<h2>Today</h2>
<p className="budget-hero">
<strong>{fmtDuration(today.seconds_remaining)}</strong> left
</p>
<div className={`budget-meter budget-meter--${meterState(today)}`}>
<div
className="budget-meter-fill"
style={{ width: `${(today.fraction_used * 100).toFixed(1)}%` }}
/>
</div>
<p className="hint">
{fmtDuration(today.seconds_used)} of {fmtDuration(today.budget_seconds)} used
{' · '}
{Math.round(today.fraction_used * 100)}%{' · '}
gaming day {today.gaming_day} (starts {today.day_starts_at})
</p>
<p className="hint">
{today.blocked ? (
<span className="budget-badge budget-badge--blocked">
Cutoff engaged — launchers are masked
</span>
) : today.next_warning_seconds === null ? (
'No further warnings today.'
) : (
`Next warning at ${fmtDuration(today.next_warning_seconds)} remaining.`
)}
</p>
</section>
)
}

127
web/src/components/PlannerView.tsx Normal file
View File

@ -0,0 +1,127 @@
import { useEffect, useMemo, useState } from 'react'
import { fetchDataset } from '../api'
import { applyFilters } from '../estimate'
import type { Filters, WebDataset, WebDefaults } from '../types'
import { FilterPanel } from './FilterPanel'
import { GameTable } from './GameTable'
import { SummaryCards } from './SummaryCards'
import { TimelineChart } from './TimelineChart'
function defaultFilters(d: WebDefaults): Filters {
return {
minCountComp: d.min_count_comp,
minComp100: d.min_comp_100_polls,
minConfidenceSum: d.min_confidence_sum,
protonMode: 'playable',
protonMinTier: d.min_playable_tier,
protonTreatMissingAsPass: true,
dailyHours: 4,
basis: 'leisure',
maxHoursPerGame: 0,
playtimeMode: 'all',
includeNoData: false,
fallbackHours: 20,
excluded: new Set<number>(),
search: '',
targetDate: '',
}
}
/** The backlog-completion planner: the app's original single screen. */
export function PlannerView() {
const [dataset, setDataset] = useState<WebDataset | null>(null)
const [filters, setFilters] = useState<Filters | null>(null)
const [error, setError] = useState<string | null>(null)
useEffect(() => {
fetchDataset()
.then((d) => {
setDataset(d)
setFilters(defaultFilters(d.defaults))
})
.catch((e: unknown) => setError(e instanceof Error ? e.message : String(e)))
}, [])
const result = useMemo(
() => (dataset && filters ? applyFilters(dataset, filters) : null),
[dataset, filters],
)
if (error) {
return (
<div className="status">
<p className="error">Could not load data: {error}</p>
<p className="hint">
Is the backend running? Start it with <code>./run.sh serve</code>.
</p>
</div>
)
}
if (!dataset || !filters || !result) {
return (
<div className="status">
<p className="hint">Loading your backlog…</p>
</div>
)
}
const update = (patch: Partial<Filters>) => setFilters({ ...filters, ...patch })
const toggleExclude = (appId: number) => {
const next = new Set(filters.excluded)
if (next.has(appId)) next.delete(appId)
else next.add(appId)
setFilters({ ...filters, excluded: next })
}
const tableRows = result.rows.filter((r) => r.passesFilters)
return (
<>
<header className="app-head">
<div>
<h1>Backlog Completion Planner</h1>
<p className="sub">
{dataset.state.current_game_name && (
<>
Currently playing <strong>{dataset.state.current_game_name}</strong>{' '}
·{' '}
</>
)}
{dataset.state.games_done_since_start} games finished since{' '}
{dataset.state.enforcement_started_at.slice(0, 10) || '—'} ·{' '}
{dataset.games.length} candidates
</p>
</div>
</header>
<div className="layout">
<FilterPanel
filters={filters}
defaults={dataset.defaults}
update={update}
onReset={() => setFilters(defaultFilters(dataset.defaults))}
/>
<main className="content">
<SummaryCards
result={result}
filters={filters}
state={dataset.state}
presets={dataset.defaults.hours_per_day_presets}
defaultQualifying={dataset.default_summary.qualifying}
paceVsHltb={dataset.pace_vs_hltb}
/>
<TimelineChart result={result} filters={filters} state={dataset.state} />
<GameTable
rows={tableRows}
search={filters.search}
onSearch={(s) => update({ search: s })}
onToggleExclude={toggleExclude}
/>
</main>
</div>
</>
)
}

23
web/src/format.test.ts
View File

@ -1,6 +1,8 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import {
daysUntil,
fmtAgo,
fmtDuration,
fmtEta,
fmtHours,
fmtHoursPrecise,
@ -66,3 +68,24 @@ describe('daysUntil', () => {
expect(daysUntil('2026-06-08')).toBe(10)
})
})
describe('fmtDuration', () => {
it('formats hours, minutes and seconds', () => {
expect(fmtDuration(26846)) .toBe('7h 27m')
expect(fmtDuration(2640)).toBe('44m')
expect(fmtDuration(38)).toBe('38s')
expect(fmtDuration(-5)).toBe('0s')
})
})
describe('fmtAgo', () => {
const now = new Date('2026-08-28T21:00:00Z')
it('describes recent and older readings', () => {
expect(fmtAgo('2026-08-28T20:56:00Z', now)).toBe('4m ago')
expect(fmtAgo('2026-08-28T20:59:55Z', now)).toBe('just now')
})
it('handles a missing or unparsable timestamp', () => {
expect(fmtAgo('', now)).toBe('unknown')
expect(fmtAgo('not a date', now)).toBe('unknown')
})
})

20
web/src/format.ts
View File

@ -42,3 +42,23 @@ export function daysUntil(isoDateStr: string): number {
const ms = target.getTime() - today.getTime()
return Math.ceil(ms / (1000 * 60 * 60 * 24))
}
/** Format a second count as "7h 27m" (or "44m" / "38s" when short). */
export function fmtDuration(seconds: number): string {
const total = Math.max(0, Math.round(seconds))
const h = Math.floor(total / 3600)
const m = Math.floor((total % 3600) / 60)
if (h > 0) return `${h}h ${m}m`
if (m > 0) return `${m}m`
return `${total}s`
}
/** Describe how long ago an ISO-8601 timestamp was, e.g. "4m ago". */
export function fmtAgo(iso: string, now: Date = new Date()): string {
if (!iso) return 'unknown'
const then = new Date(iso)
const seconds = (now.getTime() - then.getTime()) / 1000
if (Number.isNaN(seconds)) return 'unknown'
if (seconds < 10) return 'just now'
return `${fmtDuration(seconds)} ago`
}

110
web/src/index.css
View File

@ -533,3 +533,113 @@ td.proton {
border-radius: var(--radius-sm);
vertical-align: middle;
}
/* --- Tabs ---------------------------------------------------------------
Same shape as .segmented/.seg, but page-level rather than inside a panel. */
.tabs {
display: flex;
gap: var(--sp-1);
padding: var(--sp-3) var(--sp-4) 0;
}
.tab {
border: none;
background: transparent;
color: var(--muted);
padding: var(--sp-2) var(--sp-3);
border-radius: var(--radius-sm) var(--radius-sm) 0 0;
cursor: pointer;
font: inherit;
font-size: 14px;
border-bottom: 2px solid transparent;
}
.tab.active {
color: var(--text);
border-bottom-color: var(--accent);
font-weight: 600;
}
/* --- Gaming budget ------------------------------------------------------ */
.budget {
padding: var(--sp-4);
display: flex;
flex-direction: column;
gap: var(--sp-4);
}
.budget-today,
.budget-live,
.budget-rules {
background: var(--panel);
border: 1px solid var(--border);
border-radius: var(--radius-md);
padding: var(--sp-3);
}
.budget-hero {
font-size: 40px;
line-height: 1.1;
margin: var(--sp-2) 0;
color: var(--muted);
}
.budget-hero strong {
color: var(--text);
font-weight: 700;
}
.budget-meter {
height: 10px;
background: var(--panel-2);
border-radius: var(--radius-sm);
overflow: hidden;
}
.budget-meter-fill {
height: 100%;
background: var(--accent);
}
.budget-meter--warn .budget-meter-fill {
background: var(--warn);
}
.budget-meter--blocked .budget-meter-fill {
background: var(--danger);
}
.budget-badge {
display: inline-block;
padding: var(--sp-1) var(--sp-2);
border-radius: var(--radius-sm);
font-weight: 600;
}
.budget-badge--blocked {
background: var(--danger);
color: var(--on-fill);
}
.budget-verdict {
font-weight: 600;
color: var(--muted);
}
.budget-verdict--engaged {
color: var(--accent-2);
}
.budget-verdict--paused {
color: var(--warn);
}
.budget-rules-list {
display: grid;
grid-template-columns: max-content 1fr;
gap: var(--sp-1) var(--sp-3);
margin: 0;
font-size: 14px;
}
.budget-rules-list dt {
color: var(--muted);
}
.budget-rules-list dd {
margin: 0;
}
.chart-svg .bar {
fill: var(--accent);
}
.chart-svg .bar.over {
fill: var(--danger);
}
.chart-svg .budget-line {
stroke: var(--muted);
stroke-width: 1;
stroke-dasharray: 4 4;
}

73
web/src/test/factories.ts
View File

@ -1,7 +1,17 @@
// Shared test factories. Lives under src/test/ which is excluded from the
// app build and from coverage.
import type { Filters, PaceVsHLTB, WebDataset, WebGame, WebStateInfo } from '../types'
import type {
BudgetRules,
BudgetSession,
BudgetSnapshot,
BudgetToday,
Filters,
PaceVsHLTB,
WebDataset,
WebGame,
WebStateInfo,
} from '../types'
export function makeGame(over: Partial<WebGame> = {}): WebGame {
return {
@ -91,3 +101,64 @@ export function makeDataset(
...over,
}
}
export function makeBudgetToday(over: Partial<BudgetToday> = {}): BudgetToday {
return {
gaming_day: '2026-08-28',
day_starts_at: '06:00 local',
seconds_used: 3600,
budget_seconds: 28800,
seconds_remaining: 25200,
fraction_used: 0.125,
blocked: false,
blocked_at: 0,
next_warning_seconds: 3600,
warned_seconds: [],
...over,
}
}
export function makeBudgetSession(over: Partial<BudgetSession> = {}): BudgetSession {
return {
available: true,
observed_at: new Date().toISOString(),
state: 'engaged',
reason: 'engaged',
causes: [],
idle_seconds: 2,
screen_held: false,
game_name: 'Hollow Knight',
qualifying_count: 3,
processes: [{ pid: 42, name: 'hollow_knight' }],
...over,
}
}
export function makeBudgetRules(over: Partial<BudgetRules> = {}): BudgetRules {
return {
budget_seconds: 28800,
enforcement: true,
counts_launchers: true,
engagement_gate: true,
idle_grace_seconds: 300,
require_game_focus: true,
warn_at: [3600, 1800, 600, 300],
demo: false,
masked_launchers: [],
...over,
}
}
export function makeBudget(over: Partial<BudgetSnapshot> = {}): BudgetSnapshot {
return {
ok: true,
readable: true,
state_status: 'ok',
error: null,
today: makeBudgetToday(),
session: makeBudgetSession(),
history: [{ day: '2026-08-28', seconds: 3600 }],
rules: makeBudgetRules(),
...over,
}
}

69
web/src/types.ts
View File

@ -96,3 +96,72 @@ export interface Filters {
// Target-date planner ('' = disabled), ISO yyyy-mm-dd.
targetDate: string
}
/** Why the budget state file could or could not be read. */
export type BudgetStateStatus = 'ok' | 'missing' | 'denied' | 'corrupt'
export interface BudgetToday {
gaming_day: string
day_starts_at: string
seconds_used: number
budget_seconds: number
seconds_remaining: number
/** 0..1, already clamped by the backend. */
fraction_used: number
blocked: boolean
/** Epoch seconds; 0 means the cutoff has not engaged. */
blocked_at: number
/** Seconds-remaining at which the next warning fires; null when none left. */
next_warning_seconds: number | null
warned_seconds: number[]
}
export interface BudgetProcess {
pid: number
name: string
}
export interface BudgetSession {
available: boolean
/** ISO-8601 timestamp the daemon recorded; may be minutes old. */
observed_at: string
state: string
reason: string
causes: string[]
idle_seconds: number | null
screen_held: boolean | null
game_name: string
qualifying_count: number
processes: BudgetProcess[]
}
export interface BudgetHistoryDay {
day: string
seconds: number
}
export interface BudgetRules {
budget_seconds: number
enforcement: boolean
counts_launchers: boolean
engagement_gate: boolean
idle_grace_seconds: number
require_game_focus: boolean
warn_at: number[]
demo: boolean
masked_launchers: string[]
}
export interface BudgetSnapshot {
ok: boolean
readable: boolean
state_status: BudgetStateStatus
/** Human sentence explaining a non-ok status; null when readable. */
error: string | null
today: BudgetToday | null
session: BudgetSession
history: BudgetHistoryDay[]
rules: BudgetRules
}
export type TabId = 'planner' | 'budget'

130
web/src/useBudgetPoll.test.ts Normal file
View File

@ -0,0 +1,130 @@
import { act, renderHook, waitFor } from '@testing-library/react'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { makeBudget } from './test/factories'
import { useBudgetPoll } from './useBudgetPoll'
function setHidden(hidden: boolean) {
Object.defineProperty(document, 'hidden', { value: hidden, configurable: true })
}
describe('useBudgetPoll', () => {
beforeEach(() => {
vi.restoreAllMocks()
setHidden(false)
})
afterEach(() => {
vi.unstubAllGlobals()
vi.useRealTimers()
// defineProperty is not undone by unstubAllGlobals.
setHidden(false)
})
it('loads a snapshot on mount', async () => {
const snapshot = makeBudget()
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue({ ok: true, json: async () => snapshot }),
)
const { result } = renderHook(() => useBudgetPoll(false))
await waitFor(() => expect(result.current.data).not.toBeNull())
expect(result.current.error).toBeNull()
expect(result.current.data?.today?.gaming_day).toBe('2026-08-28')
})
it('requests the demo run when asked', async () => {
const fetchMock = vi
.fn()
.mockResolvedValue({ ok: true, json: async () => makeBudget() })
vi.stubGlobal('fetch', fetchMock)
renderHook(() => useBudgetPoll(true))
await waitFor(() => expect(fetchMock).toHaveBeenCalledWith('/api/budget?demo=1'))
})
it('reports a non-ok response', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue({ ok: false, status: 500, statusText: 'Server Error' }),
)
const { result } = renderHook(() => useBudgetPoll(false))
await waitFor(() => expect(result.current.error).toMatch(/500/))
})
it('handles a non-Error rejection', async () => {
vi.stubGlobal('fetch', vi.fn().mockRejectedValue('network down'))
const { result } = renderHook(() => useBudgetPoll(false))
await waitFor(() => expect(result.current.error).toBe('network down'))
})
it('polls again on the interval', async () => {
vi.useFakeTimers()
const fetchMock = vi
.fn()
.mockResolvedValue({ ok: true, json: async () => makeBudget() })
vi.stubGlobal('fetch', fetchMock)
renderHook(() => useBudgetPoll(false, 1000))
expect(fetchMock).toHaveBeenCalledTimes(1)
await act(async () => {
vi.advanceTimersByTime(2000)
})
expect(fetchMock).toHaveBeenCalledTimes(3)
})
it('does not poll while the tab is hidden', async () => {
vi.useFakeTimers()
setHidden(true)
const fetchMock = vi
.fn()
.mockResolvedValue({ ok: true, json: async () => makeBudget() })
vi.stubGlobal('fetch', fetchMock)
renderHook(() => useBudgetPoll(false, 1000))
await act(async () => {
vi.advanceTimersByTime(5000)
})
expect(fetchMock).not.toHaveBeenCalled()
})
it('stops polling after unmount', async () => {
vi.useFakeTimers()
const fetchMock = vi
.fn()
.mockResolvedValue({ ok: true, json: async () => makeBudget() })
vi.stubGlobal('fetch', fetchMock)
const { unmount } = renderHook(() => useBudgetPoll(false, 1000))
expect(fetchMock).toHaveBeenCalledTimes(1)
unmount()
await act(async () => {
vi.advanceTimersByTime(5000)
})
expect(fetchMock).toHaveBeenCalledTimes(1)
})
it('drops a response that arrives after unmount', async () => {
let resolve: (value: unknown) => void = () => {}
const pending = new Promise((r) => {
resolve = r
})
vi.stubGlobal('fetch', vi.fn().mockReturnValue(pending))
const { unmount, result } = renderHook(() => useBudgetPoll(false))
unmount()
await act(async () => {
resolve({ ok: true, json: async () => makeBudget() })
await pending
})
expect(result.current.data).toBeNull()
})
it('drops a rejection that arrives after unmount', async () => {
let reject: (reason: unknown) => void = () => {}
const pending = new Promise((_r, rj) => {
reject = rj
})
vi.stubGlobal('fetch', vi.fn().mockReturnValue(pending))
const { unmount, result } = renderHook(() => useBudgetPoll(false))
unmount()
await act(async () => {
reject(new Error('too late'))
await pending.catch(() => undefined)
})
expect(result.current.error).toBeNull()
})
})

52
web/src/useBudgetPoll.ts Normal file
View File

@ -0,0 +1,52 @@
import { useEffect, useState } from 'react'
import { fetchBudget } from './api'
import type { BudgetSnapshot } from './types'
export interface BudgetPollState {
data: BudgetSnapshot | null
error: string | null
}
/** Poll interval. The daemon ticks every ~3s, so this is about as fresh as the data gets. */
export const POLL_MS = 5000
/**
* Poll `/api/budget` while mounted, pausing whenever the tab is hidden.
*
* The hook is only mounted by the budget tab, so switching away stops the
* requests entirely; the `document.hidden` check covers the other case, a tab
* left open in the background for hours.
*/
export function useBudgetPoll(demo: boolean, intervalMs: number = POLL_MS): BudgetPollState {
const [data, setData] = useState<BudgetSnapshot | null>(null)
const [error, setError] = useState<string | null>(null)
useEffect(() => {
let cancelled = false
const load = () => {
if (document.hidden) return
fetchBudget(demo)
.then((snapshot) => {
// A response can arrive after unmount; committing it would warn and,
// worse, resurrect stale data on the next mount.
if (cancelled) return
setData(snapshot)
setError(null)
})
.catch((e: unknown) => {
if (cancelled) return
setError(e instanceof Error ? e.message : String(e))
})
}
load()
const id = setInterval(load, intervalMs)
return () => {
cancelled = true
clearInterval(id)
}
}, [demo, intervalMs])
return { data, error }
}