tests: block real kills/Steam spawns (a stale demo test killed the live game); enforce --demo is budget-only

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

DOCS-web-control-api.md

@@ -12,7 +12,7 @@ mirror is `web/src/api/contract.ts`; change both together or neither.
 - Privileged work goes to the **root daemon** over `/run/steam-backlog-enforcer/ctl.sock`
   (see "Control socket"). The web server never gets root.
 - **Friction lives at the trust boundary, never only in the browser.** Typed
-  phrases, countdowns, the unblock cap and the 24h exception lock are checked
+  phrases, countdowns and the unblock cap are checked
   by the server (unprivileged ops) or the daemon (privileged ops). The test:
   if `curl` can do it without the UI, it is not gated.
 - **There is no way to stop or disable the enforcer** — not in the UI, not
@@ -181,11 +181,12 @@ with explicit params, a `Progress` reporter and a `Prompter`.
 | `gaming-unblock` | `force release playtime mounts` | daemon; if unreachable the UI shows the `sudo ./run.sh gaming-unblock` fallback |
 | `abandon-pick` | `abandon {game_name}` | – |
 | `pick-manual` | `lock in {game_name}` | – |
-| `add-exception` | `request exception for {game_name}` | existing 24h lock |
+| `add-exception` | `request exception for {game_name}` | active immediately (no cooldown since de76c65); reason logged |
 | `uninstall` | `uninstall {count} games` | – |
 | `reset` | `wipe all enforcer state` | automatic timestamped backup first |
 | backup restore | `restore backup {id}` | – |
 | `enforce` restart | – | rate limit: 1 per 10 min, state flushed first |
+| `enforce` demo (`demo: 1`) | – | desktop-user job, cancellable; runs the gaming budget only and never touches the daemon's Steam-binary mounts; fails (exit 1) under a total block |
 
 `CommandSpec.friction` carries the phrase template so the UI never
 hard-codes it.

TODO-web-control-ui.md

@@ -16,7 +16,8 @@ has to type `./run.sh <cmd>`. The CLI stays as backend/recovery path; bare
 - Privileged ops go through the root daemon's `ctl.sock` (SO_PEERCRED);
   the web server never gets root.
 - Friction enforced server/daemon-side (phrases, gaming-reset 300 s
-  heartbeat countdown, unblock cap, 24h exception lock). More friction, not
+  heartbeat countdown, unblock cap; exceptions are immediate since de76c65,
+  user decision 2026-10-10). More friction, not
   less.
 - must not: any way to stop/disable the enforcer (UI, CLI or socket).
   Restart only, rate-limited, after a state flush.
@@ -95,5 +96,35 @@ has to type `./run.sh <cmd>`. The CLI stays as backend/recovery path; bare
       unreachable for coverage)
 - [x] F: run.sh bare → UI (`scripts/open_ui.sh`), .desktop entry (install.sh
       `install_desktop_entry`) — opened live, WM class verified
+- [x] H screenshot pass DONE 2026-10-10 (vmbox guest `sbe-g`, shots
+      /tmp/sbe-h/shots/): all 24 commands driven through the UI. Fixed on the
+      way (uncommitted): unhide skips like hide without Steam; enforce demo
+      button was a self-link (enforce is a "screen") → runs `{demo: 1}`,
+      restart sends `{demo: 0}`, mock mirrors the real catalog; enforce spec
+      now `cancellable` (demo loops forever, had no Cancel); demo run crashed
+      consuming the root-only restart-gap marker → demo never touches it;
+      "unsupported" hint named the daemon for a server error. serve restart
+      verified under systemd-run (PID + page reload); setup verified with a
+      bad key only (real key not copied off the guest). Live: status UI ==
+      CLI; done stays guest-only (deviation from `done`, by user rule).
+      Second round: enforce dialog per preset (`enforcePreset.ts`: demo =
+      user job, cancellable, no root chip; restart = root, not cancellable;
+      mode field hidden); demo budget log → CONFIG_DIR (was root-only
+      /var/log, EACCES); `serve` exempt from the total block (refused →
+      web unit crash-looped for the whole block); gaming-reset first attempt
+      = daemon refused commit 34 ms before ready_at (host/guest clock skew +
+      instant retype) → commit unlocks 1 s after ready_at.
+- [x] user decision 2026-10-10: keep exceptions immediate; help text, DOCS
+      and mock catalog fixed.
+- [x] user confirmation → tests, coverage, pre-commit, commit, deploy
+      (`sudo ./install.sh` + daemon restart for the _ctl_gap fix)
+- [x] 2026-10-10 test leak closed: a stale `enforce --demo` test ran the real
+      demo loop on the host, read the live /proc, billed and SIGTERMed the
+      game being played and shut Steam down. New autouse guard
+      `tests/_no_real_effects.py`: os.kill/killpg only reach pytest and its
+      own spawns, real Popen refuses Steam/kill argv, an unpatched demo tick
+      fails at once, and any blocked call fails the test at teardown. Proven
+      in vmbox: the OLD test now fails in 1 s with the fake game alive; with
+      the tripwire removed the os.kill layer alone blocks the SIGTERM.
 - [ ] F2: live verification of the finished UI
 - [x] user confirmation → tests, coverage, pre-commit, commit (2026-10-10)

steam_backlog_enforcer/_cmd_playtime.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 import os
 from typing import TYPE_CHECKING
 
