Timed store window that survives hosts reinstalls; whitelist exceptions stay visible

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

README.md

@@ -33,7 +33,8 @@ python -m python_pkg.steam_backlog_enforcer.main setup
 | `list`      | List incomplete games from snapshot        |
 | `skip`      | Skip the currently assigned game           |
 | `enforce`   | Run enforcer (block, uninstall, kill)      |
-| `unblock`   | Remove store blocking                      |
+| `unblock [minutes]` | Unblock the store for a window (default 15, max 30); the daemon keeps it open and re-blocks after |
+| `buy-dlc`   | Same 15-minute store window, to buy a game or DLC |
 | `reset`     | Reset all state                            |
 | `installed` | List currently installed Steam games       |
 | `uninstall` | Interactively uninstall non-assigned games |

steam_backlog_enforcer/_enforce_loop.py

@@ -22,6 +22,7 @@ from steam_backlog_enforcer._pick_completion import (
 from steam_backlog_enforcer._playtime import playtime_tick
 from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session
 from steam_backlog_enforcer._steam_launch import steam_is_installed
+from steam_backlog_enforcer._store_window import store_window_tick
 from steam_backlog_enforcer._total_block import (
     end_total_block_cleanup,
     enforce_total_block_tick,
@@ -132,6 +133,10 @@ def _enforce_loop_iteration(
     if total_block_needs_cleanup():
         end_total_block_cleanup()
 
+    # Before the Steam-installed guard: an expired window must re-block even
+    # when Steam is gone.
+    store_window_tick(config, state)
+
     # With no Steam client there is no library, no installs and no game
     # processes, so every branch below is a no-op at best - and at worst a
     # 3s-interval error loop trying to write manifests into a steamapps

steam_backlog_enforcer/_scanning_tampering.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any
 from steam_backlog_enforcer._allowed_games import allowed_games
 from steam_backlog_enforcer._assignment_progress import iso_to_epoch
 from steam_backlog_enforcer._snapshot import load_snapshot
+from steam_backlog_enforcer._whitelist_locking import get_approved_exception_ids
 from steam_backlog_enforcer.enforcer import send_notification
 from steam_backlog_enforcer.game_install import _echo
 from steam_backlog_enforcer.steam_api import SteamAPIClient
@@ -32,7 +33,8 @@ def _legitimately_played(state: State) -> set[int]:
     lands in ``finished_app_ids`` -- but the achievements that finished it were
     earned legitimately, so comparing it against a stale snapshot would report
     the user's own completion as tampering. Every game ever manually picked,
-    plus every finished game, is therefore exempt.
+    plus every finished game and every approved whitelist exception, is therefore
+    exempt.
 
     Args:
         state: Current enforcer state.
@@ -42,6 +44,7 @@ def _legitimately_played(state: State) -> set[int]:
     """
     exempt = {app_id for app_id, _ in allowed_games(state)}
     exempt.update(state.finished_app_ids)
+    exempt.update(get_approved_exception_ids())
     exempt.update(
         pick["app_id"] for pick in state.manual_picks if pick.get("app_id") is not None
     )

steam_backlog_enforcer/_store_tools.py

@@ -11,19 +11,11 @@ from pathlib import Path
 import shutil
 import subprocess
 
-# Path to the hosts install script. _REPO_ROOT resolves to $HOME (this
-# module lives two levels below it); the script itself is in the
-# linux_configuration checkout under testsAndMisc, not directly under $HOME.
-_REPO_ROOT = Path(__file__).resolve().parents[2]
-HOSTS_INSTALL_SCRIPT = (
-    _REPO_ROOT
-    / "testsAndMisc"
-    / "linux_configuration"
-    / "scripts"
-    / "periodic_background"
-    / "hosts"
-    / "install.sh"
-)
+# Path to the hosts install script. _SRC_ROOT is ~/src (this module lives two
+# levels below it); the generator moved out of testsAndMisc into its own
+# hosts-blocker repo, whose custom_entries.hosts carries the Steam Store block.
+_SRC_ROOT = Path(__file__).resolve().parents[2]
+HOSTS_INSTALL_SCRIPT = _SRC_ROOT / "hosts-blocker" / "install.sh"
 
 # iptables chain name for our blocking rules.
 IPTABLES_CHAIN = "STEAM_ENFORCER"

steam_backlog_enforcer/_store_window.py

@@ -0,0 +1,103 @@
+"""A bounded window in which the Steam store stays unblocked.
+
+``run.sh unblock`` used to comment the store lines out of ``/etc/hosts`` once
+and walk away. Anything that reinstalls the hosts file -- the hourly
+``periodic-system-maintenance`` job, any other ``hosts-blocker/install.sh`` run
+-- put the block straight back, sometimes within seconds, so a purchase could
+not be finished. The window moves that job into the daemon: until the
+deadline it re-applies the unblock whenever the block reappears, and after it
+re-blocks exactly once.
+
+Only the CLI writes the deadline (``State.store_unblocked_until``). The daemon
+runs as root and never saves state for this: a root-written ``state.json``
+would lock the unprivileged CLI out of its own file.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+import logging
+from typing import TYPE_CHECKING
+
+from steam_backlog_enforcer.store_blocker import (
+    block_store,
+    hosts_blocks_store,
+    unblock_store,
+)
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.config import Config, State
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_WINDOW_MINUTES = 15
+MAX_WINDOW_MINUTES = 30
+
+# Deadlines this daemon process has already re-blocked for. In memory on
+# purpose (see module docstring); a restarted daemon re-blocks at startup anyway.
+_reblocked_for: set[str] = set()
+
+
+def open_store_window(state: State, minutes: int) -> datetime:
+    """Record a store window ending *minutes* from now and unblock at once.
+
+    Args:
+        state: Enforcer state; saved with the new deadline.
+        minutes: Window length, 1 to ``MAX_WINDOW_MINUTES``.
+
+    Returns:
+        The UTC deadline.
+
+    Raises:
+        ValueError: If *minutes* is out of range.
+        RuntimeError: If the store could not be unblocked right now.
+    """
+    if not 1 <= minutes <= MAX_WINDOW_MINUTES:
+        msg = f"minutes must be between 1 and {MAX_WINDOW_MINUTES}, got {minutes}"
+        raise ValueError(msg)
+    until = datetime.now(UTC) + timedelta(minutes=minutes)
+    state.store_unblocked_until = until.isoformat()
+    state.save()
+    if not unblock_store():
+        msg = "could not unblock the store (sudo?)"
+        raise RuntimeError(msg)
+    return until
+
+
+def _deadline(state: State) -> datetime | None:
+    """The window's deadline, or ``None`` when unset or unparsable."""
+    raw = state.store_unblocked_until
+    if not raw:
+        return None
+    try:
+        return datetime.fromisoformat(raw)
+    except ValueError:
+        logger.warning("Ignoring malformed store_unblocked_until %r", raw)
+        return None
+
+
+def store_window_tick(config: Config, state: State) -> None:
+    """Keep the store open inside the window; re-block once it has passed.
+
+    Runs every daemon iteration. Outside a window it only parses one string,
+    so it costs nothing; inside one it reads ``/etc/hosts`` (no subprocess)
+    and shells out only when the block has actually come back.
+
+    Args:
+        config: Enforcer configuration (``block_store`` gates the re-block).
+        state: Freshly reloaded enforcer state.
+    """
+    until = _deadline(state)
+    if until is None:
+        return
+    if datetime.now(UTC) < until:
+        if hosts_blocks_store():
+            logger.info("Store window open until %s: block came back, undoing", until)
+            unblock_store()
+        return
+    if state.store_unblocked_until in _reblocked_for:
+        return
+    _reblocked_for.add(state.store_unblocked_until)
+    if config.block_store:
+        logger.info("Store window ended at %s: re-blocking the store", until)
+        block_store()

steam_backlog_enforcer/config.py

@@ -191,6 +191,8 @@ class State:
     """``str(app_id)`` → ISO time of its latest assignment (pick ordering)."""
     released_at: dict[str, str] = field(default_factory=dict)
     """``str(app_id)`` → ISO time it was released below 100% (tampering)."""
+    store_unblocked_until: str = ""
+    """ISO deadline of the open store window (``unblock``); CLI-written only."""
 
     def skip_for_days(self, app_id: int, days: int) -> None:
         """Mark ``app_id`` as skipped for ``days`` days from now (UTC)."""

steam_backlog_enforcer/library_hider.py

@@ -32,6 +32,7 @@ from steam_backlog_enforcer._steam_launch import (
     restart_steam,
     steam_is_installed,
 )
+from steam_backlog_enforcer._whitelist_locking import get_approved_exception_ids
 
 if TYPE_CHECKING:
     from steam_backlog_enforcer.config import Config, State
@@ -81,10 +82,15 @@ def hide_other_games(
     cover games that might not yet appear in ``visibleApps`` due to
     stale MobX state.
 
+    Approved whitelist exceptions (``add-exception``) are always treated as
+    allowed, so a permanently whitelisted game such as a co-op title stays
+    visible next to the assignment instead of being hidden with the rest.
+
     Returns the total number of games hidden across all passes.
     """
     ensure_steam_debug_port()
 
+    allowed_app_ids = allowed_app_ids | get_approved_exception_ids()
     allowed_json = json.dumps(sorted(allowed_app_ids))
     extra_ids = sorted(aid for aid in owned_app_ids if aid not in allowed_app_ids)
     extra_json = json.dumps(extra_ids)

steam_backlog_enforcer/main/__init__.py

@@ -125,8 +125,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
     "install": ("Install the assigned game", cmd_install),
     "hide": ("Hide all non-assigned games in library", cmd_hide),
     "unhide": ("Unhide all games in library", cmd_unhide),
-    "unblock": ("Remove store blocking", cmd_unblock),
-    "buy-dlc": ("Temporarily unblock store to buy DLC", cmd_buy_dlc),
+    "buy-dlc": ("Unblock the store for 15 min to buy a game/DLC", cmd_buy_dlc),
     "reset": ("Reset all state", cmd_reset),
     "installed": ("List installed games", cmd_installed),
     "uninstall": ("Uninstall all non-assigned games", cmd_uninstall),
@@ -141,6 +140,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
 # Extra commands with non-standard arg handling (shown in help but not in COMMANDS).
 _EXTRA_COMMAND_DESCRIPTIONS: dict[str, str] = {
     "add-exception": "Request 24h-locked whitelist exception (use --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",
     "abandon-pick": "Undo a manual pick at any time (needs app_id)",
@@ -179,26 +179,24 @@ def _dispatch_extra_command(command: str, config: Config, state: State) -> bool:
     Returns:
         True if *command* was handled here.
     """
-    if command == "add-exception":
-        cmd_add_exception(sys.argv[2:])
-        return True
-    if command == "block-gaming":
-        cmd_block_gaming(sys.argv[2:])
-        return True
-    if command == "serve":
-        cmd_serve(sys.argv[2:])
-        return True
-    if command == "pick-manual":
-        cmd_pick_manual(config, state, sys.argv[2:])
-        return True
-    if command == "abandon-pick":
-        cmd_abandon_pick(config, state, sys.argv[2:])
-        return True
+    args = sys.argv[2:]
     if command == "enforce":
-        sys.exit(cmd_enforce(config, state, sys.argv[2:]))
+        sys.exit(cmd_enforce(config, state, args))
     if command == "gaming-unblock":
-        sys.exit(cmd_gaming_unblock(sys.argv[2:]))
+        sys.exit(cmd_gaming_unblock(args))
+    handlers: dict[str, Callable[[], object]] = {
+        "add-exception": lambda: cmd_add_exception(args),
+        "unblock": lambda: cmd_unblock(config, state, args),
+        "block-gaming": lambda: cmd_block_gaming(args),
+        "serve": lambda: cmd_serve(args),
+        "pick-manual": lambda: cmd_pick_manual(config, state, args),
+        "abandon-pick": lambda: cmd_abandon_pick(config, state, args),
+    }
+    handler = handlers.get(command)
+    if handler is None:
         return False
+    handler()
+    return True
 
 
 def main() -> None:

steam_backlog_enforcer/main/misc.py

@@ -16,14 +16,19 @@ from steam_backlog_enforcer._serve_startup import (
     ensure_port_available,
     parse_serve_args,
 )
+from steam_backlog_enforcer._store_window import (
+    DEFAULT_WINDOW_MINUTES,
+    MAX_WINDOW_MINUTES,
+    open_store_window,
+)
 from steam_backlog_enforcer._total_block import start_total_block
 from steam_backlog_enforcer._web_build import build_frontend, frontend_is_stale
 from steam_backlog_enforcer._web_server import serve
 from steam_backlog_enforcer._whitelist import validate_reason
 from steam_backlog_enforcer._whitelist_locking import add_pending_exception
 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
+from steam_backlog_enforcer.library_hider import unhide_all_games
+from steam_backlog_enforcer.store_blocker import unblock_store
 
 if TYPE_CHECKING:
     from steam_backlog_enforcer.config import Config, State
@@ -48,38 +53,34 @@ _BLOCK_GAMING_USAGE = (
 )
 
 
-def cmd_unblock(_config: Config, _state: State) -> None:
-    """Remove store blocking."""
-    if unblock_store():
-        _echo("Steam store unblocked.")
-    else:
-        _echo("Failed to unblock. Run with sudo.")
+def cmd_unblock(_config: Config, state: State, args: list[str] | None = None) -> None:
+    """Open a timed store window: unblocked now, re-blocked by the daemon after.
+
+    Usage: unblock [minutes]   (default 15, max 30)
+
+    Args:
+        _config: Unused; kept for the shared command signature.
+        state: Enforcer state; receives the window deadline.
+        args: CLI argument list after the command name.
+    """
+    raw = args[0] if args else str(DEFAULT_WINDOW_MINUTES)
+    try:
+        until = open_store_window(state, int(raw))
+    except ValueError as exc:
+        _echo(f"Usage: unblock [minutes]  (1-{MAX_WINDOW_MINUTES}): {exc}")
+        sys.exit(1)
+    except RuntimeError as exc:
+        _echo(f"Failed to unblock: {exc}")
+        sys.exit(1)
+    local = until.astimezone().strftime("%H:%M")
+    _echo(f"Steam store UNBLOCKED until {local}.")
+    _echo("The enforcer daemon keeps it open until then (even if a hosts")
+    _echo("reinstall re-blocks it) and re-blocks it afterwards.")
 
 
 def cmd_buy_dlc(config: Config, state: State) -> None:
-    """Temporarily unblock the store so the user can buy DLC."""
-    if state.current_app_id is None:
-        _echo("No game currently assigned.")
-        return
-
-    _echo(f"Current game: {state.current_game_name} (AppID={state.current_app_id})")
-    _echo("Unblocking Steam store for DLC purchase...")
-
-    if not unblock_store():
-        _echo("Failed to unblock store. Run with sudo.")
-        return
-
-    _echo("\nStore UNBLOCKED — buy your DLC now.")
-    _echo("Press Enter when you're done to re-block the store...")
-    input()
-
-    if config.block_store:
-        if block_store():
-            _echo("Store re-blocked. Restarting Steam to clear DNS cache...")
-            restart_steam()
-            _echo("Done.")
-        else:
-            _echo("Warning: failed to re-block store.")
+    """Open the default-length store window to buy a game or DLC."""
+    cmd_unblock(config, state, [])
 
 
 def cmd_reset(config: Config, state: State) -> None:

steam_backlog_enforcer/store_blocker.py

@@ -39,7 +39,12 @@ logger = logging.getLogger(__name__)
 
 
 def is_store_blocked() -> bool:
-    """Check if Steam Store domains are blocked in /etc/hosts."""
+    """Check if Steam Store domains are blocked in /etc/hosts or iptables."""
+    return hosts_blocks_store() or _is_iptables_blocked()
+
+
+def hosts_blocks_store() -> bool:
+    """Check /etc/hosts alone: a file read, no subprocess, safe to poll."""
     try:
         content = HOSTS_FILE.read_text(encoding="utf-8")
         # Check for at least the primary store domain.
@@ -55,8 +60,7 @@ def is_store_blocked() -> bool:
                     return True
     except OSError:
         pass
-
-    return _is_iptables_blocked()
+    return False
 
 
 def block_store() -> bool:

steam_backlog_enforcer/tests/test_budget_view.py

@@ -28,7 +28,12 @@ from steam_backlog_enforcer.config import Config
 # effect yet -- without this patch, collecting this module makes a real HTTP
 # call to the screen locker and pins RULES to whatever budget today happens to
 # have earned. Stub it explicitly so the constant is the earned budget always.
-with patch(
+# The carry is pinned for the same reason: conftest's redirects are not active
+# at import either, so rules_for would read the LIVE playtime_state.json and
+# fold whatever Fri-Mon carry it holds today into RULES.
+with (
+    patch("steam_backlog_enforcer._playtime_state.carry_into", return_value=0.0),
+    patch(
         "steam_backlog_enforcer._playtime_state.resolve_budget",
         side_effect=lambda config: BudgetResolution(
             seconds=float(config.base_gaming_seconds)
@@ -41,6 +46,7 @@ with patch(
             reason="stubbed: fully earned",
             reading_seconds=float(config.reading_bonus_seconds),
         ),
+    ),
 ):
     RULES = rules_for(Config(), demo=False)
 

steam_backlog_enforcer/tests/test_exception_visibility.py

@@ -0,0 +1,62 @@
+"""Approved whitelist exceptions stay visible and never count as tampering."""
+
+from __future__ import annotations
+
+import json
+import re
+from unittest.mock import patch
+
+from steam_backlog_enforcer._scanning_tampering import _legitimately_played
+from steam_backlog_enforcer.config import State
+from steam_backlog_enforcer.library_hider import hide_other_games
+
+PKG = "steam_backlog_enforcer.library_hider"
+_EXCEPTION = 3126150
+_ASSIGNED = 1091500
+_OTHER = 10
+
+
+def _hide_and_capture(allowed: set[int]) -> tuple[list[int], list[int]]:
+    """Run hide_other_games against a fake Steam; return (visible, to_hide)."""
+    captured: dict[str, str] = {}
+
+    def fake_eval(js: str) -> dict[str, object]:
+        captured["js"] = js
+        return {}
+
+    with (
+        patch(f"{PKG}.ensure_steam_debug_port"),
+        patch(f"{PKG}._evaluate_js", side_effect=fake_eval),
+        patch(f"{PKG}._cdp_result_value", return_value='{"totalHidden": 0}'),
+        patch(
+            f"{PKG}.get_approved_exception_ids",
+            return_value=frozenset({_EXCEPTION}),
+        ),
+    ):
+        hide_other_games([_EXCEPTION, _ASSIGNED, _OTHER], allowed)
+    visible = re.search(r"new Set\((\[.*?\])\)", captured["js"])
+    to_hide = re.search(r"const extraIds = (\[.*?\]);", captured["js"])
+    assert visible is not None
+    assert to_hide is not None
+    return json.loads(visible.group(1)), json.loads(to_hide.group(1))
+
+
+def test_exception_is_kept_visible() -> None:
+    visible, to_hide = _hide_and_capture({_ASSIGNED})
+    assert visible == [_ASSIGNED, _EXCEPTION]
+    assert to_hide == [_OTHER]
+
+
+def test_caller_set_is_not_mutated() -> None:
+    allowed = {_ASSIGNED}
+    _hide_and_capture(allowed)
+    assert allowed == {_ASSIGNED}
+
+
+def test_exception_is_exempt_from_tampering() -> None:
+    with patch(
+        "steam_backlog_enforcer._scanning_tampering.get_approved_exception_ids",
+        return_value=frozenset({_EXCEPTION}),
+    ):
+        exempt = _legitimately_played(State(current_app_id=_ASSIGNED))
+    assert {_EXCEPTION, _ASSIGNED} <= exempt

steam_backlog_enforcer/tests/test_main_dispatch_extra.py

@@ -47,6 +47,19 @@ class TestMainDispatchServe:
         mock_cmd.assert_called_once_with(["--port", "8123"])
 
 
+class TestMainDispatchUnblock:
+    def test_dispatches_unblock_with_minutes(self) -> None:
+        argv = ["prog", "unblock", "10"]
+        with (
+            patch.object(sys, "argv", argv),
+            patch(f"{PKG}.Config.load", return_value=Config(steam_api_key="k")),
+            patch(f"{PKG}.State.load", return_value=State()),
+            patch(f"{PKG}.cmd_unblock") as mock_cmd,
+        ):
+            main()
+        assert mock_cmd.call_args.args[2] == ["10"]
+
+
 class TestMainDispatchPickManual:
     def test_dispatches_pick_manual(self) -> None:
         argv = ["prog", "pick-manual", "489830"]

steam_backlog_enforcer/tests/test_main_misc.py

@@ -1,7 +1,11 @@
 """Tests for the main CLI: store, reset, setup, exception and block-gaming commands."""
 
+from datetime import UTC, datetime
 from unittest.mock import patch
 
+import pytest
+
+from steam_backlog_enforcer._store_window import DEFAULT_WINDOW_MINUTES
 from steam_backlog_enforcer.config import Config, State
 from steam_backlog_enforcer.main import (
     cmd_buy_dlc,
@@ -11,76 +15,59 @@ from steam_backlog_enforcer.main import (
 )
 
 PKG = "steam_backlog_enforcer.main.misc"
+_UNTIL = datetime(2026, 10, 3, 12, 0, tzinfo=UTC)
+_FIVE_MINUTES = 5
 
 
 class TestCmdUnblock:
-    """Tests for cmd_unblock."""
+    """Tests for cmd_unblock: it opens a timed store window."""
 
-    def test_success(self) -> None:
+    def test_default_window(self) -> None:
         with (
-            patch(f"{PKG}.unblock_store", return_value=True),
-            patch(f"{PKG}._echo"),
+            patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open,
+            patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_unblock(Config(), State())
+        assert mock_open.call_args.args[1] == DEFAULT_WINDOW_MINUTES
+        assert any("UNBLOCKED until" in str(c) for c in mock_echo.call_args_list)
 
-    def test_fail(self) -> None:
+    def test_explicit_minutes(self) -> None:
+        with (
+            patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open,
+            patch(f"{PKG}._echo"),
+        ):
+            cmd_unblock(Config(), State(), ["5"])
+        assert mock_open.call_args.args[1] == _FIVE_MINUTES
+
+    @pytest.mark.parametrize("raw", ["abc", "0", "31"])
+    def test_bad_minutes_exits(self, raw: str) -> None:
         with (
-            patch(f"{PKG}.unblock_store", return_value=False),
             patch(f"{PKG}._echo") as mock_echo,
+            pytest.raises(SystemExit),
+        ):
+            cmd_unblock(Config(), State(), [raw])
+        assert any("Usage" in str(c) for c in mock_echo.call_args_list)
+
+    def test_unblock_failure_exits(self) -> None:
+        with (
+            patch(f"{PKG}.open_store_window", side_effect=RuntimeError("sudo?")),
+            patch(f"{PKG}._echo") as mock_echo,
+            pytest.raises(SystemExit),
         ):
             cmd_unblock(Config(), State())
         assert any("Failed" in str(c) for c in mock_echo.call_args_list)
 
 
 class TestCmdBuyDlc:
-    """Tests for cmd_buy_dlc."""
+    """cmd_buy_dlc is the default-length window, assigned game or not."""
 
-    def test_no_game(self) -> None:
-        with patch(f"{PKG}._echo") as mock_echo:
+    def test_opens_default_window(self) -> None:
+        with (
+            patch(f"{PKG}.open_store_window", return_value=_UNTIL) as mock_open,
+            patch(f"{PKG}._echo"),
+        ):
             cmd_buy_dlc(Config(), State())
-            assert any("No game" in str(c) for c in mock_echo.call_args_list)
-
-    def test_unblock_fails(self) -> None:
-        state = State(current_app_id=1, current_game_name="G")
-        with (
-            patch(f"{PKG}.unblock_store", return_value=False),
-            patch(f"{PKG}._echo"),
-        ):
-            cmd_buy_dlc(Config(), state)
-
-    def test_success_reblock(self) -> None:
-        state = State(current_app_id=1, current_game_name="G")
-        config = Config(block_store=True)
-        with (
-            patch(f"{PKG}.unblock_store", return_value=True),
-            patch(f"{PKG}.block_store", return_value=True),
-            patch(f"{PKG}.restart_steam"),
-            patch(f"{PKG}._echo"),
-            patch("builtins.input", return_value=""),
-        ):
-            cmd_buy_dlc(config, state)
-
-    def test_reblock_fails(self) -> None:
-        state = State(current_app_id=1, current_game_name="G")
-        config = Config(block_store=True)
-        with (
-            patch(f"{PKG}.unblock_store", return_value=True),
-            patch(f"{PKG}.block_store", return_value=False),
-            patch(f"{PKG}._echo") as mock_echo,
-            patch("builtins.input", return_value=""),
-        ):
-            cmd_buy_dlc(config, state)
-            assert any("Warning" in str(c) for c in mock_echo.call_args_list)
-
-    def test_no_reblock(self) -> None:
-        state = State(current_app_id=1, current_game_name="G")
-        config = Config(block_store=False)
-        with (
-            patch(f"{PKG}.unblock_store", return_value=True),
-            patch(f"{PKG}._echo"),
-            patch("builtins.input", return_value=""),
-        ):
-            cmd_buy_dlc(config, state)
+        assert mock_open.call_args.args[1] == DEFAULT_WINDOW_MINUTES
 
 
 class TestCmdReset:

steam_backlog_enforcer/tests/test_store_window.py

@@ -0,0 +1,110 @@
+"""Tests for the timed store window (``unblock [minutes]`` + daemon tick)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _store_window
+from steam_backlog_enforcer._store_window import (
+    MAX_WINDOW_MINUTES,
+    open_store_window,
+    store_window_tick,
+)
+from steam_backlog_enforcer.config import Config, State
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+PKG = "steam_backlog_enforcer._store_window"
+
+
+@pytest.fixture(autouse=True)
+def _fresh_reblock_memory() -> Iterator[None]:
+    """Each test starts with a daemon that has re-blocked for nothing yet."""
+    with patch.object(_store_window, "_reblocked_for", set()):
+        yield
+
+
+def _state_ending_in(delta: timedelta) -> State:
+    return State(store_unblocked_until=(datetime.now(UTC) + delta).isoformat())
+
+
+class TestOpenStoreWindow:
+    @pytest.mark.parametrize("minutes", [0, MAX_WINDOW_MINUTES + 1])
+    def test_rejects_out_of_range(self, minutes: int) -> None:
+        state = State()
+        with (
+            patch(f"{PKG}.unblock_store") as mock_unblock,
+            pytest.raises(ValueError, match="between 1 and"),
+        ):
+            open_store_window(state, minutes)
+        mock_unblock.assert_not_called()
+        assert not state.store_unblocked_until
+
+    def test_saves_deadline_and_unblocks(self) -> None:
+        state = State()
+        before = datetime.now(UTC)
+        with patch(f"{PKG}.unblock_store", return_value=True) as mock_unblock:
+            until = open_store_window(state, 10)
+        mock_unblock.assert_called_once()
+        assert before + timedelta(minutes=10) <= until
+        assert until <= datetime.now(UTC) + timedelta(minutes=10)
+        assert State.load().store_unblocked_until == until.isoformat()
+
+    def test_failed_unblock_raises(self) -> None:
+        with (
+            patch(f"{PKG}.unblock_store", return_value=False),
+            pytest.raises(RuntimeError, match="could not unblock"),
+        ):
+            open_store_window(State(), 5)
+
+
+class TestStoreWindowTick:
+    def test_no_window_does_nothing(self) -> None:
+        with (
+            patch(f"{PKG}.hosts_blocks_store") as mock_check,
+            patch(f"{PKG}.block_store") as mock_block,
+        ):
+            store_window_tick(Config(), State())
+        mock_check.assert_not_called()
+        mock_block.assert_not_called()
+
+    def test_malformed_deadline_is_ignored(self) -> None:
+        with patch(f"{PKG}.block_store") as mock_block:
+            store_window_tick(Config(), State(store_unblocked_until="garbage"))
+        mock_block.assert_not_called()
+
+    def test_open_window_undoes_a_reblock(self) -> None:
+        state = _state_ending_in(timedelta(minutes=5))
+        with (
+            patch(f"{PKG}.hosts_blocks_store", return_value=True),
+            patch(f"{PKG}.unblock_store") as mock_unblock,
+        ):
+            store_window_tick(Config(), state)
+        mock_unblock.assert_called_once()
+
+    def test_open_window_leaves_open_store_alone(self) -> None:
+        state = _state_ending_in(timedelta(minutes=5))
+        with (
+            patch(f"{PKG}.hosts_blocks_store", return_value=False),
+            patch(f"{PKG}.unblock_store") as mock_unblock,
+        ):
+            store_window_tick(Config(), state)
+        mock_unblock.assert_not_called()
+
+    def test_expired_window_reblocks_exactly_once(self) -> None:
+        state = _state_ending_in(timedelta(seconds=-1))
+        with patch(f"{PKG}.block_store") as mock_block:
+            store_window_tick(Config(block_store=True), state)
+            store_window_tick(Config(block_store=True), state)
+        mock_block.assert_called_once()
+
+    def test_expired_window_respects_block_store_off(self) -> None:
+        state = _state_ending_in(timedelta(seconds=-1))
+        with patch(f"{PKG}.block_store") as mock_block:
+            store_window_tick(Config(block_store=False), state)
+        mock_block.assert_not_called()