+from steam_backlog_enforcer._enforce_demo import run_demo
 from steam_backlog_enforcer._enforce_loop import do_enforce
 from steam_backlog_enforcer._gaming_reset import reset_today
 from steam_backlog_enforcer._playtime_block import (
@@ -53,10 +54,9 @@ def cmd_enforce(config: Config, state: State, args: list[str]) -> int:
         _echo(f"Usage: enforce [{_DEMO_FLAG}]")
         return 2
 
-    demo = _DEMO_FLAG in args
-    if demo:
-        _echo("DEMO MODE: gaming budget is 60 seconds, using a separate state file.")
-    do_enforce(config, state, demo=demo)
+    if _DEMO_FLAG in args:
+        return run_demo(config)
+    do_enforce(config, state)
     return 0
 
 

steam_backlog_enforcer/_command_catalog.py

@@ -81,8 +81,11 @@ CATALOG: Final[dict[str, _Entry]] = {
     "reset": _job("system"),
     "setup": _Entry("system", "screen", mutating=True),
     # The daemon screen. Its only action is a rate-limited restart through
-    # the daemon; the 60-second demo runs as a job (params: demo=1).
-    "enforce": _Entry("system", "screen", mutating=True, privileged=True),
+    # the daemon; the 60-second demo runs as a job (params: demo=1) that
+    # loops until stopped, so the spec must say it can be cancelled.
+    "enforce": _Entry(
+        "system", "screen", mutating=True, privileged=True, cancellable=True
+    ),
     "serve": _Entry("system", "screen"),
 }
 

steam_backlog_enforcer/_command_help.py

@@ -35,7 +35,7 @@ COMMAND_DESCRIPTIONS: Final[dict[str, str]] = {
 
 # Commands with non-standard arg handling (shown in help but not in COMMANDS).
 EXTRA_COMMAND_DESCRIPTIONS: Final[dict[str, str]] = {
-    "add-exception": "Request 24h-locked whitelist exception (use --reason)",
+    "add-exception": "Whitelist a game immediately (phrase + --reason)",
     "unblock": "Unblock the store for [minutes] (default 15, max 30)",
     "serve": "Start the web UI (--port N; replaces a stale server)",
     "pick-manual": f"Pick a game by app_id, lock enforcer for {MANUAL_LOCK_DAYS} days",

steam_backlog_enforcer/_command_locks.py

@@ -40,6 +40,9 @@ _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset(
 # gaming-unblock is included because a playtime bind mount makes the total
 # block's own `pacman -R steam` fail EBUSY - it must stay reachable exactly
 # when the two collide. gaming-reset is NOT included: it shortens enforcement.
+# serve changes nothing itself (every job it starts is lock-checked on its own)
+# and refusing it took the web UI down for the whole block, crash-looping its
+# systemd unit, the first time anything restarted it.
 _TOTAL_BLOCK_EXEMPT_COMMANDS = frozenset(
-    {"status", "enforce", "gaming-status", "gaming-unblock"}
+    {"status", "enforce", "gaming-status", "gaming-unblock", "serve"}
 )

steam_backlog_enforcer/_ctl_gap.py

@@ -49,6 +49,14 @@ def record_exit() -> None:
         logger.warning("Cannot record the restart exit time; the gap goes unbilled")
 
 
+def _drop_marker() -> None:
+    """Delete the marker; a caller without root rights just leaves it."""
+    try:
+        RESTART_EXIT_FILE.unlink(missing_ok=True)
+    except OSError:
+        logger.debug("Cannot delete the restart exit marker", exc_info=True)
+
+
 def _take_exit_time() -> float | None:
     """Read and delete the marker; ``None`` if absent or unusable."""
     try:
@@ -57,7 +65,7 @@ def _take_exit_time() -> float | None:
     except OSError, ValueError, KeyError, TypeError:
         return None
     finally:
-        RESTART_EXIT_FILE.unlink(missing_ok=True)
+        _drop_marker()
     return float(exited) if isinstance(exited, int | float) else None
 
 
@@ -76,8 +84,12 @@ def settle_restart_gap(
         base_interval: The normal tick interval (lower bound for the widened one).
         demo: Whether this is a demo run (never settles).
     """
+    # A demo run (often the desktop user, via the web UI) must not touch the
+    # daemon's root-only marker at all, let alone consume it.
+    if demo:
+        return
     exited = _take_exit_time()
-    if demo or exited is None:
+    if exited is None:
         return
     gap = time.time() - exited
     if not 0 < gap <= MAX_GAP_SECONDS:

steam_backlog_enforcer/_enforce_demo.py

@@ -0,0 +1,76 @@
+"""``enforce --demo``: the gaming budget alone, on a 60-second budget.
+
+The demo shows the budget's escalation -- warnings, Steam shutdown, killing
+the games -- without waiting hours. It runs nothing else of the enforce loop:
+the store block, install guard, unassigned-game kills, total-block upkeep and
+``chattr`` are the daemon's, and a copy run as the desktop user (the web UI)
+could only fail at them, silently. It never touches the real Steam-binary
+mounts either (:func:`._playtime_cutoff.reconcile_for`).
+
+Whatever it cannot do, it refuses out loud: under a total gaming block Steam
+is uninstalled and the budget is not counted, so there is nothing to show.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING
+
+from steam_backlog_enforcer._echo import _echo
+from steam_backlog_enforcer._enforce_loop import ENFORCE_INTERVAL
+from steam_backlog_enforcer._playtime import playtime_tick
+from steam_backlog_enforcer._playtime_session import new_session
+from steam_backlog_enforcer._total_block import (
+    get_total_block_status,
+    is_total_block_active,
+)
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.config import Config
+
+EXIT_REFUSED = 1
+
+
+def _refuse_under_total_block(when: str) -> int:
+    """Say why a demo cannot run while a total block is on.
+
+    Args:
+        when: How the block was found ("cannot start" / "stopped").
+
+    Returns:
+        The exit code to leave with.
+    """
+    days = get_total_block_status().days_remaining
+    _echo(
+        f"Demo {when}: total gaming block active ({days:.1f} day(s) left). "
+        "Steam is uninstalled and the budget is not counted, so there is "
+        "nothing for a demo budget to run against."
+    )
+    return EXIT_REFUSED
+
+
+def run_demo(config: Config) -> int:
+    """Run the demo budget until stopped (Ctrl+C or a cancelled job).
+
+    Args:
+        config: Enforcer configuration.
+
+    Returns:
+        Process exit code: 0 when stopped, 1 when a total block refused it.
+    """
+    if is_total_block_active():
+        return _refuse_under_total_block("cannot start")
+    _echo("DEMO MODE: gaming budget is 60 seconds, using a separate state file.")
+    _echo("  Runs the budget only: warnings, Steam shutdown, game kills.")
+    _echo("  The store, installs and Steam's binaries stay with the daemon.")
+    _echo(f"  Ticking every {ENFORCE_INTERVAL}s until stopped.\n")
+    session = new_session(demo=True)
+    try:
+        while True:
+            if is_total_block_active():
+                return _refuse_under_total_block("stopped")
+            playtime_tick(config, interval=ENFORCE_INTERVAL, session=session, demo=True)
+            time.sleep(ENFORCE_INTERVAL)
+    except KeyboardInterrupt:
+        _echo("\nDemo stopped.")
+    return 0

steam_backlog_enforcer/_playtime.py

@@ -34,10 +34,7 @@ import time
 from typing import TYPE_CHECKING
 
 from steam_backlog_enforcer._gaming_days import gaming_day_key
-from steam_backlog_enforcer._playtime_block import (
-    mounted_targets,
-    reconcile,
-)
+from steam_backlog_enforcer._playtime_block import mounted_targets
 from steam_backlog_enforcer._playtime_budget import (
     accumulate,
     roll_over,
@@ -46,6 +43,7 @@ from steam_backlog_enforcer._playtime_cutoff import (
     _begin_cutoff,
     _sustain_block,
     _warn,
+    reconcile_for,
 )
 from steam_backlog_enforcer._playtime_procs import attributed_key, qualifying_pids
 from steam_backlog_enforcer._playtime_state import (
@@ -105,7 +103,7 @@ def playtime_tick(
         # The total block runs `pacman -R steam` every tick; our bind mounts
         # would make that fail EBUSY. It is strictly stronger — get out of its
         # way, and stop accruing against a budget nobody can spend.
-        reconcile(should_block=False)
+        reconcile_for(rules, should_block=False)
         save_state(state, demo=demo)
         session.history.observe(state, demo=demo)
         return
@@ -212,11 +210,11 @@ def _policy(
     if not rules.enforcement:
         # A kill switch must still release: leaving live mounts behind would
         # make "disabled" mean "permanently blocked".
-        reconcile(should_block=False)
+        reconcile_for(rules, should_block=False)
         return state
 
     if state.seconds < rules.budget_seconds:
-        reconcile(should_block=False)
+        reconcile_for(rules, should_block=False)
         if state.is_blocked():
             state = _release_after_raise(state, rules)
         # A raise that never reached the cutoff lifts remaining back above

steam_backlog_enforcer/_playtime_cutoff.py

@@ -35,6 +35,22 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def reconcile_for(rules: PlaytimeRules, *, should_block: bool) -> None:
+    """Drive the real Steam-binary mounts -- never from a demo.
+
+    The mounts are the daemon's. A demo run as root used to release a real,
+    spent-budget block on every under-budget tick (the daemon re-applied it 3 s
+    later); run as the desktop user (the web UI) it failed without a word. So a
+    demo's policy runs in full except this step, which _begin_cutoff announces.
+
+    Args:
+        rules: Policy for this tick (``rules.demo`` marks a demo run).
+        should_block: Whether the launchers should currently be masked.
+    """
+    if not rules.demo:
+        reconcile(should_block=should_block)
+
+
 def _warn(state: PlaytimeState, rules: PlaytimeRules) -> PlaytimeState:
     """Fire the due warning, if any, and record that it fired.
 
@@ -83,6 +99,12 @@ def _begin_cutoff(
     )
     request_steam_shutdown()
     kill_gaming_processes(_kill_set(rules), force=False)
+    if rules.demo:
+        logger.warning(
+            "Demo: in %.0f s the real block would mask Steam's binaries; only the "
+            "daemon does that. This demo keeps killing games until you stop it.",
+            _SHUTDOWN_GRACE_SECONDS,
+        )
     return replace(state, blocked_at=now.timestamp())
 
 
@@ -104,7 +126,7 @@ def _sustain_block(
     """
     elapsed = now.timestamp() - state.blocked_at
     if elapsed >= _SHUTDOWN_GRACE_SECONDS:
-        reconcile(should_block=True)
+        reconcile_for(rules, should_block=True)
     kill_gaming_processes(_kill_set(rules), force=elapsed >= rules.sigkill_after)
     return state
 

steam_backlog_enforcer/_playtime_log.py

@@ -29,13 +29,17 @@ import logging.handlers
 from pathlib import Path
 from typing import TYPE_CHECKING, Final, cast
 
+from steam_backlog_enforcer.config import CONFIG_DIR
+
 if TYPE_CHECKING:
     from steam_backlog_enforcer._playtime_state import PlaytimeRules, PlaytimeState
 
 logger = logging.getLogger(__name__)
 
 BUDGET_LOG_FILE: Final = Path("/var/log/steam-backlog-enforcer/budget.jsonl")
-BUDGET_DEMO_LOG_FILE: Final = Path("/var/log/steam-backlog-enforcer/budget-demo.jsonl")
+# Beside the demo state file, not in root-only /var/log: the web UI runs the
+# demo as the desktop user, and a demo record carries no enforcement weight.
+BUDGET_DEMO_LOG_FILE: Final = CONFIG_DIR / "budget-demo.jsonl"
 
 _MAX_BYTES: Final = 5 * 1024 * 1024
 _BACKUP_COUNT: Final = 5

steam_backlog_enforcer/jobs/_registry.py

@@ -102,7 +102,11 @@ def _enforce_demo(
     if not params.get("demo"):
         command = "enforce"
         raise PrivilegedViaDaemonError(command)
-    return {"exit_code": cmd_enforce(Config.load(), State.load(), ["--demo"])}
+    code = cmd_enforce(Config.load(), State.load(), ["--demo"])
+    if code:
+        # A refused demo is a failed job, not a quiet "Succeeded".
+        raise SystemExit(code)
+    return {"exit_code": code}
 
 
 def _restore_backup(

steam_backlog_enforcer/main/install.py

@@ -8,6 +8,7 @@ from steam_backlog_enforcer._actions import allowed_app_ids, allowed_games
 from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
 from steam_backlog_enforcer._progress import current_progress
 from steam_backlog_enforcer._prompter import confirm_phrase
+from steam_backlog_enforcer._steam_errors import SteamUnavailableError
 from steam_backlog_enforcer._steam_state import is_game_fully_installed
 from steam_backlog_enforcer.game_install import (
     _echo,
@@ -127,7 +128,13 @@ def cmd_unhide(config: Config, _state: State) -> None:
         return
 
     _echo("Unhiding all games...")
-    count = unhide_all_games(owned_ids)
+    # Same contract as cmd_hide: no reachable Steam means nothing to change
+    # this time, not a failed command.
+    try:
+        count = unhide_all_games(owned_ids)
+    except SteamUnavailableError as exc:
+        _echo(f"Library unhiding: skipped ({exc})")
+        return
     _echo(f"Unhidden {count} games.")
 
     if count > 0:

steam_backlog_enforcer/tests/_no_real_effects.py

@@ -0,0 +1,199 @@
+"""Autouse backstop: no test may signal a foreign process or drive Steam.
+
+The module-level ``subprocess`` patches in ``_no_subprocess`` only cover
+the call sites someone remembered to list. On 2026-10-10 a stale test let
+the real ``enforce --demo`` loop run: it read the live process table, billed
+the game the developer was playing and SIGTERMed it, then shut Steam down.
+Nothing stopped it, because nothing guarded ``os.kill``.
+
+This guard sits under every call site instead of beside them:
+
+* ``os.kill`` / ``os.killpg`` reach only pytest itself and processes it
+  spawned (by ppid walk over the *real* ``/proc``, plus every pid the
+  ``Popen`` wrapper let through, which survives a reparent to a subreaper).
+  Signal 0 is a probe and always passes; pids <= 0 never do.
+* ``subprocess.Popen`` refuses any argv that names a Steam/kill tool.
+
+A blocked call raises :class:`RealEffectError` (a ``BaseException``, so an
+``except Exception`` in the code under test cannot swallow it) *and* is
+recorded, and the test fails at teardown even if something caught it.
+Tests that patch ``os.kill`` or a module's ``subprocess`` themselves
+override this layer for their duration, as before.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path, PurePath
+import shlex
+import subprocess
+from typing import TYPE_CHECKING, Any
+from unittest.mock import patch
+
+import pytest
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator, Sequence
+
+    _Args = str | bytes | os.PathLike[str] | Sequence[str | bytes | os.PathLike[str]]
+
+# Hardcoded on purpose: the package's own _PROC constants are redirected to
+# empty tmp dirs by other fixtures, which would make every child look foreign.
+_REAL_PROC = Path("/proc")
+# Captured at import: _no_subprocess patches ``subprocess.Popen`` on the shared
+# module for every test, so the name alone would resolve to its MagicMock.
+# Guarding the class itself also covers code that bound it before that.
+_REAL_POPEN = subprocess.Popen
+_MAX_ANCESTRY = 64
+_DENIED_TOOLS = frozenset(
+    {
+        "steam",
+        "steam.sh",
+        "steam-game-installer",
+        "xdg-open",
+        "pkill",
+        "killall",
+        "kill",
+    }
+)
+
+
+class RealEffectError(BaseException):
+    """A test reached a real process signal or a real Steam command."""
+
+
+def _parent_of(pid: int) -> int | None:
+    """Return *pid*'s parent from the real ``/proc``, or None if unreadable."""
+    try:
+        raw = (_REAL_PROC / str(pid) / "stat").read_text(encoding="utf-8")
+    except OSError:
+        return None
+    # comm may contain spaces and parens; the state and ppid follow the last ")".
+    fields = raw.rsplit(")", 1)[-1].split()
+    return int(fields[1]) if len(fields) > 1 else None
+
+
+def _is_ours(pid: int, spawned: set[int]) -> bool:
+    """Tell whether *pid* is pytest itself or one of its descendants."""
+    if pid <= 0:
+        return False
+    owners = spawned | {os.getpid()}
+    current: int | None = pid
+    for _ in range(_MAX_ANCESTRY):
+        if current is None or current <= 1:
+            return False
+        if current in owners:
+            return True
+        current = _parent_of(current)
+    return False
+
+
+def _argv(args: _Args, *, via_shell: bool) -> list[str]:
+    """Normalise Popen's *args* into a flat list of strings."""
+    if isinstance(args, (str, bytes, os.PathLike)):
+        text = os.fsdecode(args)
+        return shlex.split(text) if via_shell else [text]
+    return [os.fsdecode(arg) for arg in args]
+
+
+def _drives_steam(argv: Sequence[str]) -> bool:
+    """Tell whether *argv* would launch, stop or signal Steam or a game."""
+    for arg in argv:
+        if PurePath(arg).name in _DENIED_TOOLS:
+            return True
+        if arg.startswith("steam://") or arg == "-shutdown":
+            return True
+    return False
+
+
+def _fail_if_blocked(blocked: Sequence[str]) -> None:
+    """Fail the test if anything was blocked, even if the code caught it."""
+    if blocked:
+        pytest.fail("real side effect(s) blocked: " + "; ".join(blocked))
+
+
+@pytest.fixture(autouse=True)
+def _no_real_effects() -> Iterator[list[str]]:
+    """Block foreign signals and Steam-driving spawns; fail the test if hit.
+
+    Yields the list of blocked calls, so a test of the guard itself can
+    inspect and clear it before teardown.
+    """
+    blocked: list[str] = []
+    spawned: set[int] = set()
+    real_kill, real_killpg = os.kill, os.killpg
+    real_popen_init: Callable[..., None] = _REAL_POPEN.__init__
+
+    def _refuse(what: str) -> RealEffectError:
+        blocked.append(what)
+        return RealEffectError(f"test tried a real effect: {what}")
+
+    def guarded_kill(pid: int, sig: int) -> None:
+        if sig != 0 and not _is_ours(pid, spawned):
+            msg = f"os.kill(pid={pid}, sig={sig})"
+            raise _refuse(msg)
+        real_kill(pid, sig)
+
+    def guarded_killpg(pgid: int, sig: int) -> None:
+        if sig != 0 and not _is_ours(pgid, spawned):
+            msg = f"os.killpg(pgid={pgid}, sig={sig})"
+            raise _refuse(msg)
+        real_killpg(pgid, sig)
+
+    def guarded_popen_init(
+        self: subprocess.Popen[Any], args: _Args, *a: object, **kw: object
+    ) -> None:
+        argv = _argv(args, via_shell=bool(kw.get("shell")))
+        exe = kw.get("executable")
+        if isinstance(exe, (str, bytes, os.PathLike)):
+            argv.append(os.fsdecode(exe))
+        if _drives_steam(argv):
+            # Popen.__del__ reads this; without it a refused spawn warns.
+            vars(self)["_child_created"] = False
+            msg = f"subprocess.Popen({argv!r})"
+            raise _refuse(msg)
+        real_popen_init(self, args, *a, **kw)
+        spawned.add(self.pid)
+
+    def tripwire(*_args: object, **_kwargs: object) -> None:
+        msg = "unpatched enforce --demo loop (playtime_tick)"
+        raise _refuse(msg)
+
+    with (
+        patch.object(os, "kill", guarded_kill),
+        patch.object(os, "killpg", guarded_killpg),
+        patch.object(_REAL_POPEN, "__init__", guarded_popen_init),
+        # The demo loop never ends on its own; a test reaching it unpatched
+        # must fail at once rather than hang (test_enforce_demo patches it).
+        patch("steam_backlog_enforcer._enforce_demo.playtime_tick", tripwire),
+    ):
+        yield blocked
+    _fail_if_blocked(blocked)
+
+
+# ──────────────────────────────────────────────────────────────
+# Live process-table isolation
+#
+# Steam games are found by scanning the real /proc for SteamAppId
+# (enforcer.get_running_steam_game_pids hardcodes the path). Unpatched,
+# that reads the *developer's* process table: running the suite while
+# playing made three restart tests fail on 2026-08-28, and let a demo
+# loop bill and kill the live game on 2026-10-10. Tests that care
+# patch these names themselves.
+# ──────────────────────────────────────────────────────────────
+
+
+@pytest.fixture(autouse=True)
+def _no_live_games() -> Iterator[None]:
+    """Report no running games unless a test says otherwise."""
+    with (
+        patch(
+            "steam_backlog_enforcer._steam_restart_guard.get_running_steam_game_pids",
+            return_value={},
+        ),
+        patch(
+            "steam_backlog_enforcer._playtime_procs.get_running_steam_game_pids",
+            return_value={},
+        ),
+    ):
+        yield

steam_backlog_enforcer/tests/conftest.py

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

steam_backlog_enforcer/tests/test_cmd_playtime.py

@@ -29,19 +29,27 @@ def _echoed(mock_echo: object) -> str:
 
 class TestCmdEnforce:
     def test_runs_in_production_by_default(self) -> None:
-        with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo"):
+        with (
+            patch(f"{PKG}.do_enforce") as mock_run,
+            patch(f"{PKG}.run_demo") as mock_demo,
+            patch(f"{PKG}._echo"),
+        ):
             assert cmd_enforce(Config(), State(), []) == 0
-        assert mock_run.call_args.kwargs["demo"] is False
+        mock_run.assert_called_once()
+        mock_demo.assert_not_called()
 
-    def test_demo_flag_enables_demo(self) -> None:
-        with patch(f"{PKG}.do_enforce") as mock_run, patch(f"{PKG}._echo"):
+    def test_demo_flag_runs_the_budget_only_demo(self) -> None:
+        with (
+            patch(f"{PKG}.do_enforce") as mock_run,
+            patch(f"{PKG}.run_demo", return_value=0) as mock_demo,
+        ):
             assert cmd_enforce(Config(), State(), ["--demo"]) == 0
-        assert mock_run.call_args.kwargs["demo"] is True
+        mock_demo.assert_called_once()
+        mock_run.assert_not_called()
 
-    def test_demo_announces_itself(self) -> None:
-        with patch(f"{PKG}.do_enforce"), patch(f"{PKG}._echo") as mock_echo:
-            cmd_enforce(Config(), State(), ["--demo"])
-        assert "DEMO MODE" in _echoed(mock_echo)
+    def test_demo_exit_code_is_passed_on(self) -> None:
+        with patch(f"{PKG}.run_demo", return_value=1):
+            assert cmd_enforce(Config(), State(), ["--demo"]) == 1
 
     def test_unknown_flag_exits_nonzero_without_running(self) -> None:
         """A mistyped --Demo must not silently run the real 8-hour budget."""

steam_backlog_enforcer/tests/test_command_catalog.py

@@ -46,6 +46,17 @@ class TestSpec:
         assert spec["friction"] is None
         assert spec["locked_reason"] is None
 
+    def test_enforce_screen_is_cancellable(self) -> None:
+        # Its demo is a job that loops until stopped; without the flag the UI
+        # offered no Cancel for it.
+        entry = _command_catalog.CATALOG["enforce"]
+        spec = _command_catalog._spec("enforce", entry, "d", None)
+        assert (spec["kind"], spec["cancellable"], spec["privileged"]) == (
+            "screen",
+            True,
+            True,
+        )
+
     def test_job_takes_flags_from_registry(self) -> None:
         flags = JobFlags(mutating=False, cancellable=True, privileged=True)
         with patch.dict(_command_catalog.JOB_FLAGS, {"scan": flags}):

steam_backlog_enforcer/tests/test_ctl_restart.py

@@ -8,8 +8,8 @@ attribute so the redirect applies.
 from __future__ import annotations
 
 import json
+from pathlib import Path
 import time
-from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -18,9 +18,6 @@ from steam_backlog_enforcer import _ctl_gap, _ctl_restart
 from steam_backlog_enforcer._ctl_protocol import CtlError
 from steam_backlog_enforcer.tests._ctl_fixtures import make_ctx
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
 
 def _write_last(value: object) -> None:
     path = _ctl_restart.RESTART_STATE_FILE
@@ -153,7 +150,14 @@ class TestRestartGap:
     def test_no_marker_bills_nothing(self) -> None:
         self._settle().assert_not_called()
 
-    def test_demo_never_bills_but_spends_marker(self) -> None:
+    def test_demo_never_touches_the_marker(self) -> None:
+        # The marker is the daemon's (root-only); a demo -- often the desktop
+        # user's web job -- must neither crash on it nor spend it.
         path = self._marker(json.dumps({"exited_at": time.time() - 20}))
         self._settle(demo=True).assert_not_called()
-        assert not path.exists()
+        assert path.exists()
+
+    def test_undeletable_marker_is_still_read(self) -> None:
+        self._marker(json.dumps({"exited_at": 123.0}))
+        with patch.object(Path, "unlink", side_effect=PermissionError("root")):
+            assert _ctl_gap._take_exit_time() == 123.0

steam_backlog_enforcer/tests/test_enforce_demo.py

@@ -0,0 +1,67 @@
+"""Tests for the budget-only ``enforce --demo`` loop."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from steam_backlog_enforcer._enforce_demo import EXIT_REFUSED, run_demo
+from steam_backlog_enforcer.config import Config
+
+PKG = "steam_backlog_enforcer._enforce_demo"
+
+
+def _echoed(mock_echo: MagicMock) -> str:
+    return " ".join(str(c) for c in mock_echo.call_args_list)
+
+
+def _block(days: float = 2.0) -> SimpleNamespace:
+    return SimpleNamespace(days_remaining=days)
+
+
+class TestRunDemo:
+    def test_refuses_to_start_under_a_total_block(self) -> None:
+        with (
+            patch(f"{PKG}.is_total_block_active", return_value=True),
+            patch(f"{PKG}.get_total_block_status", return_value=_block(2.0)),
+            patch(f"{PKG}.playtime_tick") as tick,
+            patch(f"{PKG}._echo") as echo,
+        ):
+            assert run_demo(Config()) == EXIT_REFUSED
+        tick.assert_not_called()
+        assert "Demo cannot start: total gaming block active (2.0 day(s) left)" in (
+            _echoed(echo)
+        )
+
+    def test_ticks_only_the_demo_budget_until_interrupted(self) -> None:
+        with (
+            patch(f"{PKG}.is_total_block_active", return_value=False),
+            patch(f"{PKG}.new_session", return_value="session") as session,
+            patch(f"{PKG}.playtime_tick") as tick,
+            patch(f"{PKG}.time.sleep", side_effect=[None, KeyboardInterrupt]),
+            patch(f"{PKG}._echo") as echo,
+        ):
+            assert run_demo(Config()) == 0
+        session.assert_called_once_with(demo=True)
+        assert tick.call_count == 2
+        assert all(c.kwargs["demo"] is True for c in tick.call_args_list)
+        assert tick.call_args.kwargs["session"] == "session"
+        out = _echoed(echo)
+        assert "DEMO MODE" in out
+        assert "Runs the budget only" in out
+        assert "Demo stopped." in out
+
+    def test_stops_loudly_when_a_total_block_starts_mid_run(self) -> None:
+        with (
+            patch(f"{PKG}.is_total_block_active", side_effect=[False, False, True]),
+            patch(f"{PKG}.get_total_block_status", return_value=_block(1.0)),
+            patch(f"{PKG}.new_session"),
+            patch(f"{PKG}.playtime_tick") as tick,
+            patch(f"{PKG}.time.sleep"),
+            patch(f"{PKG}._echo") as echo,
+        ):
+            assert run_demo(Config()) == EXIT_REFUSED
+        tick.assert_called_once()
+        assert "Demo stopped: total gaming block active (1.0 day(s) left)" in (
+            _echoed(echo)
+        )

steam_backlog_enforcer/tests/test_jobs_registry.py

@@ -86,6 +86,15 @@ class TestEnforceDemo:
         assert command.call_args.args[2] == ["--demo"]
         assert data == {"exit_code": 0}
 
+    def test_refused_demo_fails_the_job(self) -> None:
+        """A non-zero demo exit is a failed job, never a quiet success."""
+        with (
+            patch(_PREFIX + "cmd_enforce", return_value=1),
+            pytest.raises(SystemExit) as exc,
+        ):
+            _registry._enforce_demo({"demo": 1}, _PROGRESS, _PROMPTER)
+        assert exc.value.code == 1
+
     def test_real_enforce_goes_to_daemon(self) -> None:
         """Without ``demo`` the job refuses."""
         with pytest.raises(PrivilegedViaDaemonError):

steam_backlog_enforcer/tests/test_main_install.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from unittest.mock import patch
 
+from steam_backlog_enforcer._steam_errors import SteamUnavailableError
 from steam_backlog_enforcer.config import Config, State
 from steam_backlog_enforcer.main import (
     cmd_hide,
@@ -181,6 +182,18 @@ class TestCmdUnhide:
         ):
             cmd_unhide(Config(), State())
 
+    def test_skips_without_steam_like_hide(self) -> None:
+        with (
+            patch(f"{PKG}.get_all_owned_app_ids", return_value=[1]),
+            patch(
+                f"{PKG}.unhide_all_games",
+                side_effect=SteamUnavailableError("Steam is not installed"),
+            ),
+            patch(f"{PKG}._echo") as echo,
+        ):
+            cmd_unhide(Config(), State())
+        echo.assert_called_with("Library unhiding: skipped (Steam is not installed)")
+
     def test_unhides_zero(self) -> None:
         with (
             patch(f"{PKG}.get_all_owned_app_ids", return_value=[1]),

steam_backlog_enforcer/tests/test_main_shared.py

@@ -200,10 +200,12 @@ class TestEnforceTotalBlockLock:
         gaming-unblock is present because a playtime bind mount makes the
         total block's own `pacman -R steam` fail EBUSY - it has to stay
         reachable exactly when the two collide. gaming-reset is deliberately
-        absent: it would shorten enforcement.
+        absent: it would shorten enforcement. serve changes nothing itself
+        (each job it starts is lock-checked) and refusing it took the web UI
+        down for the whole block.
         """
         assert (
-            frozenset({"status", "enforce", "gaming-status", "gaming-unblock"})
+            frozenset({"status", "enforce", "gaming-status", "gaming-unblock", "serve"})
             == _TOTAL_BLOCK_EXEMPT_COMMANDS
         )
         assert "gaming-reset" not in _TOTAL_BLOCK_EXEMPT_COMMANDS

steam_backlog_enforcer/tests/test_no_real_effects.py

@@ -0,0 +1,191 @@
+"""Tests for the autouse real-effect guard in ``_no_real_effects``.
+
+Every call that the guard blocks is cleared from its record before the
+test ends; otherwise the guard's own teardown would (correctly) fail it.
+
+``subprocess.Popen`` / ``subprocess.run`` are MagicMocks during tests (see
+``_no_subprocess``), so real spawns here go through the captured class.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer._enforce_demo import run_demo
+from steam_backlog_enforcer.config import Config
+from steam_backlog_enforcer.tests import _no_real_effects as guard
+from steam_backlog_enforcer.tests._no_real_effects import (
+    _REAL_POPEN,
+    RealEffectError,
+    _argv,
+    _drives_steam,
+    _fail_if_blocked,
+    _is_ours,
+    _parent_of,
+)
+
+# Never a live pid: if the guard ever let it through, the kernel would only
+# answer ProcessLookupError (unlike pid 1, which a root run would SIGTERM).
+_NO_SUCH_PID = int(Path("/proc/sys/kernel/pid_max").read_text(encoding="utf-8"))
+
+
+@pytest.fixture
+def blocked(request: pytest.FixtureRequest) -> list[str]:
+    """The autouse guard's record of blocked calls, for inspection."""
+    record: list[str] = request.getfixturevalue("_no_real_effects")
+    return record
+
+
+@pytest.fixture
+def child() -> object:
+    """A real, short-lived child of pytest in its own session."""
+    proc = _REAL_POPEN(["sleep", "30"], start_new_session=True)
+    yield proc
+    proc.kill()
+    proc.wait()
+
+
+class TestDrivesSteam:
+    @pytest.mark.parametrize(
+        "argv",
+        [
+            ["steam"],
+            ["/usr/bin/steam", "-silent"],
+            ["sudo", "-u", "kuhy", "steam.sh"],
+            ["xdg-open", "https://example.com"],
+            ["runuser", "steam-game-installer", "620"],
+            ["pkill", "-f", "x"],
+            ["killall", "x"],
+            ["/bin/kill", "-TERM", "42"],
+            ["foo", "steam://rungameid/620"],
+            ["foo", "-shutdown"],
+        ],
+    )
+    def test_steam_and_kill_tools_are_denied(self, argv: list[str]) -> None:
+        assert _drives_steam(argv)
+
+    def test_ordinary_commands_pass(self) -> None:
+        assert not _drives_steam(["git", "status", "--steam"])
+
+
+class TestArgv:
+    def test_string_without_shell_is_one_program(self) -> None:
+        assert _argv("steam -shutdown", via_shell=False) == ["steam -shutdown"]
+
+    def test_string_with_shell_is_split(self) -> None:
+        assert _argv("steam -shutdown", via_shell=True) == ["steam", "-shutdown"]
+
+    def test_bytes_and_paths_are_decoded(self) -> None:
+        assert _argv([b"a", Path("/b"), "c"], via_shell=False) == ["a", "/b", "c"]
+        assert _argv(Path("/usr/bin/steam"), via_shell=False) == ["/usr/bin/steam"]
+
+
+class TestIsOurs:
+    def test_pytest_itself(self) -> None:
+        assert _is_ours(os.getpid(), set())
+
+    @pytest.mark.parametrize("pid", [0, -1])
+    def test_group_and_broadcast_pids_never(self, pid: int) -> None:
+        assert not _is_ours(pid, set())
+
+    def test_init_and_missing_pids_never(self) -> None:
+        assert not _is_ours(1, set())
+        assert not _is_ours(_NO_SUCH_PID, set())
+
+    def test_descendants(self, child: subprocess.Popen[bytes]) -> None:
+        assert _is_ours(child.pid, set())
+
+    def test_reparented_spawns_still_count(self) -> None:
+        with patch.object(guard, "_parent_of", return_value=1):
+            assert _is_ours(4242, {4242})
+            assert not _is_ours(4242, set())
+
+    def test_a_cyclic_table_gives_up(self) -> None:
+        with patch.object(guard, "_parent_of", return_value=4243):
+            assert not _is_ours(4242, set())
+
+
+class TestParentOf:
+    def test_reads_ppid_after_last_paren(self, tmp_path: Path) -> None:
+        (tmp_path / "7").mkdir()
+        (tmp_path / "7" / "stat").write_text("7 (a) b) S 3 7 7 0", encoding="utf-8")
+        with patch.object(guard, "_REAL_PROC", tmp_path):
+            assert _parent_of(7) == 3
+
+    def test_truncated_stat(self, tmp_path: Path) -> None:
+        (tmp_path / "7").mkdir()
+        (tmp_path / "7" / "stat").write_text("7 (a) S", encoding="utf-8")
+        with patch.object(guard, "_REAL_PROC", tmp_path):
+            assert _parent_of(7) is None
+
+
+class TestGuardedKill:
+    def test_foreign_signal_is_refused_and_recorded(self, blocked: list[str]) -> None:
+        with pytest.raises(RealEffectError):
+            os.kill(_NO_SUCH_PID, signal.SIGTERM)
+        assert blocked == [f"os.kill(pid={_NO_SUCH_PID}, sig={signal.SIGTERM})"]
+        blocked.clear()
+
+    def test_broadcast_is_refused(self, blocked: list[str]) -> None:
+        with pytest.raises(RealEffectError):
+            os.kill(-1, signal.SIGTERM)
+        with pytest.raises(RealEffectError):
+            os.killpg(0, signal.SIGTERM)
+        assert len(blocked) == 2
+        blocked.clear()
+
+    def test_probe_passes_through(self) -> None:
+        with pytest.raises(ProcessLookupError):
+            os.kill(_NO_SUCH_PID, 0)
+        with pytest.raises(ProcessLookupError):
+            os.killpg(_NO_SUCH_PID, 0)
+
+    def test_own_children_can_be_signalled(
+        self, child: subprocess.Popen[bytes]
+    ) -> None:
+        os.kill(child.pid, signal.SIGSTOP)
+        os.killpg(child.pid, signal.SIGKILL)
+        assert child.wait(timeout=5) == -signal.SIGKILL
+
+
+class TestGuardedPopen:
+    def test_steam_spawn_is_refused(self, blocked: list[str]) -> None:
+        with pytest.raises(RealEffectError):
+            _REAL_POPEN(["steam", "-shutdown"])
+        with pytest.raises(RealEffectError):
+            _REAL_POPEN(["anything"], executable="/usr/bin/steam")
+        assert len(blocked) == 2
+        blocked.clear()
+
+    def test_ordinary_spawns_run(self) -> None:
+        proc = _REAL_POPEN([sys.executable, "-c", "print(1)"], stdout=subprocess.PIPE)
+        out, _ = proc.communicate(timeout=10)
+        assert out.strip() == b"1"
+
+
+class TestDemoTripwire:
+    def test_unpatched_demo_loop_fails_at_once(self, blocked: list[str]) -> None:
+        """The real loop never returns on its own: it must not even tick."""
+        with (
+            patch("steam_backlog_enforcer._enforce_demo._echo"),
+            pytest.raises(RealEffectError),
+        ):
+            run_demo(Config())
+        assert blocked == ["unpatched enforce --demo loop (playtime_tick)"]
+        blocked.clear()
+
+
+class TestFailIfBlocked:
+    def test_fails_naming_each_call(self) -> None:
+        with pytest.raises(pytest.fail.Exception, match="a; b"):
+            _fail_if_blocked(["a", "b"])
+
+    def test_silent_when_clean(self) -> None:
+        _fail_if_blocked([])

steam_backlog_enforcer/tests/test_playtime_cutoff.py

@@ -11,6 +11,7 @@ from steam_backlog_enforcer._playtime_cutoff import (
     _kill_set,
     _sustain_block,
     _warn,
+    reconcile_for,
 )
 from steam_backlog_enforcer._playtime_notify import (
     notify_desktop_user,
@@ -37,7 +38,7 @@ def quiet_tick() -> object:
         mocks = {
             name: stack.enter_context(patch(f"{where}.{name}"))
             for name, where in (
-                ("reconcile", PKG),
+                ("reconcile", cutoff),
                 ("request_steam_shutdown", cutoff),
                 ("kill_gaming_processes", cutoff),
                 ("notify_desktop_user", cutoff),
@@ -79,6 +80,37 @@ class TestBeginCutoff:
         mock_notify.assert_called_once()
         assert out.blocked_at == NOW.timestamp()
 
+    def test_demo_says_out_loud_that_it_will_not_mask(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        state = PlaytimeState(day_key=TODAY, seconds=60.0)
+        with (
+            patch("steam_backlog_enforcer._playtime_cutoff.request_steam_shutdown"),
+            patch("steam_backlog_enforcer._playtime_cutoff.kill_gaming_processes"),
+            patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"),
+            patch(
+                "steam_backlog_enforcer._playtime_cutoff._kill_set", return_value=set()
+            ),
+        ):
+            _begin_cutoff(state, _rules(demo=True), now=NOW)
+            assert "only the daemon does that" in caplog.text
+            caplog.clear()
+            _begin_cutoff(state, _rules(), now=NOW)
+            assert "only the daemon" not in caplog.text
+
+
+class TestReconcileFor:
+    def test_production_drives_the_mounts(self) -> None:
+        with patch("steam_backlog_enforcer._playtime_cutoff.reconcile") as rec:
+            reconcile_for(_rules(), should_block=True)
+        rec.assert_called_once_with(should_block=True)
+
+    @pytest.mark.parametrize("should_block", [True, False])
+    def test_demo_never_touches_them(self, *, should_block: bool) -> None:
+        with patch("steam_backlog_enforcer._playtime_cutoff.reconcile") as rec:
+            reconcile_for(_rules(demo=True), should_block=should_block)
+        rec.assert_not_called()
+
 
 class TestSustainBlock:
     def _run(
@@ -116,6 +148,11 @@ class TestSustainBlock:
         _, mock_kill = self._run(29.0)
         mock_kill.assert_called_once_with({7}, force=False)
 
+    def test_demo_never_mounts_but_keeps_killing(self) -> None:
+        mock_rec, mock_kill = self._run(5.0, demo=True)
+        mock_rec.assert_not_called()
+        mock_kill.assert_called_once()
+
     def test_demo_escalates_sooner(self) -> None:
         _, mock_kill = self._run(11.0, demo=True)
         mock_kill.assert_called_once_with({7}, force=True)

steam_backlog_enforcer/tests/test_playtime_policy.py

@@ -41,7 +41,7 @@ def quiet_tick() -> object:
         mocks = {
             name: stack.enter_context(patch(f"{where}.{name}"))
             for name, where in (
-                ("reconcile", PKG),
+                ("reconcile", cutoff),
                 ("request_steam_shutdown", cutoff),
                 ("kill_gaming_processes", cutoff),
                 ("notify_desktop_user", cutoff),
@@ -61,7 +61,7 @@ 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.reconcile") as mock_rec,
             patch(
                 "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
             ) as mock_notify,
@@ -74,7 +74,7 @@ class TestPolicyBelowBudget:
     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.reconcile"),
             patch(
                 "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
             ) as mock_notify,
@@ -111,7 +111,7 @@ class TestPolicyEnforcementDisabled:
         """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.reconcile") as mock_rec,
             patch(
                 "steam_backlog_enforcer._playtime_cutoff.request_steam_shutdown"
             ) as mock_shutdown,
@@ -142,7 +142,7 @@ class TestPolicyBudgetRoseMidDay:
     def test_the_block_is_released(self) -> None:
         """reconcile is asked to unmount, which is what re-enables gaming."""
         with (
-            patch("steam_backlog_enforcer._playtime.reconcile") as mock_rec,
+            patch("steam_backlog_enforcer._playtime_cutoff.reconcile") as mock_rec,
             patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"),
         ):
             _policy(self._blocked_at_six_hours(), _rules(), now=NOW)
@@ -151,7 +151,7 @@ class TestPolicyBudgetRoseMidDay:
     def test_blocked_at_is_cleared(self) -> None:
         """Otherwise the UI reports "blocked" while gaming works fine."""
         with (
-            patch("steam_backlog_enforcer._playtime.reconcile"),
+            patch("steam_backlog_enforcer._playtime_cutoff.reconcile"),
             patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"),
         ):
             out = _policy(self._blocked_at_six_hours(), _rules(), now=NOW)
@@ -163,7 +163,7 @@ class TestPolicyBudgetRoseMidDay:
         Left alone, the extra two hours would arrive with no warning at all.
         """
         with (
-            patch("steam_backlog_enforcer._playtime.reconcile"),
+            patch("steam_backlog_enforcer._playtime_cutoff.reconcile"),
             patch(
                 "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
             ) as mock_notify,
@@ -179,7 +179,7 @@ class TestPolicyBudgetRoseMidDay:
         spent = 8 * 3600 - 3000.0
         state = PlaytimeState(day_key=TODAY, seconds=spent, warned_seconds=[3600])
         with (
-            patch("steam_backlog_enforcer._playtime.reconcile"),
+            patch("steam_backlog_enforcer._playtime_cutoff.reconcile"),
             patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"),
         ):
             out = _policy(state, _rules(), now=NOW)
@@ -196,7 +196,7 @@ class TestPolicyBudgetRoseMidDay:
             day_key=TODAY, seconds=5 * 3600.0, warned_seconds=[3600, 1800, 600, 300]
         )
         with (
-            patch("steam_backlog_enforcer._playtime.reconcile"),
+            patch("steam_backlog_enforcer._playtime_cutoff.reconcile"),
             patch(
                 "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
             ) as mock_notify,

steam_backlog_enforcer/tests/test_playtime_tick.py

@@ -40,7 +40,7 @@ def quiet_tick() -> object:
         mocks = {
             name: stack.enter_context(patch(f"{where}.{name}"))
             for name, where in (
-                ("reconcile", PKG),
+                ("reconcile", cutoff),
                 ("request_steam_shutdown", cutoff),
                 ("kill_gaming_processes", cutoff),
                 ("notify_desktop_user", cutoff),

web/mock/catalog.ts

@@ -58,7 +58,7 @@ export const BASE: Base[] = [
   spec('unblock', 'Unblock the store for a few minutes', 'store', 'job', {
     params: [MINUTES], friction: phrase('unblock the store for {minutes} minutes'), privileged: true,
   }),
-  spec('add-exception', 'Request a 24h-locked whitelist exception', 'store', 'job', {
+  spec('add-exception', 'Whitelist a game immediately (phrase + reason)', 'store', 'job', {
     params: [APP_ID, { name: 'reason', label: 'Reason', type: 'string', required: true, help: 'Why this game must be allowed.' }],
     friction: phrase('request exception for {game_name}'),
   }),
@@ -70,12 +70,11 @@ export const BASE: Base[] = [
     friction: phrase('wipe all enforcer state'),
   }),
   spec('setup', 'First-time setup: Steam API key and Steam ID', 'system', 'screen'),
-  spec('enforce', 'Restart the enforcer daemon, or run a 60 s demo budget', 'system', 'job', {
+  spec('enforce', 'Run enforcer: block, uninstall, kill, hide (--demo for a 60s budget)', 'system', 'screen', {
     params: [{
-      name: 'mode', label: 'Mode', type: 'string', required: true, default: 'restart',
-      help: '"restart" (rate-limited: once per 10 min) or "demo" (60 s budget).',
+      name: 'demo', label: 'Demo mode (1 = 60-second budget)', type: 'int', required: false, default: 0, min: 0, max: 1,
     }],
-    privileged: true,
+    privileged: true, cancellable: true,
   }),
   spec('gaming-reset', "Reset today's gaming counter", 'gaming', 'job', {
     friction: phrase("reset today's gaming budget", 300), privileged: true,
@@ -90,7 +89,7 @@ const MANUAL_EXEMPT = new Set<CommandName>([
   'done', 'check', 'status', 'enforce', 'setup', 'serve', 'abandon-pick', 'pick-manual',
   'gaming-status', 'gaming-unblock', 'gaming-reset',
 ])
-const TOTAL_EXEMPT = new Set<CommandName>(['status', 'enforce', 'gaming-status', 'gaming-unblock'])
+const TOTAL_EXEMPT = new Set<CommandName>(['status', 'enforce', 'gaming-status', 'gaming-unblock', 'serve'])
 
 export function lockedReason(name: CommandName): string | null {
   if (world.lock === 'total' && !TOTAL_EXEMPT.has(name)) {

web/mock/http.ts

@@ -76,8 +76,5 @@ export function validateParams(spec: CommandSpec, raw: unknown): Record<string,
     const v = checkParam(p, input[p.name])
     if (v !== undefined) out[p.name] = v
   }
-  if (spec.name === 'enforce' && out.mode !== 'restart' && out.mode !== 'demo') {
-    throw new HttpError(400, 'invalid_params', 'Mode must be "restart" or "demo".')
-  }
   return out
 }

web/mock/routes.ts

@@ -68,13 +68,14 @@ on('GET', /^\/api\/jobs$/, () => listJobs())
 on('POST', /^\/api\/jobs$/, async ({ req, res }) => {
   const body = (await readJson(req)) as Partial<JobRequest>
   const spec = findSpec(String(body.command))
-  if (!spec || spec.kind !== 'job') throw new HttpError(400, 'unknown_command', `Unknown command "${body.command}".`)
+  // enforce is a "screen" in the catalog but still runs as a job, as on the real server.
+  if (!spec || (spec.kind !== 'job' && spec.name !== 'enforce')) throw new HttpError(400, 'unknown_command', `Unknown command "${body.command}".`)
   if (spec.locked_reason) throw new HttpError(409, 'locked', spec.locked_reason)
   const params = validateParams(spec, body.params)
   if (spec.privileged && world.daemon === 'unreachable') {
     throw new HttpError(503, 'daemon_unreachable', 'The root daemon is not answering on /run/steam-backlog-enforcer/ctl.sock.')
   }
-  if (spec.name === 'enforce' && params.mode === 'restart' && world.restartAvailableAt && world.restartAvailableAt > Date.now()) {
+  if (spec.name === 'enforce' && params.demo !== 1 && world.restartAvailableAt && world.restartAvailableAt > Date.now()) {
     throw new HttpError(429, 'rate_limited', 'The daemon was restarted less than 10 minutes ago.')
   }
   if (spec.friction) {

web/mock/scripts.ts

@@ -117,7 +117,7 @@ const unblock: Script = async (ctx) => {
 }
 
 const enforce: Script = async (ctx) => {
-  if (ctx.params.mode === 'demo') {
+  if (ctx.params.demo === 1) {
     await walk(ctx, 'Demo enforcement (60 s budget)', ['arm', 'tick', 'tick', 'warn', 'cutoff'], 900)
     return { summary: 'Demo run finished: cutoff engaged after 60 s.' }
   }

web/src/api/errors.ts

@@ -20,7 +20,7 @@ const EXPLAIN: Record<FailureCode, { title: string; hint: string }> = {
   not_cancellable: { title: 'Cannot cancel', hint: 'This command cannot be stopped once started.' },
   op_failed: { title: 'Action failed', hint: 'The daemon or the job process hit an error. See the job log or the daemon journal on System.' },
   server_stale: { title: 'Server is restarting', hint: 'It was running outdated code and restarts on the current code. Retry in a few seconds.' },
-  unsupported: { title: 'Not supported here', hint: 'The enforcer daemon cannot do this in its current mode.' },
+  unsupported: { title: 'Not supported here', hint: 'Nothing supervises this process (it is not running under systemd), so a restart would be a stop.' },
   network: { title: 'Server not reachable', hint: 'Is it running? Start it with ./run.sh serve.' },
   http: { title: 'Unexpected server error', hint: 'Check the server log.' },
 }

web/src/commands/PendingCountdown.test.tsx

@@ -39,15 +39,26 @@ describe('PendingCountdown', () => {
   it('counts down on the server’s clock', async () => {
     stubApi({ 'POST /api/pending/p1/heartbeat': pending() })
     mount()
-    expect(screen.getByText('0:30')).toBeInTheDocument()
+    // One second past ready_at: a slightly fast browser clock must not offer
+    // a commit the daemon refuses.
+    expect(screen.getByText('0:31')).toBeInTheDocument()
     expect(screen.getByText('until commit unlocks')).toBeInTheDocument()
     await tick(10_000)
-    expect(screen.getByText('0:20')).toBeInTheDocument()
-    expect(screen.getByRole('progressbar', { name: 'Countdown' })).toHaveAttribute('aria-valuenow', '33')
+    expect(screen.getByText('0:21')).toBeInTheDocument()
+    expect(screen.getByRole('progressbar', { name: 'Countdown' })).toHaveAttribute('aria-valuenow', '32')
     expect(screen.getByText(/Closing this dialog/)).toBeInTheDocument()
     expect(screen.getByRole('button', { name: 'Cancel countdown' })).toHaveFocus()
   })
 
+  it('keeps the commit locked until a second after the server’s ready time', async () => {
+    stubApi({ 'POST /api/pending/p1/heartbeat': pending() })
+    mount()
+    await tick(30_000)
+    expect(screen.getByText('until commit unlocks')).toBeInTheDocument()
+    await tick(1_000)
+    expect(screen.getByText('Ready to commit')).toBeInTheDocument()
+  })
+
   it('heartbeats at half the interval, and takes the server’s new ready time', async () => {
     const mock = stubApi({ 'POST /api/pending/p1/heartbeat': () => pending({ ready_at: '2026-10-10T10:01:00Z' }) })
     mount()
@@ -56,7 +67,7 @@ describe('PendingCountdown', () => {
     await tick(1)
     expect(mock).toHaveBeenCalledTimes(1)
     expect(screen.getByText(/Heartbeat every 5 s · last 0 s ago/)).toBeInTheDocument()
-    expect(screen.getByText('0:55')).toBeInTheDocument() // pushed out by the server
+    expect(screen.getByText('0:56')).toBeInTheDocument() // pushed out by the server
   })
 
   it('never heartbeats faster than once a second', async () => {
@@ -100,7 +111,7 @@ describe('PendingCountdown', () => {
     mount()
     window.dispatchEvent(new Event('pagehide'))
     await tick(0)
-    expect(screen.getByText('0:30')).toBeInTheDocument()
+    expect(screen.getByText('0:31')).toBeInTheDocument()
   })
 
   it('stops heartbeating and listening on unmount', async () => {

web/src/commands/PendingCountdown.tsx

@@ -9,6 +9,13 @@ import { useNow } from '../ui/useNow'
 import { phraseMatches } from './catalog'
 import { PhraseField } from './PhraseField'
 
+/**
+ * The daemon checks `ready_at` on its own clock; unlocking the commit a
+ * second late keeps a slightly fast browser clock from offering a commit the
+ * daemon refuses (seen in vmbox: refused 34 ms early, countdown_running).
+ */
+const CLOCK_SLACK_MS = 1000
+
 interface Props {
   pending: PendingAction
   onCommitted: (job: Job) => void
@@ -53,7 +60,7 @@ export function PendingCountdown({ pending: initial, onCommitted, onCancel, onRe
   const cancel = useMutation({ mutationFn: () => api.dropPending(pending.id), onSettled: onCancel })
 
   const armed = Date.parse(pending.armed_at)
-  const ready = Date.parse(pending.ready_at)
+  const ready = Date.parse(pending.ready_at) + CLOCK_SLACK_MS
   const left = Math.max(0, (ready - now) / 1000)
   const isReady = left <= 0
   const lapsed = beatError instanceof ApiFailure && beatError.code === 'pending_lapsed'

web/src/commands/RunnerDialog.test.tsx

@@ -45,6 +45,18 @@ describe('RunnerDialog', () => {
     expect(await run('Arm countdown')).toBeInTheDocument()
   })
 
+  it('runs the enforce preset it was opened for, with no mode field to change it', async () => {
+    const demo = { name: 'demo', label: 'Demo mode', type: 'int' as const, required: false, default: 0, min: 0, max: 1 }
+    const mock = stubApi(routes([makeSpec('enforce', { kind: 'screen', params: [demo], privileged: true, cancellable: true })], { 'POST /api/jobs': makeJob({ id: 'j1' }) }))
+    open('enforce', { demo: 1 })
+    const button = await run('Run enforce')
+    expect(screen.getByText('Cancellable')).toBeInTheDocument()
+    expect(screen.queryByLabelText(/Demo mode/)).toBeNull()
+    expect(screen.queryByText('Runs as root via the daemon')).toBeNull()
+    await userEvent.click(button)
+    expect(bodiesOf(mock, 'POST /api/jobs')).toEqual([{ command: 'enforce', params: { demo: 1 } }])
+  })
+
   it('tags a state-changing command, with no countdown for an immediate one', async () => {
     stubApi(routes([makeSpec('scan', { friction: { phrase_template: 'x' } })]))
     open('scan')

web/src/commands/RunnerDialog.tsx

@@ -9,6 +9,7 @@ import { JobProgress } from '../jobs/JobProgress'
 import { Dialog } from '../ui/Dialog'
 import { CommandLine, ErrorNotice, Loading, Notice } from '../ui/Notice'
 import { commandLabel, fillPhrase, findCommand, phraseMatches } from './catalog'
+import { enforcePreset } from './enforcePreset'
 import { PendingCountdown } from './PendingCountdown'
 import { ParamsForm } from './ParamsForm'
 import { initialValues, paramError, toParams } from './params'
@@ -45,7 +46,7 @@ export function RunnerDialog({ name, preset, onClose }: Props) {
   return (
     <Dialog open title={commandLabel(name)} onClose={close} wide>
       {spec ? (
-        <Runner spec={spec} preset={preset} onClose={close} onArmed={(id) => (armed.current = id)} />
+        <Runner {...enforcePreset(spec, preset)} preset={preset} onClose={close} onArmed={(id) => (armed.current = id)} />
       ) : loadError ? (
         <ErrorNotice error={loadError} />
       ) : commands ? (
@@ -59,12 +60,14 @@ export function RunnerDialog({ name, preset, onClose }: Props) {
 
 interface RunnerProps {
   spec: CommandSpec
+  /** Params sent as-is, outside the form (see enforcePreset). */
+  fixed: Record<string, number | string>
   preset?: Record<string, number | string>
   onClose: () => void
   onArmed: (pendingId: string | null) => void
 }
 
-function Runner({ spec, preset, onClose, onArmed }: RunnerProps) {
+function Runner({ spec, fixed, preset, onClose, onArmed }: RunnerProps) {
   const { games, ctx } = useGameLookup()
   const [values, setValues] = useState(() => initialValues(spec.params, preset))
   const [typed, setTyped] = useState('')
@@ -80,7 +83,7 @@ function Runner({ spec, preset, onClose, onArmed }: RunnerProps) {
     },
   })
 
-  const params = toParams(spec.params, values)
+  const params = { ...fixed, ...toParams(spec.params, values) }
   const errors = spec.params.map((p) => paramError(p, values[p.name] ?? '', games)).filter(Boolean)
   const phrase = spec.friction ? fillPhrase(spec.friction.phrase_template, params, ctx) : null
   const countdown = spec.friction?.countdown_seconds

web/src/commands/enforcePreset.test.ts

@@ -0,0 +1,34 @@
+import { describe, expect, it } from 'vitest'
+import type { ParamSpec } from '../api/contract'
+import { makeSpec } from '../test/fixtures'
+import { enforcePreset } from './enforcePreset'
+
+const DEMO: ParamSpec = { name: 'demo', label: 'Demo mode', type: 'int', required: false, default: 0, min: 0, max: 1 }
+const enforce = makeSpec('enforce', { kind: 'screen', params: [DEMO], privileged: true, cancellable: true })
+
+describe('enforcePreset', () => {
+  it('leaves other commands and an unpreset enforce alone', () => {
+    const scan = makeSpec('scan')
+    expect(enforcePreset(scan, { demo: 1 })).toEqual({ spec: scan, fixed: {} })
+    expect(enforcePreset(enforce, undefined)).toEqual({ spec: enforce, fixed: {} })
+    expect(enforcePreset(enforce, { other: 1 })).toEqual({ spec: enforce, fixed: {} })
+  })
+
+  it('makes the demo a cancellable user job with no mode field', () => {
+    const { spec, fixed } = enforcePreset(enforce, { demo: 1 })
+    expect(fixed).toEqual({ demo: 1 })
+    expect(spec.params).toEqual([])
+    expect(spec.privileged).toBe(false)
+    expect(spec.cancellable).toBe(true)
+    expect(spec.description).toMatch(/60-second demo budget/)
+  })
+
+  it('makes the restart a root op that cannot be cancelled', () => {
+    const { spec, fixed } = enforcePreset(enforce, { demo: '0' })
+    expect(fixed).toEqual({ demo: 0 })
+    expect(spec.params).toEqual([])
+    expect(spec.privileged).toBe(true)
+    expect(spec.cancellable).toBe(false)
+    expect(spec.description).toMatch(/^Restart the enforcer daemon/)
+  })
+})

web/src/commands/enforcePreset.ts

@@ -0,0 +1,31 @@
+import type { CommandSpec } from '../api/contract'
+
+/**
+ * `enforce` is two operations behind one catalog spec: `demo: 1` is a 60 s
+ * demo that runs as a desktop-user job and loops until cancelled, `demo: 0`
+ * is the daemon restart over ctl.sock (the server's `is_privileged`). The
+ * System buttons preset one of them, so the dialog shows that operation's
+ * own flags and keeps the mode out of the form — a restart dialog must not
+ * turn into a demo, nor claim it can be cancelled.
+ *
+ * Returns the spec to render and the params to send whatever the form holds.
+ */
+export function enforcePreset(
+  spec: CommandSpec,
+  preset: Record<string, number | string> | undefined,
+): { spec: CommandSpec; fixed: Record<string, number | string> } {
+  if (spec.name !== 'enforce' || preset?.demo === undefined) return { spec, fixed: {} }
+  const demo = Number(preset.demo) === 1
+  return {
+    spec: {
+      ...spec,
+      description: demo
+        ? 'Run the enforcer with a 60-second demo budget and its own state file. It keeps running until you cancel it.'
+        : 'Restart the enforcer daemon. State is flushed first and systemd brings it straight back; once per 10 minutes.',
+      params: spec.params.filter((p) => p.name !== 'demo'),
+      privileged: !demo,
+      cancellable: demo,
+    },
+    fixed: { demo: demo ? 1 : 0 },
+  }
+}

web/src/screens/system/DaemonCard.test.tsx

@@ -42,7 +42,7 @@ describe('DaemonCard', () => {
   it('restarts through the runner when allowed', async () => {
     show(daemon())
     await userEvent.click(await screen.findByRole('button', { name: 'Restart daemon…' }))
-    expect(run).toHaveBeenCalledWith('enforce', { mode: 'restart' })
+    expect(run).toHaveBeenCalledWith('enforce', { demo: 0 })
   })
 
   it('refuses a restart while rate-limited, showing the wait', async () => {
@@ -74,7 +74,7 @@ describe('DaemonCard', () => {
   it('offers the demo run', async () => {
     show(daemon())
     await userEvent.click(await screen.findByRole('button', { name: 'Demo run (enforce --demo)…' }))
-    expect(run).toHaveBeenCalledWith('enforce', { mode: 'demo' })
+    expect(run).toHaveBeenCalledWith('enforce', { demo: 1 })
   })
 
   it('shows loading, then a failure', async () => {

web/src/screens/system/DaemonCard.tsx

@@ -1,7 +1,6 @@
 import { useQuery } from '@tanstack/react-query'
 import { useEffect, useRef } from 'react'
 import { q } from '../../api/queries'
-import { CommandButton } from '../../commands/CommandCard'
 import { useRunner } from '../../commands/runnerContext'
 import { ago, clock } from '../../ui/format'
 import { CommandLine, ErrorNotice, Loading, Notice } from '../../ui/Notice'
@@ -49,12 +48,16 @@ export function DaemonCard() {
           className="btn btn-ghost"
           aria-disabled={limited || data.state !== 'running'}
           onClick={() => {
-            if (!limited && data.state === 'running') run('enforce', { mode: 'restart' })
+            if (!limited && data.state === 'running') run('enforce', { demo: 0 })
           }}
         >
           {limited ? `Restart available in ${clock(waitMs / 1000)}` : 'Restart daemon…'}
         </button>
-        <CommandButton name="enforce" preset={{ mode: 'demo' }} label="Demo run (enforce --demo)…" />
+        {/* enforce is a "screen" command, so CommandButton would render a link
+            back to this page; open the runner directly, as restart does. */}
+        <button type="button" className="btn btn-ghost" onClick={() => run('enforce', { demo: 1 })}>
+          Demo run (enforce --demo)…
+        </button>
       </div>
       <h3 className="h3">Journal</h3>
       <pre ref={logRef} className="journal" tabIndex={0} aria-label="Daemon journal (live)">