Add Librarian-only research window to total block

kuhy 2026-11-11: game-design research needs Librarian (app 4197610)
actually played during the 6-day total block. New root command
total-block-research --until T opens ONE window per lock: <= 48 h,
never past the lock end, a rerun never extends it.

The command only validates and records (/var/lib/steam-backlog-
enforcer, root 0700; lock id + deadline + two latches). The daemon
opens and closes everything: Steam hosts names, the store block, the
total-block chain (IP cache kept) and the steam package, reinstalled
with only guard-lib's package-block hook masked for that one pacman
call (--hookdir, proved in a guest). Lock file, hook and end date are
never touched. Browser-game sites stay null-routed.

Only Librarian may run: loop-local allowed set, strict mode (no
whitelist exceptions, no zero-achievement games), launchers and
counted games killed; state.json is never read or written.

Expiry latches the record, then the ordinary tick re-applies hosts
(now reconciled each tick), chain, uninstall and purge; the normal
store block is restored once, unconditionally.

Failure modes: missing/corrupt/unparseable record = no window and a
corrupt one also refuses a new window (fail closed); a clock set
back cannot reopen a latched window.

Checked: 3290 tests at 100% coverage; whole-tree pylint 10/10 run in
three parts (the hook exceeds the 60 s cap, so SKIP=pylint here); vmbox
guest with real guardctl package-block and the real unit, window
opened, Librarian-only enforced, clock moved past the deadline and the
full block re-applied. Not verified on the host.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWWzuCEg1m87jpKyRrw

README.md

@@ -36,6 +36,7 @@ python -m python_pkg.steam_backlog_enforcer.main setup
 | `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                            |
+| `total-block-research --until T` | Root. One window per total block (at most 48 h): Steam and the store open, only Librarian may be played; the daemon re-applies the full block at `T` |
 | `installed` | List currently installed Steam games       |
 | `uninstall` | Interactively uninstall non-assigned games |
 | `setup`     | First-time configuration                   |

steam_backlog_enforcer/_cmd_total_block_research.py

@@ -0,0 +1,71 @@
+"""``total-block-research``: lend Steam for one bounded window, Librarian only.
+
+kuhy 2026-11-11: game-design research needs "Librarian: Tidy Up the Arcane
+Library" actually played during a 6-day total block. This command opens ONE
+window per lock, ending at ``--until`` (at most 48 h away, never past the
+lock's end, never extended by a rerun). It only validates and records the
+window; the enforcer daemon opens Steam within a pass and re-applies the full
+block at the deadline (:mod:`steam_backlog_enforcer._research_tick`).
+
+Root-only: the record sits in a root-only state directory. The lock and its
+end date are untouched, so this can never shorten or end a total block.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+import os
+
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer._echo import _echo
+
+_USAGE = (
+    "Usage: total-block-research --until <ISO time, local unless it has an offset>"
+    "\n  e.g. total-block-research --until 2026-11-13T00:00"
+    "\n  One window per total block, at most 48 h, Librarian only."
+)
+
+
+def _until_argument(args: list[str]) -> str | None:
+    """The value of ``--until X`` or ``--until=X``, or None for any other shape."""
+    match args:
+        case ["--until", value]:
+            return value
+        case [single] if single.startswith("--until="):
+            return single.partition("=")[2]
+        case _:
+            return None
+
+
+def cmd_total_block_research(args: list[str]) -> int:
+    """Open the research window.
+
+    Args:
+        args: Arguments after the command name: ``--until <time>``.
+
+    Returns:
+        Process exit code: 0 recorded, 1 refused or not root, 2 bad usage.
+    """
+    raw = _until_argument(args)
+    if raw is None:
+        _echo(_USAGE)
+        return 2
+    try:
+        until = window.parse_until(raw)
+    except ValueError as exc:
+        _echo(f"{exc}\n{_USAGE}")
+        return 2
+    if os.geteuid() != 0:
+        _echo("total-block-research requires root (the record is root-only).")
+        return 1
+    try:
+        record = window.open_research_window(until)
+    except window.ResearchRefusedError as exc:
+        _echo(f"Refused: {exc}")
+        return 1
+    ends = datetime.fromtimestamp(record.until).astimezone()
+    _echo(f"Research window recorded until {ends:%Y-%m-%d %H:%M %Z}.")
+    _echo(f"Only {window.LIBRARIAN_NAME} (AppID {window.LIBRARIAN_APP_ID}) may run.")
+    _echo("The enforcer opens Steam on its next pass and re-applies the full")
+    _echo("total block at the deadline. This window cannot be reopened or extended.")
+    return 0

steam_backlog_enforcer/_command_help.py

@@ -44,6 +44,7 @@ EXTRA_COMMAND_DESCRIPTIONS: Final[dict[str, str]] = {
     "enforce": "Run enforcer: block, uninstall, kill, hide (--demo for a 60s budget)",
     "gaming-unblock": "Force-release playtime bind mounts (root; recovery hatch)",
     "total-block-refresh": "Apply the current domain list to a live total block (root)",
+    "total-block-research": "One Librarian-only Steam window per total block (root)",
 }
 
 ALL_COMMAND_DESCRIPTIONS: Final[dict[str, str]] = (

steam_backlog_enforcer/_command_locks.py

@@ -32,12 +32,18 @@ _MANUAL_LOCK_EXEMPT_COMMANDS = frozenset(
         "gaming-reset",
         # Reconciles the block's domain list; cannot shorten or end a lock.
         "total-block-refresh",
+        # Records the one bounded research window (see _research_window); the
+        # lock itself and its end date stay untouched.
+        "total-block-research",
     }
 )
 
 # Commands that remain usable while a total gaming block is active. Far
 # stricter than _MANUAL_LOCK_EXEMPT_COMMANDS: no done/pick/reset/
-# add-exception - there is no in-app way to shorten a total block.
+# add-exception - nothing here shortens or ends a total block. The one bounded
+# opening is total-block-research: root-only, a single <= 48 h window per lock
+# that only Librarian may use, recorded in a root-only state directory; the
+# daemon re-applies the full block when it expires (the lock's end is fixed).
 #
 # 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
@@ -53,5 +59,6 @@ _TOTAL_BLOCK_EXEMPT_COMMANDS = frozenset(
         "gaming-unblock",
         "serve",
         "total-block-refresh",
+        "total-block-research",
     }
 )

steam_backlog_enforcer/_enforce_guards.py

@@ -13,6 +13,7 @@ from steam_backlog_enforcer._actions import allowed_games
 from steam_backlog_enforcer.enforcer import send_notification
 from steam_backlog_enforcer.game_uninstall import (
     get_installed_games,
+    is_infrastructure_app,
     is_protected_app,
     uninstall_game,
 )
@@ -36,23 +37,26 @@ def allowed_names(state: State) -> str:
     return ", ".join(names) if names else "your assigned game"
 
 
-def guard_installed_games(allowed: set[int]) -> int:
+def guard_installed_games(allowed: set[int], *, strict: bool = False) -> int:
     """Remove any unauthorized game manifests + files.  Runs every loop.
 
     Args:
         allowed: Every app id that may stay installed — the assignment plus
             any concurrent manual picks.
+        strict: Spare only runtimes besides *allowed*, not approved exceptions
+            or zero-achievement games (the research window).
 
     Returns number of games removed this pass.
     """
     if not allowed:
         return 0
+    spared = is_infrastructure_app if strict else is_protected_app
     installed = get_installed_games()
     count = 0
     for app_id, name in installed:
         if app_id in allowed:
             continue
-        if is_protected_app(app_id):
+        if spared(app_id):
             continue
 
         logger.warning(

steam_backlog_enforcer/_enforce_loop.py

@@ -24,11 +24,11 @@ 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._research_tick import total_block_pass
 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,
     is_total_block_active,
     total_block_needs_cleanup,
 )
@@ -107,9 +107,10 @@ def _enforce_loop_iteration(
 
     # Total block takes priority over the assigned-game enforcement below -
     # while active, don't fight ourselves (e.g. installing the assigned
-    # game while total-block tries to keep Steam uninstalled).
+    # game while total-block tries to keep Steam uninstalled). An open
+    # research window is decided inside total_block_pass.
     if is_total_block_active():
-        enforce_total_block_tick()
+        total_block_pass(config)
         return
 
     if total_block_needs_cleanup():

steam_backlog_enforcer/_pacman.py

@@ -1,6 +1,6 @@
 """Thin, reusable wrappers around the pacman package manager.
 
-These three operations (query owner, query installed, remove) carry no
+These operations (query owner, query installed, remove, install) carry no
 total-block policy of their own -- they are the generic primitives the
 total-block purge is written in terms of, which is why they live here
 rather than inside it.
@@ -9,8 +9,14 @@ rather than inside it.
 from __future__ import annotations
 
 import logging
+from pathlib import Path
 import shutil
 import subprocess
+import tempfile
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +29,14 @@ _SUDO = shutil.which("sudo") or "/usr/bin/sudo"
 PACMAN = shutil.which("pacman.orig") or shutil.which("pacman") or "/usr/bin/pacman.orig"
 
 
+# pacman's own default hook directories, spelled out because a --hookdir on the
+# command line REPLACES them (proved in a throwaway Arch guest, pacman 7.1).
+# Later directories win over earlier ones for a hook of the same name, so a
+# /dev/null symlink in a directory listed last switches that one hook off.
+_HOOK_DIRS = ("/usr/share/libalpm/hooks", "/etc/pacman.d/hooks")
+_INSTALL_TIMEOUT_SECONDS = 900
+
+
 def pacman_owner(path: str) -> str | None:
     """Return the pacman package name that owns *path*, or None."""
     result = subprocess.run(
@@ -76,3 +90,47 @@ def uninstall_package(package: str) -> bool:
         result.stderr[-500:] if result.stderr else "",
     )
     return False
+
+
+def install_package(package: str, *, masked_hooks: Sequence[str] = ()) -> bool:
+    """Install *package* via pacman, with the named hooks off for this call only.
+
+    The mask lives in a temporary hook directory, so the masked hook file is
+    never edited, moved or removed: the very next pacman call sees it exactly as
+    before. Every other hook (guard-lib unlock/relock, the playtime mount
+    release, ...) still runs.
+
+    Args:
+        package: Package to install (``--needed``: a present one is a no-op).
+        masked_hooks: Hook file names to disable for this one transaction.
+
+    Returns:
+        True when pacman exited 0; False on any failure, which is logged.
+    """
+    with tempfile.TemporaryDirectory(prefix="pacman-mask-") as mask_dir:
+        hookdirs: list[str] = []
+        if masked_hooks:
+            for name in masked_hooks:
+                (Path(mask_dir) / name).symlink_to("/dev/null")
+            dirs = (*_HOOK_DIRS, mask_dir)
+            hookdirs = [arg for d in dirs for arg in ("--hookdir", d)]
+        try:
+            result = subprocess.run(
+                [_SUDO, PACMAN, "-S", "--needed", "--noconfirm", *hookdirs, package],
+                capture_output=True,
+                text=True,
+                timeout=_INSTALL_TIMEOUT_SECONDS,
+                check=False,
+            )
+        except OSError, subprocess.SubprocessError:
+            logger.exception("Failed to run pacman -S %s", package)
+            return False
+    if result.returncode == 0:
+        return True
+    logger.error(
+        "pacman -S %s failed (rc=%d): %s",
+        package,
+        result.returncode,
+        result.stderr[-500:] if result.stderr else "",
+    )
+    return False

steam_backlog_enforcer/_research_net.py

@@ -0,0 +1,108 @@
+"""Making the network and the Steam package match an open research window.
+
+Every pass of an open window calls :func:`reconcile_open_window`, which is
+idempotent and cheap once converged (one /etc/hosts read, one ``iptables -S``,
+one ``pacman -Qi``). The daemon is the only actor: the command just records the
+window, so a daemon that still runs the old code, or a restart half way, cannot
+leave two writers fighting over hosts, the firewall and pacman.
+
+The opening is deliberately narrower than "the block is off":
+
+* /etc/hosts keeps the browser-game sites null-routed; only Steam names open
+  (:data:`steam_backlog_enforcer._total_block_domains.RESEARCH_BLOCKED_DOMAINS`);
+* the normal Steam-store block (hosts lines + its iptables chain) is lifted for
+  checkout, and re-lifted whenever the hourly hosts reinstall puts it back;
+* the total-block iptables chain is taken down, its IP cache kept for expiry;
+* Steam is reinstalled if the block removed it, with only the guard-lib
+  package-block pacman hook masked for that one transaction. The hook file, the
+  lock file and the lock's end date are never touched.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+import time
+from typing import TYPE_CHECKING, Final
+
+from steam_backlog_enforcer._pacman import install_package
+from steam_backlog_enforcer._total_block import _PACKAGE_BLOCK_NAME
+from steam_backlog_enforcer._total_block_hosts import refresh_total_block_hosts
+from steam_backlog_enforcer._total_block_iptables import (
+    release_total_block_iptables,
+    total_block_chain_exists,
+)
+from steam_backlog_enforcer._total_block_purge import STEAM_PACKAGE, is_steam_installed
+from steam_backlog_enforcer.store_blocker import (
+    flush_dns_cache,
+    hosts_blocks_store,
+    unblock_store,
+)
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer._research_window import ResearchRecord
+
+logger = logging.getLogger(__name__)
+
+DB_LOCK_FILE: Final = Path("/var/lib/pacman/db.lck")
+RETRY_AFTER_SECONDS: Final = 60.0
+
+# Next monotonic time each retried action ("steam", "store") may run again.
+RETRY: Final[dict[str, float]] = {}
+# Windows (by opened_at) whose one-time release this daemon process has done.
+RELEASED: Final[set[float]] = set()
+
+
+def monotonic() -> float:
+    """Monotonic seconds (a seam for tests)."""
+    return time.monotonic()
+
+
+def attempt_due(key: str) -> bool:
+    """Whether the action *key* may run now (it is not in its backoff)."""
+    return monotonic() >= RETRY.get(key, 0.0)
+
+
+def note_failure(key: str) -> None:
+    """Hold the action *key* back for :data:`RETRY_AFTER_SECONDS`."""
+    RETRY[key] = monotonic() + RETRY_AFTER_SECONDS
+
+
+def block_hook_name() -> str:
+    """File name of guard-lib's pacman hook for the steam package-block."""
+    return f"guard-lib-block-{_PACKAGE_BLOCK_NAME}.hook"
+
+
+def ensure_steam_installed() -> bool:
+    """Reinstall Steam if the total block removed it.
+
+    Returns:
+        True when the package is installed afterwards. A running pacman
+        transaction is not raced and a failure backs off, so a broken mirror
+        cannot turn the 3 s tick into an install loop.
+    """
+    if is_steam_installed():
+        return True
+    if DB_LOCK_FILE.exists() or not attempt_due("steam"):
+        return False
+    if install_package(STEAM_PACKAGE, masked_hooks=(block_hook_name(),)):
+        logger.info("Research window: Steam reinstalled")
+        return True
+    note_failure("steam")
+    return False
+
+
+def reconcile_open_window(record: ResearchRecord) -> None:
+    """One idempotent pass making the machine match the open window *record*."""
+    first = record.opened_at not in RELEASED
+    hosts_ok = refresh_total_block_hosts(steam_open=True)
+    if first:
+        flush_dns_cache()
+    store_ok = True
+    if first or hosts_blocks_store():
+        store_ok = unblock_store()
+    if total_block_chain_exists():
+        release_total_block_iptables()
+    if first and hosts_ok and store_ok:
+        RELEASED.add(record.opened_at)
+    ensure_steam_installed()

steam_backlog_enforcer/_research_record.py

@@ -0,0 +1,136 @@
+"""The research record: strict parse, atomic root-only storage.
+
+One JSON file in ``/var/lib/steam-backlog-enforcer`` (the unit's
+``StateDirectory``: root only, 0700), not in the user-owned config directory,
+so the unprivileged user can neither forge a window nor delete the used-up
+marker. It holds numbers and two latches only; no free text. Anything that
+does not parse exactly, or that no legitimate open could have produced, is an
+error: the caller treats that as "no window" (fail closed).
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+import json
+import math
+from pathlib import Path
+from typing import Final
+
+from steam_backlog_enforcer.config import atomic_write
+
+MAX_WINDOW_SECONDS: Final = 48 * 3600.0
+RESEARCH_FILE: Final = Path("/var/lib/steam-backlog-enforcer/total_block_research.json")
+
+_NUMBER_KEYS: Final = ("lock_started_at", "lock_until", "opened_at", "until")
+_FLAG_KEYS: Final = ("closed", "store_reblocked")
+
+
+class ResearchStateError(ValueError):
+    """The record exists but cannot be trusted."""
+
+
+@dataclass(frozen=True)
+class ResearchRecord:
+    """One research window, bound to the total-block lock it was spent on."""
+
+    lock_started_at: float
+    lock_until: float
+    opened_at: float
+    until: float
+    closed: bool = False
+    """Latched once the deadline passed: a clock set back cannot reopen it."""
+    store_reblocked: bool = False
+    """Latched once the normal store block is back after the window."""
+
+    def is_open(self, at: float) -> bool:
+        """Whether the window is open at *at* (epoch seconds)."""
+        return not self.closed and self.opened_at <= at < self.until
+
+    def to_json(self) -> str:
+        """The record as the JSON text stored on disk."""
+        return json.dumps(asdict(self), indent=2) + "\n"
+
+
+def _number(data: dict[str, object], key: str) -> float:
+    """A finite real number from *data*; booleans and strings are refused."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"{key} is not a number"
+        raise ResearchStateError(msg)
+    try:
+        number = float(value)
+    except OverflowError as exc:
+        msg = f"{key} is out of range"
+        raise ResearchStateError(msg) from exc
+    if not math.isfinite(number):
+        msg = f"{key} is not finite"
+        raise ResearchStateError(msg)
+    return number
+
+
+def _flag(data: dict[str, object], key: str) -> bool:
+    """A real boolean from *data*, False when absent."""
+    value = data.get(key, False)
+    if not isinstance(value, bool):
+        msg = f"{key} is not a boolean"
+        raise ResearchStateError(msg)
+    return value
+
+
+def parse_record(raw: str) -> ResearchRecord:
+    """Parse and validate the stored text; there is no repair.
+
+    Raises:
+        ResearchStateError: If it is not JSON, has a missing or ill-typed
+            field, or breaks the policy (see :func:`_check_policy`).
+    """
+    try:
+        data = json.loads(raw)
+    except ValueError as exc:
+        msg = "not JSON"
+        raise ResearchStateError(msg) from exc
+    if not isinstance(data, dict):
+        msg = "not a JSON object"
+        raise ResearchStateError(msg)
+    numbers = {key: _number(data, key) for key in _NUMBER_KEYS}
+    flags = {key: _flag(data, key) for key in _FLAG_KEYS}
+    record = ResearchRecord(**numbers, **flags)
+    _check_policy(record)
+    return record
+
+
+def _check_policy(record: ResearchRecord) -> None:
+    """Refuse a record no legitimate open could have produced."""
+    if not record.lock_started_at < record.lock_until:
+        msg = "lock ends before it starts"
+    elif not record.opened_at < record.until:
+        msg = "window ends before it opens"
+    elif record.until - record.opened_at > MAX_WINDOW_SECONDS:
+        msg = "window is longer than 48 h"
+    elif record.until > record.lock_until:
+        msg = "window outlives its lock"
+    else:
+        return
+    raise ResearchStateError(msg)
+
+
+def read_record() -> ResearchRecord | None:
+    """The stored record, or ``None`` when no window was ever opened.
+
+    Raises:
+        ResearchStateError: If the file exists but is unreadable or invalid.
+    """
+    try:
+        raw = RESEARCH_FILE.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except (OSError, UnicodeDecodeError) as exc:
+        msg = f"cannot read {RESEARCH_FILE}: {exc}"
+        raise ResearchStateError(msg) from exc
+    return parse_record(raw)
+
+
+def write_record(record: ResearchRecord) -> None:
+    """Store *record* atomically, root only (0600, directory 0700)."""
+    RESEARCH_FILE.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+    atomic_write(RESEARCH_FILE, record.to_json(), mode=0o600)

steam_backlog_enforcer/_research_tick.py

@@ -0,0 +1,109 @@
+"""One daemon pass under a total block: the research window or the full block.
+
+The enforce loop hands its whole total-block branch here. With a research
+window open (see :mod:`steam_backlog_enforcer._research_window`) Steam is left
+installed and reachable, and the *only* thing kept is the one-allowed-game
+rule, bent to a single game: the loop-local allowed set is ``{Librarian}``. The
+real assignment in ``state.json`` is never read or written, so there is
+nothing to restore afterwards. Everything else the block does to Steam itself
+(kill, uninstall, purge) is skipped while the window is open.
+
+Once the deadline passes the record is latched closed (a clock set back cannot
+reopen it) and every pass is the ordinary :func:`enforce_total_block_tick`
+again: it kills Steam, removes the package, purges its files, rebuilds the
+firewall chain from the kept IP cache, and puts the full hosts section back.
+The lock's end date is never involved.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+import logging
+from typing import TYPE_CHECKING
+
+from leisure_watch.procscan import LAUNCHER_PROCESS_NAMES, kill_target_names
+
+from steam_backlog_enforcer import _research_net as net
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer import config as config_module
+from steam_backlog_enforcer._enforce_guards import guard_installed_games
+from steam_backlog_enforcer._research_net import reconcile_open_window
+from steam_backlog_enforcer._steam_launch import steam_is_installed
+from steam_backlog_enforcer._total_block import enforce_total_block_tick
+from steam_backlog_enforcer._total_block_purge import kill_and_uninstall_launchers
+from steam_backlog_enforcer.enforcer import (
+    enforce_allowed_game,
+    kill_processes_by_name,
+    send_notification,
+)
+from steam_backlog_enforcer.store_blocker import restore_store_block
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer._research_record import ResearchRecord
+    from steam_backlog_enforcer.config import Config
+
+logger = logging.getLogger(__name__)
+
+
+def _latch_closed(record: ResearchRecord) -> ResearchRecord:
+    """Persist that the window is over; blocking does not wait for the write."""
+    closed = replace(record, closed=True)
+    try:
+        window.write_record(closed)
+    except OSError:
+        logger.exception("Could not latch the research window closed")
+    return closed
+
+
+def _reblock_store(config: Config, record: ResearchRecord) -> None:
+    """Put the normal store block back once, after the window ended.
+
+    The total block's own hosts section covers the store until the lock ends,
+    but ``end_total_block_cleanup`` does not restore the normal block, so
+    without this the store would stay open after the lock. Unconditional
+    (:func:`restore_store_block`): the section being back makes the store look
+    blocked already.
+    """
+    if not net.attempt_due("store"):
+        return
+    if config.block_store and not restore_store_block():
+        net.note_failure("store")
+        return
+    try:
+        window.write_record(replace(record, store_reblocked=True))
+    except OSError:
+        logger.exception("Could not record that the store block is back")
+
+
+def _research_pass(record: ResearchRecord) -> None:
+    """Keep Steam open and Librarian the only game while the window is open."""
+    reconcile_open_window(record)
+    # Non-Steam launchers and counted games are still forbidden; Steam itself
+    # is not killed, uninstalled or purged (that is the block's tick).
+    kill_and_uninstall_launchers(LAUNCHER_PROCESS_NAMES)
+    kill_processes_by_name(kill_target_names(config_module.CONFIG_FILE))
+    if not steam_is_installed():
+        return
+    allowed = {window.LIBRARIAN_APP_ID}
+    killed = enforce_allowed_game(allowed, kill_unauthorized=True, strict=True)
+    for _pid, app_id in killed:
+        send_notification("⛔🎮", f"{app_id} → 🎯 {window.LIBRARIAN_NAME}")
+    guard_installed_games(allowed, strict=True)
+
+
+def total_block_pass(config: Config) -> None:
+    """The total-block branch of one enforce pass.
+
+    Args:
+        config: Loaded configuration (``block_store`` gates the store re-block).
+    """
+    at = window.now()
+    record = window.live_record(at)
+    if record is not None and record.is_open(at):
+        _research_pass(record)
+        return
+    if record is not None and not record.closed and at >= record.until:
+        record = _latch_closed(record)
+    enforce_total_block_tick()
+    if record is not None and record.closed and not record.store_reblocked:
+        _reblock_store(config, record)

steam_backlog_enforcer/_research_window.py

@@ -0,0 +1,165 @@
+"""The total-block research window: validity, lock binding and how it opens.
+
+kuhy 2026-11-11: to research a game by playing it ("Librarian: Tidy Up the
+Arcane Library", app 4197610) he needs Steam during a total block. This module
+is the one bounded way to get it, and the rules are all here:
+
+* **At most 48 hours**, as an explicit expiry (``--until``), never an
+  open-ended switch; a rerun never extends it.
+* **Once per lock**: the record names the lock it was spent on (that lock's
+  ``started_at``/``until``) and stays after the window ends.
+* **Fail closed**: a missing, unreadable, unparsable or out-of-policy record
+  (:mod:`steam_backlog_enforcer._research_record`) is "no window", i.e. the full
+  total block. A corrupt record also refuses a new window instead of resetting
+  the once-per-lock memory.
+* **Never shortens the lock**: nothing here touches the guard-lib lock file or
+  its pacman hook. The window ends by the clock, then the daemon re-applies the
+  whole block (see :mod:`steam_backlog_enforcer._research_tick`).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+import logging
+import time
+from typing import Final
+
+from steam_backlog_enforcer import _research_record as store
+from steam_backlog_enforcer._research_record import (
+    MAX_WINDOW_SECONDS,
+    ResearchRecord,
+    ResearchStateError,
+    read_record,
+    write_record,
+)
+from steam_backlog_enforcer._total_block import get_total_block_status
+
+logger = logging.getLogger(__name__)
+
+LIBRARIAN_APP_ID: Final = 4197610
+LIBRARIAN_NAME: Final = "Librarian: Tidy Up the Arcane Library!"
+
+# Same-lock comparison tolerance: the lock file holds whole seconds, but the
+# status goes through a datetime and back.
+_LOCK_TOLERANCE_SECONDS: Final = 1.0
+
+# Corrupt-record reports already made; cleared by the next healthy read so a
+# record that breaks again is reported again, but not every 3 s tick.
+LOGGED: Final[set[str]] = set()
+
+
+class ResearchRefusedError(Exception):
+    """A window may not be opened; the message says why."""
+
+
+def now() -> float:
+    """The wall clock in epoch seconds (a seam for tests)."""
+    return time.time()
+
+
+def current_lock(at: float) -> tuple[float, float] | None:
+    """``(started_at, until)`` of the total block in force at *at*, else ``None``."""
+    status = get_total_block_status()
+    if status.started_at is None or status.until is None:
+        return None
+    until = status.until.timestamp()
+    if at >= until:
+        return None
+    return status.started_at.timestamp(), until
+
+
+def _same_lock(record: ResearchRecord, lock: tuple[float, float]) -> bool:
+    """Whether *record* was spent on the lock identified by *lock*."""
+    return (
+        abs(record.lock_started_at - lock[0]) < _LOCK_TOLERANCE_SECONDS
+        and abs(record.lock_until - lock[1]) < _LOCK_TOLERANCE_SECONDS
+    )
+
+
+def live_record(at: float | None = None) -> ResearchRecord | None:
+    """The record for the lock in force, failing closed to ``None``.
+
+    ``None`` covers: no lock, no record, a record for an earlier lock, and a
+    record that cannot be trusted (reported once, not every tick).
+    """
+    when = now() if at is None else at
+    lock = current_lock(when)
+    if lock is None:
+        return None
+    try:
+        record = read_record()
+    except ResearchStateError as exc:
+        message = f"research record ignored, full block stays: {exc}"
+        if message not in LOGGED:
+            LOGGED.add(message)
+            logger.exception(message)
+        return None
+    LOGGED.clear()
+    if record is None or not _same_lock(record, lock):
+        return None
+    return record
+
+
+def parse_until(raw: str) -> datetime:
+    """Parse ``--until``: ISO 8601, local time unless it carries an offset.
+
+    Raises:
+        ValueError: If *raw* is not an ISO date-time.
+    """
+    try:
+        parsed = datetime.fromisoformat(raw)
+    except ValueError as exc:
+        msg = f"--until {raw!r} is not an ISO date-time (e.g. 2026-11-13T00:00)"
+        raise ValueError(msg) from exc
+    return parsed.astimezone() if parsed.tzinfo is None else parsed
+
+
+def open_research_window(until: datetime) -> ResearchRecord:
+    """Record a research window ending at *until* for the lock in force.
+
+    Args:
+        until: The expiry (timezone-aware).
+
+    Returns:
+        The stored record.
+
+    Raises:
+        ResearchRefusedError: With the reason: no active total block, an
+            unreadable record (fail closed), a window already spent on this
+            lock, an expiry that is past, over 48 h away or beyond the lock's
+            end, or a record that cannot be written.
+    """
+    at = now()
+    lock = current_lock(at)
+    if lock is None:
+        msg = "no active total block; nothing to research"
+        raise ResearchRefusedError(msg)
+    try:
+        previous = read_record()
+    except ResearchStateError as exc:
+        msg = f"research record unreadable ({exc}); refusing (fail closed)"
+        raise ResearchRefusedError(msg) from exc
+    if previous is not None and _same_lock(previous, lock):
+        ended = datetime.fromtimestamp(previous.until).astimezone()
+        msg = f"a window was already used for this lock (until {ended:%F %H:%M})"
+        raise ResearchRefusedError(msg)
+    deadline = until.timestamp()
+    if deadline <= at:
+        msg = "--until must be in the future"
+    elif deadline - at > MAX_WINDOW_SECONDS:
+        msg = "--until is more than 48 h from now"
+    elif deadline > lock[1]:
+        msg = "--until is after the end of the total block"
+    else:
+        return _store(ResearchRecord(lock[0], lock[1], at, deadline))
+    raise ResearchRefusedError(msg)
+
+
+def _store(record: ResearchRecord) -> ResearchRecord:
+    """Write *record*, turning a write failure into a refusal."""
+    try:
+        write_record(record)
+    except OSError as exc:
+        msg = f"cannot write {store.RESEARCH_FILE}: {exc}"
+        raise ResearchRefusedError(msg) from exc
+    return record

steam_backlog_enforcer/_total_block.py

@@ -2,17 +2,15 @@
 
 This module owns the *state* of a total block -- the guard-lib-protected lock
 file and the status derived from it -- and orchestrates the two independent
-mechanisms that enforce it:
+mechanisms that enforce it: :mod:`steam_backlog_enforcer._total_block_purge`
+(Steam not installed) and :mod:`steam_backlog_enforcer._total_block_hosts` plus
+:mod:`steam_backlog_enforcer._total_block_iptables` (Steam unreachable).
 
-* :mod:`steam_backlog_enforcer._total_block_purge` makes Steam not installed;
-* :mod:`steam_backlog_enforcer._total_block_hosts` and
-  :mod:`steam_backlog_enforcer._total_block_iptables` make it unreachable.
-
-There is no in-app command to lift a block early. Tamper-resistance comes
-from guard-lib (~/guard-lib): the lock file's ``until`` timestamp is
-protected by a bind-mounted, chattr-immutable file-guard instance, and pacman
-itself refuses to reinstall/upgrade the ``steam`` package while the lock is
-active (package-block).
+No in-app command lifts a block early; the one bounded exception is a single
+<= 48 h Librarian-only window per lock (:mod:`._research_window`) that never
+touches the lock or its end date. Tamper-resistance is guard-lib's: the lock
+file's ``until`` sits in a chattr-immutable file-guard, and pacman refuses to
+install ``steam`` while the lock is active (package-block).
 """
 
 from __future__ import annotations
@@ -29,6 +27,7 @@ from leisure_watch.procscan import LAUNCHER_PROCESS_NAMES, kill_target_names
 from steam_backlog_enforcer import config
 from steam_backlog_enforcer._total_block_hosts import (
     apply_total_block_hosts,
+    refresh_total_block_hosts,
     remove_total_block_hosts,
 )
 from steam_backlog_enforcer._total_block_iptables import (
@@ -218,7 +217,7 @@ def enforce_total_block_tick() -> None:
     purge_steam_and_proton()
 
     apply_total_block_iptables()
-    apply_total_block_hosts()
+    refresh_total_block_hosts()  # not apply: that keeps a window's open section
 
 
 def end_total_block_cleanup() -> None:

steam_backlog_enforcer/_total_block_domains.py

@@ -80,6 +80,15 @@ _ALL_TOTAL_BLOCK_DOMAINS = [
     if domain not in TOTAL_BLOCK_EXEMPT_DOMAINS
 ]
 
+# What /etc/hosts keeps null-routed while a research window has Steam open
+# (``total-block-research``): the browser-game sites only. Steam is the one
+# thing the window lends out, and Librarian is the one game it may be used for.
+RESEARCH_BLOCKED_DOMAINS = [
+    domain
+    for domain in _expand_with_www(GAME_WEBSITE_DOMAINS)
+    if domain not in TOTAL_BLOCK_EXEMPT_DOMAINS
+]
+
 # The null-route redirect target used to make a blocked domain resolve
 # nowhere - built from parts rather than the literal so linters don't
 # mistake it for a socket bind-all-interfaces address (it never is one).

steam_backlog_enforcer/_total_block_hosts.py

@@ -19,6 +19,7 @@ from steam_backlog_enforcer._hosts_protection import (
 from steam_backlog_enforcer._total_block_domains import (
     _ALL_TOTAL_BLOCK_DOMAINS,
     NULL_ROUTE_IP,
+    RESEARCH_BLOCKED_DOMAINS,
 )
 from steam_backlog_enforcer.config import HOSTS_FILE
 
@@ -28,10 +29,14 @@ _HOSTS_BLOCK_BEGIN = "# BEGIN steam-backlog-enforcer total-block\n"
 _HOSTS_BLOCK_END = "# END steam-backlog-enforcer total-block\n"
 
 
-def _expected_block() -> str:
-    """The enforcer's /etc/hosts section as it should currently read."""
+def _expected_block(*, steam_open: bool = False) -> str:
+    """The enforcer's /etc/hosts section as it should currently read.
+
+    With *steam_open* (a research window) the section lists the game sites only.
+    """
+    domains = RESEARCH_BLOCKED_DOMAINS if steam_open else _ALL_TOTAL_BLOCK_DOMAINS
     lines = [_HOSTS_BLOCK_BEGIN]
-    lines += [f"{NULL_ROUTE_IP} {domain}\n" for domain in _ALL_TOTAL_BLOCK_DOMAINS]
+    lines += [f"{NULL_ROUTE_IP} {domain}\n" for domain in domains]
     lines.append(_HOSTS_BLOCK_END)
     return "".join(lines)
 
@@ -61,14 +66,32 @@ def apply_total_block_hosts() -> bool:
     return True
 
 
-def refresh_total_block_hosts() -> bool:
+def _write_hosts(content: str, failure: str) -> bool:
+    """Write *content* through the guard-lib unlock/relock; False on failure."""
+    try:
+        _disable_hosts_protection()
+        _sudo_write_hosts(content)
+    except OSError, subprocess.SubprocessError:
+        logger.exception(failure)
+        return False
+    finally:
+        _enable_hosts_protection()
+    return True
+
+
+def refresh_total_block_hosts(*, steam_open: bool = False) -> bool:
     """Bring an existing section in line with the current domain list.
 
     Unlike :func:`apply_total_block_hosts` (which leaves a present section
     alone, so the 3 s tick stays cheap), this replaces the section when its
-    text differs -- how a list change reaches an already-active block. Only the
-    text between the markers changes; a missing section is installed; a
-    BEGIN without END, or an unreadable file, is left untouched (False).
+    text differs -- how a list change reaches an already-active block, and how
+    the full list returns when a research window ends. Only the text between
+    the markers changes; a BEGIN without END, or an unreadable file, is left
+    untouched (False). A missing section is installed (the full one, or with
+    *steam_open* the research one).
+
+    Args:
+        steam_open: Write the research-window section (game sites only).
     """
     try:
         content = HOSTS_FILE.read_text(encoding="utf-8")
@@ -76,28 +99,25 @@ def refresh_total_block_hosts() -> bool:
         logger.exception("Failed to read /etc/hosts")
         return False
 
+    expected = _expected_block(steam_open=steam_open)
     start = content.find(_HOSTS_BLOCK_BEGIN)
     if start == -1:
-        return apply_total_block_hosts()
+        if not steam_open:
+            return apply_total_block_hosts()
+        head = content if content.endswith("\n") else content + "\n"
+        return _write_hosts(head + expected, "Failed to write research hosts entries")
     end_at = content.find(_HOSTS_BLOCK_END, start)
     if end_at == -1:
         logger.error("total-block section has no END marker; not touching hosts")
         return False
 
     end = end_at + len(_HOSTS_BLOCK_END)
-    expected = _expected_block()
     if content[start:end] == expected:
         return True
-
-    try:
-        _disable_hosts_protection()
-        _sudo_write_hosts(content[:start] + expected + content[end:])
-    except OSError, subprocess.SubprocessError:
-        logger.exception("Failed to refresh total-block hosts entries")
-        return False
-    finally:
-        _enable_hosts_protection()
-    return True
+    return _write_hosts(
+        content[:start] + expected + content[end:],
+        "Failed to refresh total-block hosts entries",
+    )
 
 
 def remove_total_block_hosts() -> bool:

steam_backlog_enforcer/_total_block_iptables.py

@@ -194,6 +194,31 @@ def refresh_total_block_iptables() -> bool:
     return True
 
 
+def total_block_chain_exists() -> bool:
+    """Whether the total-block chain is present (a single iptables call)."""
+    return _chain_ips() is not None
+
+
+def release_total_block_iptables() -> bool:
+    """Take the chain and its OUTPUT hook down but keep the IP cache.
+
+    How a research window opens the firewall. Unlike
+    :func:`remove_total_block_iptables` the cache stays, so when the window
+    ends :func:`apply_total_block_iptables` rebuilds the chain with every IP it
+    dropped before, not only the ones DNS happens to answer that day.
+
+    Returns:
+        True once the chain is verifiably gone; False when it is still there or
+        iptables cannot be run.
+    """
+    try:
+        drop_chain(IPTABLES_CHAIN)
+        return not total_block_chain_exists()
+    except OSError, subprocess.SubprocessError:
+        logger.exception("Failed to release total-block iptables rules")
+        return False
+
+
 def remove_total_block_iptables() -> bool:
     """Remove the total-block iptables chain and its OUTPUT hook."""
     try:

steam_backlog_enforcer/enforcer.py

@@ -20,7 +20,10 @@ from steam_backlog_enforcer._desktop_env import (
     resolve_desktop_user,
 )
 from steam_backlog_enforcer._store_tools import run_quiet
-from steam_backlog_enforcer.game_uninstall import is_protected_app
+from steam_backlog_enforcer.game_uninstall import (
+    is_infrastructure_app,
+    is_protected_app,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -59,6 +62,7 @@ def enforce_allowed_game(
     allowed_app_ids: set[int],
     *,
     kill_unauthorized: bool = True,
+    strict: bool = False,
 ) -> list[tuple[int, int]]:
     """Check running games; optionally kill unauthorized ones.
 
@@ -67,11 +71,14 @@ def enforce_allowed_game(
             assignment plus any concurrent manual picks. Empty disables the
             check entirely (nothing is assigned, so nothing is unauthorized).
         kill_unauthorized: Whether to kill violating processes.
+        strict: Spare only runtimes besides the allowed set, not approved
+            exceptions or zero-achievement games (the research window).
 
     Returns list of (pid, app_id) that were killed or detected.
     """
     if not allowed_app_ids:
         return []
+    spared = is_infrastructure_app if strict else is_protected_app
     running = get_running_steam_game_pids()
     violations: list[tuple[int, int]] = []
 
@@ -81,7 +88,7 @@ def enforce_allowed_game(
         # Skip Steam client itself (app_id 0 or very low IDs).
         if app_id == 0:
             continue
-        if is_protected_app(app_id):
+        if spared(app_id):
             continue
 
         violations.append((pid, app_id))

steam_backlog_enforcer/game_uninstall.py

@@ -170,6 +170,16 @@ def uninstall_other_games(allowed_app_ids: set[int]) -> int:
     return count
 
 
+def is_infrastructure_app(app_id: int) -> bool:
+    """Return True if *app_id* is a Steam runtime or compatibility tool.
+
+    The strict subset of :func:`is_protected_app`: no exception, no zero-
+    achievement game. A research window uses it so that nothing but the one
+    allowed game survives besides the runtimes that game needs.
+    """
+    return app_id in PROTECTED_APP_IDS
+
+
 def is_protected_app(app_id: int) -> bool:
     """Return True if *app_id* must never be uninstalled.
 
@@ -186,7 +196,7 @@ def is_protected_app(app_id: int) -> bool:
         True if the app should be left alone by the enforcer.
     """
     return (
-        app_id in PROTECTED_APP_IDS
+        is_infrastructure_app(app_id)
         or app_id in get_approved_exception_ids()
         or app_id in zero_achievement_ids()
     )

steam_backlog_enforcer/main/__init__.py

@@ -24,6 +24,7 @@ from steam_backlog_enforcer._cmd_playtime import (
     cmd_gaming_unblock,
 )
 from steam_backlog_enforcer._cmd_total_block_refresh import cmd_total_block_refresh
+from steam_backlog_enforcer._cmd_total_block_research import cmd_total_block_research
 from steam_backlog_enforcer._command_help import (
     COMMAND_DESCRIPTIONS,
     EXTRA_COMMAND_DESCRIPTIONS,
@@ -185,6 +186,8 @@ def _dispatch_extra_command(command: str, config: Config, state: State) -> bool:
         sys.exit(cmd_gaming_unblock(args))
     if command == "total-block-refresh":
         sys.exit(cmd_total_block_refresh(args))
+    if command == "total-block-research":
+        sys.exit(cmd_total_block_research(args))
     handlers: dict[str, Callable[[], object]] = {
         "add-exception": lambda: cmd_add_exception(args),
         "unblock": lambda: cmd_unblock(config, state, args),

steam_backlog_enforcer/main/_shared.py

@@ -55,7 +55,8 @@ def _show_total_block_lock_message(status: TotalBlockStatus) -> None:
     _echo(
         "\nSteam has been uninstalled, all known game/launcher processes are"
         "\nbeing killed on sight, and Steam + game-website domains are blocked."
-        "\nThere is NO in-app command to lift this early."
+        "\nThere is NO in-app command to lift this early (except one bounded"
+        "\nLibrarian-only research window per block: total-block-research)."
         f"\n\nAllowed commands: {', '.join(sorted(_TOTAL_BLOCK_EXEMPT_COMMANDS))}"
     )
     _echo("=" * 60 + "\n")

steam_backlog_enforcer/store_blocker.py

@@ -93,6 +93,23 @@ def block_store() -> bool:
     return False
 
 
+def restore_store_block() -> bool:
+    """Re-enable the normal store block, even if something else also blocks it.
+
+    :func:`block_store` returns early at "already blocked", and the total
+    block's own /etc/hosts section null-routes the same domains: with that
+    section back, the normal lines would stay commented out after a research
+    window and the store would open the day the lock ends. This always
+    uncomments them and rebuilds the store chain.
+
+    Returns True if at least one layer was restored.
+    """
+    hosts_ok = _reblock_hosts()
+    ipt_ok = _block_store_iptables()
+    flush_dns_cache()
+    return hosts_ok or ipt_ok
+
+
 def _block_via_hosts_install() -> bool:
     """Run the hosts install.sh to apply /etc/hosts with Steam Store entries.
 

steam_backlog_enforcer/tests/_isolate_research.py

@@ -0,0 +1,40 @@
+"""Autouse isolation for the total-block research window.
+
+The record lives in ``/var/lib/steam-backlog-enforcer`` (root only) and the
+install step checks pacman's real ``db.lck``. A test that reached either would
+open or burn a window on the developer's machine, so both are redirected to
+``tmp_path`` for every test, and the module-level memory (corrupt-record log
+dedupe, install backoff, windows already released) starts empty each time.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _research_net, _research_window
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture(autouse=True)
+def _isolate_research(tmp_path: Path) -> Iterator[None]:
+    """Point the research record and pacman lock at tmp_path; reset memory."""
+    _research_window.LOGGED.clear()
+    _research_net.RETRY.clear()
+    _research_net.RELEASED.clear()
+    with (
+        patch(
+            "steam_backlog_enforcer._research_record.RESEARCH_FILE",
+            tmp_path / "research" / "total_block_research.json",
+        ),
+        patch(
+            "steam_backlog_enforcer._research_net.DB_LOCK_FILE",
+            tmp_path / "pacman" / "db.lck",
+        ),
+    ):
+        yield

steam_backlog_enforcer/tests/_research_fixtures.py

@@ -0,0 +1,80 @@
+"""Fixtures for the daemon-pass tests: every effect of one pass is a mock."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from steam_backlog_enforcer._research_tick import total_block_pass
+from steam_backlog_enforcer.config import Config
+from steam_backlog_enforcer.tests._research_helpers import NOW, lock
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+PKG = "steam_backlog_enforcer._research_tick"
+BLOCK = "steam_backlog_enforcer._total_block"
+NOW_PATCH = "steam_backlog_enforcer._research_window.now"
+
+
+@dataclass
+class Mocks:
+    """The collaborators of one pass, all stand-ins."""
+
+    block_tick: MagicMock
+    reconcile: MagicMock
+    launchers: MagicMock
+    counted: MagicMock
+    steam_installed: MagicMock
+    enforce: MagicMock
+    guard: MagicMock
+    notify: MagicMock
+    restore_store: MagicMock
+    steam_kill: MagicMock
+    steam_uninstall: MagicMock
+    steam_purge: MagicMock
+
+
+@pytest.fixture
+def tm(total_block_paths: Paths) -> Iterator[Mocks]:
+    """An active total block with every effect replaced by a mock."""
+    lock(total_block_paths)
+    with (
+        patch(f"{PKG}.enforce_total_block_tick") as block_tick,
+        patch(f"{PKG}.reconcile_open_window") as reconcile,
+        patch(f"{PKG}.kill_and_uninstall_launchers", return_value=[]) as launchers,
+        patch(f"{PKG}.kill_processes_by_name", return_value=[]) as counted,
+        patch(f"{PKG}.steam_is_installed", return_value=True) as installed,
+        patch(f"{PKG}.enforce_allowed_game", return_value=[]) as enforce,
+        patch(f"{PKG}.guard_installed_games", return_value=0) as guard,
+        patch(f"{PKG}.send_notification") as notify,
+        patch(f"{PKG}.restore_store_block", return_value=True) as restore_store,
+        patch(f"{BLOCK}.kill_steam_and_launchers") as steam_kill,
+        patch(f"{BLOCK}.uninstall_steam_package") as steam_uninstall,
+        patch(f"{BLOCK}.purge_steam_and_proton") as steam_purge,
+    ):
+        yield Mocks(
+            block_tick,
+            reconcile,
+            launchers,
+            counted,
+            installed,
+            enforce,
+            guard,
+            notify,
+            restore_store,
+            steam_kill,
+            steam_uninstall,
+            steam_purge,
+        )
+
+
+def run_pass(now: float = NOW, *, block_store: bool = True) -> None:
+    """Run one total-block pass with the clock at *now*."""
+    with patch(NOW_PATCH, return_value=now):
+        total_block_pass(Config(block_store=block_store))

steam_backlog_enforcer/tests/_research_helpers.py

@@ -0,0 +1,37 @@
+"""Shared fixtures-as-functions for the research-window tests.
+
+One total block (6 days, as the live one) and one record inside it, so each
+test only spells out the field it is about.
+"""
+
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import TYPE_CHECKING
+
+from steam_backlog_enforcer._research_record import ResearchRecord
+from steam_backlog_enforcer.tests._total_block_paths import write_lock
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+LOCK_START = 1_794_233_872.0
+LOCK_END = LOCK_START + 6 * 86400
+NOW = LOCK_START + 2 * 86400
+HOUR = 3600.0
+
+
+def lock(paths: Paths) -> None:
+    """Write the 6-day total-block lock the records below belong to."""
+    write_lock(paths, LOCK_START, LOCK_END, days=6)
+
+
+def make_record(**changes: float | bool) -> ResearchRecord:
+    """A valid record opened a minute before NOW and open for one more hour."""
+    base = ResearchRecord(
+        lock_started_at=LOCK_START,
+        lock_until=LOCK_END,
+        opened_at=NOW - 60,
+        until=NOW + HOUR,
+    )
+    return replace(base, **changes)

steam_backlog_enforcer/tests/conftest.py

@@ -29,6 +29,7 @@ from steam_backlog_enforcer.tests._isolate_incidents import _isolate_incidents
 from steam_backlog_enforcer.tests._isolate_leisure import _isolate_leisure
 from steam_backlog_enforcer.tests._isolate_playtime import _isolate_playtime
 from steam_backlog_enforcer.tests._isolate_reading import _isolate_reading
+from steam_backlog_enforcer.tests._isolate_research import _isolate_research
 from steam_backlog_enforcer.tests._isolate_shadow import _isolate_shadow
 from steam_backlog_enforcer.tests._no_fast_install import _no_fast_install
 from steam_backlog_enforcer.tests._no_live_focus_project import _no_live_focus_project
@@ -39,6 +40,7 @@ from steam_backlog_enforcer.tests._no_real_effects import (
 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
+from steam_backlog_enforcer.tests._research_fixtures import tm
 
 # Re-exported so ruff --fix does not delete the imports above: pytest
 # registers autouse fixtures by name, so they look unused to the linter.
@@ -51,6 +53,7 @@ __all__ = [
     "_isolate_leisure",
     "_isolate_playtime",
     "_isolate_reading",
+    "_isolate_research",
     "_isolate_shadow",
     "_no_fast_install",
     "_no_live_focus_project",
@@ -58,6 +61,7 @@ __all__ = [
     "_no_real_effects",
     "_no_systemd_run",
     "_no_workout_http",
+    "tm",
 ]
 
 from steam_backlog_enforcer.tests._total_block_paths import (

steam_backlog_enforcer/tests/test_cmd_total_block_research.py

@@ -0,0 +1,123 @@
+"""The `total-block-research` command: root-only, bounded, lock-exempt."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import sys
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _cmd_total_block_research as cmd
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer._command_gate import UNCONFIGURED_OK
+from steam_backlog_enforcer._command_help import ALL_COMMAND_DESCRIPTIONS
+from steam_backlog_enforcer._command_locks import (
+    _MANUAL_LOCK_EXEMPT_COMMANDS,
+    _TOTAL_BLOCK_EXEMPT_COMMANDS,
+)
+from steam_backlog_enforcer.config import Config, State
+from steam_backlog_enforcer.main import _dispatch_extra_command
+from steam_backlog_enforcer.tests._research_helpers import HOUR, NOW, lock
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+PKG = "steam_backlog_enforcer._cmd_total_block_research"
+NAME = "total-block-research"
+UNTIL = datetime.fromtimestamp(NOW + 5 * HOUR, tz=UTC).isoformat()
+
+
+@pytest.fixture(autouse=True)
+def _root_with_lock(total_block_paths: Paths) -> Iterator[None]:
+    """Every test runs as root inside an active total block, clock at NOW."""
+    lock(total_block_paths)
+    with (
+        patch(f"{PKG}.os.geteuid", return_value=0),
+        patch("steam_backlog_enforcer._research_window.now", return_value=NOW),
+    ):
+        yield
+
+
+class TestCommand:
+    """Argument handling and the refusals."""
+
+    def test_opens_the_window(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Test that a valid expiry records the window and exits 0."""
+        assert cmd.cmd_total_block_research(["--until", UNTIL]) == 0
+        record = window.read_record()
+        assert record is not None
+        assert record.until == NOW + 5 * HOUR
+        assert "Librarian" in capsys.readouterr().out
+
+    def test_equals_form(self) -> None:
+        """Test that --until=VALUE works too."""
+        assert cmd.cmd_total_block_research([f"--until={UNTIL}"]) == 0
+
+    @pytest.mark.parametrize(
+        "args",
+        [[], ["--until"], ["--until", "x", "y"], ["--soon"], ["--until", "nope"]],
+    )
+    def test_bad_usage_is_2_and_writes_nothing(self, args: list[str]) -> None:
+        """Test that malformed arguments are usage errors."""
+        assert cmd.cmd_total_block_research(args) == 2
+        assert window.read_record() is None
+
+    def test_requires_root(self) -> None:
+        """Test that non-root exits 1 before touching state."""
+        with patch(f"{PKG}.os.geteuid", return_value=1000):
+            assert cmd.cmd_total_block_research(["--until", UNTIL]) == 1
+        assert window.read_record() is None
+
+    def test_over_48_hours_is_refused(self, capsys: pytest.CaptureFixture[str]) -> None:
+        """Test that the cap is enforced by the command."""
+        late = datetime.fromtimestamp(NOW + 49 * HOUR, tz=UTC).isoformat()
+        assert cmd.cmd_total_block_research(["--until", late]) == 1
+        assert "48" in capsys.readouterr().out
+        assert window.read_record() is None
+
+    def test_second_run_is_refused(self) -> None:
+        """Test that the second run is refused and the first expiry stands."""
+        assert cmd.cmd_total_block_research(["--until", UNTIL]) == 0
+        first = window.read_record()
+        later = datetime.fromtimestamp(NOW + 30 * HOUR, tz=UTC).isoformat()
+        assert cmd.cmd_total_block_research(["--until", later]) == 1
+        assert window.read_record() == first
+
+    def test_no_active_block_is_refused(self, total_block_paths: Paths) -> None:
+        """Test that without a lock the command exits 1."""
+        total_block_paths.lock_file.unlink()
+        assert cmd.cmd_total_block_research(["--until", UNTIL]) == 1
+
+
+class TestRegistration:
+    """Reachable during the lock it loosens, and nowhere else."""
+
+    def test_lock_exempt(self) -> None:
+        """Test that the total block lets it through (and the nested manual set)."""
+        assert NAME in _TOTAL_BLOCK_EXEMPT_COMMANDS
+        assert _TOTAL_BLOCK_EXEMPT_COMMANDS <= _MANUAL_LOCK_EXEMPT_COMMANDS
+
+    def test_needs_a_configured_install(self) -> None:
+        """Test that a missing API key is a loud failure, not a silent pass."""
+        assert NAME not in UNCONFIGURED_OK
+
+    def test_has_help_text(self) -> None:
+        """Test that help lists it."""
+        assert NAME in ALL_COMMAND_DESCRIPTIONS
+
+    def test_dispatch_hands_over_the_argv_tail_and_exits(self) -> None:
+        """Test that main dispatch passes the tail and exits with its code."""
+        argv = ["prog", NAME, "--until", UNTIL]
+        with (
+            patch.object(sys, "argv", argv),
+            patch("steam_backlog_enforcer.main.cmd_total_block_research") as handler,
+            pytest.raises(SystemExit) as exc_info,
+        ):
+            handler.return_value = 7
+            _dispatch_extra_command(NAME, Config(), State())
+        assert exc_info.value.code == 7
+        handler.assert_called_once_with(["--until", UNTIL])

steam_backlog_enforcer/tests/test_enforce_loop_part3.py

@@ -25,13 +25,13 @@ class TestEnforceLoopIterationTotalBlock:
         state = State(current_app_id=1, current_game_name="G")
         with (
             patch(f"{PKG}.is_total_block_active", return_value=True),
-            patch(f"{PKG}.enforce_total_block_tick") as mock_tick,
+            patch(f"{PKG}.total_block_pass") as mock_tick,
             patch(f"{PKG}.enforce_allowed_game") as mock_enforce,
             patch(f"{PKG}.guard_installed_games") as mock_guard,
             patch(f"{ENFORCE_STEPS_PKG}.is_game_installed") as mock_installed,
         ):
             _enforce_loop_iteration(config, state, session=fake_session())
-        mock_tick.assert_called_once()
+        mock_tick.assert_called_once_with(config)
         mock_enforce.assert_not_called()
         mock_guard.assert_not_called()
         mock_installed.assert_not_called()
@@ -43,7 +43,7 @@ class TestEnforceLoopIterationTotalBlock:
         with (
             patch(f"{PKG}.is_total_block_active", return_value=False),
             patch(f"{PKG}.total_block_needs_cleanup", return_value=False),
-            patch(f"{PKG}.enforce_total_block_tick") as mock_tick,
+            patch(f"{PKG}.total_block_pass") as mock_tick,
             patch(f"{ENFORCE_STEPS_PKG}.is_game_installed", return_value=True),
         ):
             _enforce_loop_iteration(config, state, session=fake_session())
@@ -132,7 +132,7 @@ class TestPlaytimeIsEnforcedFirst:
         with (
             patch(f"{PKG}.playtime_tick") as mock_playtime,
             patch(f"{PKG}.is_total_block_active", return_value=True),
-            patch(f"{PKG}.enforce_total_block_tick"),
+            patch(f"{PKG}.total_block_pass"),
         ):
             _enforce_loop_iteration(config, state, session=fake_session())
         mock_playtime.assert_called_once()
@@ -144,7 +144,7 @@ class TestPlaytimeIsEnforcedFirst:
         with (
             patch(f"{PKG}.playtime_tick") as mock_playtime,
             patch(f"{PKG}.is_total_block_active", return_value=True),
-            patch(f"{PKG}.enforce_total_block_tick"),
+            patch(f"{PKG}.total_block_pass"),
         ):
             _enforce_loop_iteration(config, state, session=fake_session())
         assert mock_playtime.call_args.kwargs["interval"] == 3
@@ -157,7 +157,7 @@ class TestPlaytimeIsEnforcedFirst:
         with (
             patch(f"{PKG}.playtime_tick") as mock_playtime,
             patch(f"{PKG}.is_total_block_active", return_value=True),
-            patch(f"{PKG}.enforce_total_block_tick"),
+            patch(f"{PKG}.total_block_pass"),
         ):
             _enforce_loop_iteration(config, state, session=fake_session(), demo=True)
         assert mock_playtime.call_args.kwargs["demo"] is True

steam_backlog_enforcer/tests/test_main_shared.py

@@ -238,6 +238,7 @@ class TestEnforceTotalBlockLock:
         assert (
             frozenset({"status", "enforce", "gaming-status", "gaming-unblock", "serve"})
             | {"total-block-refresh"}  # reconciles the list; cannot shorten the lock
+            | {"total-block-research"}  # <= 48 h, once per lock, Librarian only
             == _TOTAL_BLOCK_EXEMPT_COMMANDS
         )
         assert "gaming-reset" not in _TOTAL_BLOCK_EXEMPT_COMMANDS

steam_backlog_enforcer/tests/test_pacman_install.py

@@ -0,0 +1,88 @@
+"""install_package: pacman -S with named hooks masked for that one call."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+import subprocess
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from steam_backlog_enforcer import _pacman
+
+PKG = "steam_backlog_enforcer._pacman"
+HOOK = "guard-lib-block-steam-block.hook"
+
+
+def _hookdirs(argv: list[str]) -> list[str]:
+    """The values of every --hookdir in *argv*, in order."""
+    return [argv[i + 1] for i, arg in enumerate(argv) if arg == "--hookdir"]
+
+
+class TestInstallPackage:
+    """The command line, the mask directory and the outcomes."""
+
+    def test_plain_install_without_masks(self) -> None:
+        """Test that no mask means no --hookdir at all (all hooks run)."""
+        run = MagicMock(return_value=MagicMock(returncode=0, stderr=""))
+        with patch(f"{PKG}.subprocess.run", run):
+            assert _pacman.install_package("steam") is True
+        argv = run.call_args.args[0]
+        assert argv[1:] == [_pacman.PACMAN, "-S", "--needed", "--noconfirm", "steam"]
+
+    def test_masked_hook_is_a_dev_null_symlink_in_the_last_hookdir(self) -> None:
+        """Test that the mask wins and every other hook directory is kept."""
+        seen: dict[str, object] = {}
+
+        def fake_run(argv: list[str], **_kw: object) -> MagicMock:
+            dirs = _hookdirs(argv)
+            mask = Path(dirs[-1])
+            seen["dirs"] = dirs
+            seen["entries"] = {p.name: p.readlink() for p in mask.iterdir()}
+            seen["mask"] = mask
+            return MagicMock(returncode=0, stderr="")
+
+        with patch(f"{PKG}.subprocess.run", side_effect=fake_run):
+            assert _pacman.install_package("steam", masked_hooks=(HOOK,)) is True
+        dirs = seen["dirs"]
+        assert isinstance(dirs, list)
+        assert dirs[:2] == ["/usr/share/libalpm/hooks", "/etc/pacman.d/hooks"]
+        assert seen["entries"] == {HOOK: Path("/dev/null")}
+        mask = seen["mask"]
+        assert isinstance(mask, Path)
+        assert not mask.exists()
+
+    def test_nonzero_exit_is_false_and_logged(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test that a failed transaction is reported with pacman's stderr."""
+        run = MagicMock(return_value=MagicMock(returncode=1, stderr="boom 404"))
+        with (
+            patch(f"{PKG}.subprocess.run", run),
+            caplog.at_level(logging.ERROR),
+        ):
+            assert _pacman.install_package("steam", masked_hooks=(HOOK,)) is False
+        assert "boom 404" in caplog.text
+
+    @pytest.mark.parametrize(
+        "error", [OSError("no sudo"), subprocess.TimeoutExpired("pacman", 1)]
+    )
+    def test_launch_errors_are_false(self, error: Exception) -> None:
+        """Test that a missing binary or a timeout never raises."""
+        with patch(f"{PKG}.subprocess.run", side_effect=error):
+            assert _pacman.install_package("steam") is False
+
+    def test_mask_dir_is_removed_even_when_pacman_fails(self) -> None:
+        """Test that no temp directory outlives a failed install."""
+        masks: list[Path] = []
+
+        def fake_run(argv: list[str], **_kw: object) -> MagicMock:
+            masks.append(Path(_hookdirs(argv)[-1]))
+            gone = "gone"
+            raise OSError(gone)
+
+        with patch(f"{PKG}.subprocess.run", side_effect=fake_run):
+            assert _pacman.install_package("steam", masked_hooks=(HOOK,)) is False
+        assert masks
+        assert not masks[0].exists()

steam_backlog_enforcer/tests/test_research_expiry.py

@@ -0,0 +1,83 @@
+"""The window's end: latch, full block back, store block back once."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer._research_net import RETRY_AFTER_SECONDS
+from steam_backlog_enforcer.tests._research_fixtures import Mocks, run_pass
+from steam_backlog_enforcer.tests._research_helpers import HOUR, NOW, make_record
+
+
+class TestExpiry:
+    """At the deadline the full block returns, once, and the window stays shut."""
+
+    def test_expiry_latches_then_reapplies_the_full_block(self, tm: Mocks) -> None:
+        """Test that a passed deadline closes the record and runs the block."""
+        window.write_record(make_record())
+        run_pass(NOW + 2 * HOUR)
+        record = window.read_record()
+        assert record is not None
+        assert record.closed is True
+        tm.block_tick.assert_called_once_with()
+        tm.reconcile.assert_not_called()
+
+    def test_store_is_reblocked_once(self, tm: Mocks) -> None:
+        """Test that the normal store block returns once and is remembered."""
+        window.write_record(make_record())
+        run_pass(NOW + 2 * HOUR)
+        tm.restore_store.assert_called_once_with()
+        record = window.read_record()
+        assert record is not None
+        assert record.store_reblocked is True
+        run_pass(NOW + 3 * HOUR)
+        tm.restore_store.assert_called_once_with()
+
+    def test_failed_store_reblock_is_retried_after_a_pause(self, tm: Mocks) -> None:
+        """Test that the flag is only set once the store is really blocked."""
+        clock = [1000.0]
+        tm.restore_store.return_value = False
+        window.write_record(make_record())
+        with patch(
+            "steam_backlog_enforcer._research_net.monotonic",
+            side_effect=lambda: clock[0],
+        ):
+            run_pass(NOW + 2 * HOUR)
+            record = window.read_record()
+            assert record is not None
+            assert record.store_reblocked is False
+            run_pass(NOW + 2 * HOUR + 3)
+            assert tm.restore_store.call_count == 1
+            clock[0] += RETRY_AFTER_SECONDS + 1
+            tm.restore_store.return_value = True
+            run_pass(NOW + 2 * HOUR + 70)
+        assert tm.restore_store.call_count == 2
+        record = window.read_record()
+        assert record is not None
+        assert record.store_reblocked is True
+
+    def test_store_not_managed_is_just_flagged(self, tm: Mocks) -> None:
+        """Test that block_store=False in config leaves the store alone."""
+        window.write_record(make_record())
+        run_pass(NOW + 2 * HOUR, block_store=False)
+        tm.restore_store.assert_not_called()
+        record = window.read_record()
+        assert record is not None
+        assert record.store_reblocked is True
+
+    def test_clock_rolled_back_after_expiry_cannot_reopen(self, tm: Mocks) -> None:
+        """Test that the latch beats a clock set back into the old window."""
+        window.write_record(make_record())
+        run_pass(NOW + 2 * HOUR)
+        tm.reconcile.reset_mock()
+        run_pass(NOW)
+        tm.reconcile.assert_not_called()
+
+    def test_latch_write_failure_still_blocks(self, tm: Mocks) -> None:
+        """Test that an unwritable record cannot keep the window open."""
+        window.write_record(make_record())
+        with patch.object(window, "write_record", side_effect=OSError("ro")):
+            run_pass(NOW + 2 * HOUR)
+        tm.block_tick.assert_called_once_with()
+        tm.reconcile.assert_not_called()

steam_backlog_enforcer/tests/test_research_hosts.py

@@ -0,0 +1,184 @@
+"""The research window's /etc/hosts half: Steam names open, game sites shut."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer._total_block_domains import (
+    _ALL_TOTAL_BLOCK_DOMAINS,
+    GAME_WEBSITE_DOMAINS,
+    RESEARCH_BLOCKED_DOMAINS,
+    TOTAL_BLOCK_DOMAINS,
+    TOTAL_BLOCK_EXEMPT_DOMAINS,
+)
+from steam_backlog_enforcer._total_block_hosts import (
+    _HOSTS_BLOCK_BEGIN,
+    _HOSTS_BLOCK_END,
+    apply_total_block_hosts,
+    refresh_total_block_hosts,
+)
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+PKG = "steam_backlog_enforcer._total_block_hosts"
+STEAM_NAMES = (
+    "store.steampowered.com",
+    "checkout.steampowered.com",
+    "steamcommunity.com",
+    "api.steampowered.com",
+    "login.steampowered.com",
+    "steamcontent.com",
+    "steamstatic.com",
+    "cdn.steamstatic.com",
+)
+GAME_SITES = ("newgrounds.com", "www.newgrounds.com", "poki.com", "www.y8.com")
+
+
+class _Hosts:
+    """Plays back what a refresh writes as the file's new content."""
+
+    def __init__(self, paths: Paths, content: str) -> None:
+        self.path = paths.hosts_file
+        self.path.write_text(content, encoding="utf-8")
+        self.writes = 0
+
+    def write(self, content: str) -> None:
+        """Stand-in for the sudo-tee write."""
+        self.writes += 1
+        self.path.write_text(content, encoding="utf-8")
+
+    def text(self) -> str:
+        """The file as it is now."""
+        return self.path.read_text(encoding="utf-8")
+
+
+@pytest.fixture(name="hosts")
+def _hosts(total_block_paths: Paths) -> _Hosts:
+    """A hosts file with the full section between unrelated lines."""
+    full = _HOSTS_BLOCK_BEGIN + "".join(
+        f"0.0.0.0 {d}\n" for d in _ALL_TOTAL_BLOCK_DOMAINS
+    )
+    return _Hosts(
+        total_block_paths,
+        "127.0.0.1 localhost\n" + full + _HOSTS_BLOCK_END + "0.0.0.0 evil.example\n",
+    )
+
+
+def _refresh(hosts: _Hosts, *, steam_open: bool) -> bool:
+    with (
+        patch(f"{PKG}._disable_hosts_protection"),
+        patch(f"{PKG}._enable_hosts_protection"),
+        patch(f"{PKG}._sudo_write_hosts", side_effect=hosts.write),
+    ):
+        return refresh_total_block_hosts(steam_open=steam_open)
+
+
+class TestDomainTable:
+    """The research list is the game sites, split off the same source."""
+
+    def test_is_exactly_the_game_sites_with_www(self) -> None:
+        """Test that no Steam name is in the research block list."""
+        assert not set(RESEARCH_BLOCKED_DOMAINS) & set(TOTAL_BLOCK_DOMAINS)
+        for site in GAME_SITES:
+            assert site in RESEARCH_BLOCKED_DOMAINS
+        assert set(GAME_WEBSITE_DOMAINS) <= set(RESEARCH_BLOCKED_DOMAINS)
+
+    def test_union_with_steam_names_is_the_full_list(self) -> None:
+        """Test that the split loses nothing and keeps the full-list order."""
+        steam_part = [
+            d for d in _ALL_TOTAL_BLOCK_DOMAINS if d not in RESEARCH_BLOCKED_DOMAINS
+        ]
+        assert set(steam_part) | set(RESEARCH_BLOCKED_DOMAINS) == set(
+            _ALL_TOTAL_BLOCK_DOMAINS
+        )
+
+    def test_exempt_domains_stay_out(self) -> None:
+        """Test that the itch exception holds in the research list too."""
+        assert not set(RESEARCH_BLOCKED_DOMAINS) & set(TOTAL_BLOCK_EXEMPT_DOMAINS)
+
+
+class TestOpenSection:
+    """Opening swaps the section body and nothing else."""
+
+    def test_open_drops_steam_names_and_keeps_game_sites(self, hosts: _Hosts) -> None:
+        """Test that the open section null-routes only the game sites."""
+        assert _refresh(hosts, steam_open=True) is True
+        text = hosts.text()
+        for name in STEAM_NAMES:
+            assert f"0.0.0.0 {name}\n" not in text
+        for site in GAME_SITES:
+            assert f"0.0.0.0 {site}\n" in text
+
+    def test_surrounding_lines_survive(self, hosts: _Hosts) -> None:
+        """Test that only the text between the markers changed."""
+        _refresh(hosts, steam_open=True)
+        text = hosts.text()
+        assert text.startswith("127.0.0.1 localhost\n" + _HOSTS_BLOCK_BEGIN)
+        assert text.endswith(_HOSTS_BLOCK_END + "0.0.0.0 evil.example\n")
+
+    def test_already_open_section_is_not_rewritten(self, hosts: _Hosts) -> None:
+        """Test that a converged tick performs no write."""
+        _refresh(hosts, steam_open=True)
+        hosts.writes = 0
+        assert _refresh(hosts, steam_open=True) is True
+        assert hosts.writes == 0
+
+    def test_missing_section_gets_the_open_variant(self, hosts: _Hosts) -> None:
+        """Test that an absent section is created open, never full."""
+        hosts.path.write_text("127.0.0.1 localhost\n", encoding="utf-8")
+        assert _refresh(hosts, steam_open=True) is True
+        text = hosts.text()
+        assert _HOSTS_BLOCK_BEGIN in text
+        assert "0.0.0.0 store.steampowered.com\n" not in text
+        assert "0.0.0.0 newgrounds.com\n" in text
+
+    def test_missing_section_without_trailing_newline(self, hosts: _Hosts) -> None:
+        """Test that the new section starts on its own line."""
+        hosts.path.write_text("127.0.0.1 localhost", encoding="utf-8")
+        _refresh(hosts, steam_open=True)
+        assert hosts.text().startswith("127.0.0.1 localhost\n" + _HOSTS_BLOCK_BEGIN)
+
+    def test_unreadable_file_fails_closed(self, hosts: _Hosts) -> None:
+        """Test that a missing hosts file is a False, not an exception."""
+        hosts.path.unlink()
+        assert _refresh(hosts, steam_open=True) is False
+
+    def test_begin_without_end_is_left_alone(self, hosts: _Hosts) -> None:
+        """Test that a damaged section is not guessed at."""
+        hosts.path.write_text(_HOSTS_BLOCK_BEGIN + "0.0.0.0 x\n", encoding="utf-8")
+        assert _refresh(hosts, steam_open=True) is False
+        assert hosts.writes == 0
+
+    @pytest.mark.usefixtures("hosts")
+    def test_write_failure_is_false(self) -> None:
+        """Test that a failing sudo tee is reported."""
+        with (
+            patch(f"{PKG}._disable_hosts_protection"),
+            patch(f"{PKG}._enable_hosts_protection") as relock,
+            patch(f"{PKG}._sudo_write_hosts", side_effect=OSError("tee")),
+        ):
+            assert refresh_total_block_hosts(steam_open=True) is False
+        relock.assert_called_once()
+
+
+class TestClose:
+    """After the window the full list comes back through the ordinary refresh."""
+
+    def test_full_refresh_restores_every_steam_name(self, hosts: _Hosts) -> None:
+        """Test that refresh without steam_open re-null-routes Steam."""
+        _refresh(hosts, steam_open=True)
+        assert _refresh(hosts, steam_open=False) is True
+        text = hosts.text()
+        for name in STEAM_NAMES:
+            assert f"0.0.0.0 {name}\n" in text
+
+    def test_apply_still_leaves_a_present_section_alone(self, hosts: _Hosts) -> None:
+        """Test that the cheap per-tick apply keeps its old contract."""
+        _refresh(hosts, steam_open=True)
+        hosts.writes = 0
+        assert apply_total_block_hosts() is True
+        assert hosts.writes == 0

steam_backlog_enforcer/tests/test_research_iptables.py

@@ -0,0 +1,134 @@
+"""The research window's firewall half: drop the chain, keep the IP memory."""
+
+from __future__ import annotations
+
+import socket
+from unittest.mock import MagicMock, patch
+
+from steam_backlog_enforcer._total_block_iptables import (
+    IPTABLES_CHAIN,
+    _load_cached_ips,
+    _save_cached_ips,
+    apply_total_block_iptables,
+    release_total_block_iptables,
+    total_block_chain_exists,
+)
+
+PKG = "steam_backlog_enforcer._total_block_iptables"
+STEAM_IP = "23.1.1.1"
+CDN_IP = "23.2.2.2"
+SITE_IP = "172.67.1.1"
+
+
+class FakeChain:
+    """A fake iptables holding one chain, its rules and its OUTPUT hook."""
+
+    def __init__(self, ips: list[str], *, exists: bool = True) -> None:
+        self.rules = list(ips)
+        self.exists = exists
+        self.hooked = exists
+        self.calls: list[list[str]] = []
+        self.stuck = False
+
+    def run(self, cmd: list[str], **_kw: object) -> MagicMock:
+        """Answer one iptables call."""
+        self.calls.append(cmd)
+        args = cmd[2:]  # [sudo, iptables, <verb>, ...]
+        verb = args[0]
+        ok = MagicMock(returncode=0, stdout="")
+        if verb == "-S":
+            if not self.exists:
+                return MagicMock(returncode=1, stdout="")
+            lines = [f"-A {IPTABLES_CHAIN} -d {ip}/32 -j DROP" for ip in self.rules]
+            return MagicMock(returncode=0, stdout="\n".join(lines) + "\n")
+        if verb == "-C":
+            return MagicMock(returncode=0 if self.hooked else 1)
+        if verb == "-I":
+            self.hooked = True
+        elif verb == "-N":
+            self.exists = True
+        elif verb == "-D" and args[1] == "OUTPUT":
+            self.hooked = False
+        elif verb == "-F":
+            self.rules.clear()
+        elif verb == "-X" and not self.stuck:
+            self.exists = False
+        elif verb == "-A":
+            self.rules.append(args[args.index("-d") + 1])
+        return ok
+
+
+def _dns(*names: str, ip: str = STEAM_IP) -> MagicMock:
+    def getaddrinfo(host: str, *_a: object, **_k: object) -> list[tuple[object, ...]]:
+        if host not in names:
+            raise socket.gaierror
+        return [(None, None, None, None, (ip, 443))]
+
+    return MagicMock(side_effect=getaddrinfo)
+
+
+class TestChainExists:
+    """The cheap per-tick probe."""
+
+    def test_present(self) -> None:
+        """Test that a listed chain exists."""
+        with patch(f"{PKG}.subprocess.run", side_effect=FakeChain([STEAM_IP]).run):
+            assert total_block_chain_exists() is True
+
+    def test_absent(self) -> None:
+        """Test that a missing chain does not."""
+        fake = FakeChain([], exists=False)
+        with patch(f"{PKG}.subprocess.run", side_effect=fake.run):
+            assert total_block_chain_exists() is False
+
+
+class TestRelease:
+    """Opening the window removes the chain but remembers what it dropped."""
+
+    def test_chain_and_hook_are_gone_and_cache_survives(self) -> None:
+        """Test that the IP cache is the memory the expiry rebuild needs."""
+        _save_cached_ips({STEAM_IP, CDN_IP})
+        fake = FakeChain([STEAM_IP, CDN_IP])
+        with patch(f"{PKG}.subprocess.run", side_effect=fake.run):
+            assert release_total_block_iptables() is True
+        assert fake.exists is False
+        assert fake.hooked is False
+        assert _load_cached_ips() == {STEAM_IP, CDN_IP}
+
+    def test_absent_chain_is_already_released(self) -> None:
+        """Test that releasing twice is fine."""
+        fake = FakeChain([], exists=False)
+        with patch(f"{PKG}.subprocess.run", side_effect=fake.run):
+            assert release_total_block_iptables() is True
+
+    def test_chain_that_will_not_go_is_false(self) -> None:
+        """Test that a failed delete is reported, not assumed."""
+        fake = FakeChain([STEAM_IP])
+        fake.stuck = True
+        with patch(f"{PKG}.subprocess.run", side_effect=fake.run):
+            assert release_total_block_iptables() is False
+
+    def test_iptables_that_cannot_run_is_false(self) -> None:
+        """Test that a missing iptables fails closed to 'not released'."""
+        with patch(f"{PKG}.subprocess.run", side_effect=OSError):
+            assert release_total_block_iptables() is False
+
+
+class TestRebuildAfterRelease:
+    """At expiry the ordinary tick rebuilds the whole chain from the cache."""
+
+    def test_cached_steam_ips_are_dropped_again(self) -> None:
+        """Test that apply restores cached IPs plus freshly resolved ones."""
+        _save_cached_ips({STEAM_IP, CDN_IP})
+        fake = FakeChain([STEAM_IP, CDN_IP])
+        with patch(f"{PKG}.subprocess.run", side_effect=fake.run):
+            release_total_block_iptables()
+        dns = _dns("store.steampowered.com", ip=SITE_IP)
+        with (
+            patch(f"{PKG}.subprocess.run", side_effect=fake.run),
+            patch(f"{PKG}.socket.getaddrinfo", dns),
+        ):
+            assert apply_total_block_iptables() is True
+        assert sorted(fake.rules) == sorted([STEAM_IP, CDN_IP, SITE_IP])
+        assert fake.exists is True
+        assert fake.hooked is True

steam_backlog_enforcer/tests/test_research_net.py

@@ -0,0 +1,165 @@
+"""Opening the network and the package for a research window."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _research_net as net
+from steam_backlog_enforcer.tests._research_helpers import make_record
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+PKG = "steam_backlog_enforcer._research_net"
+
+
+class _World:
+    """Records every collaborator call in order and answers from knobs."""
+
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+        self.steam_installed = False
+        self.chain_present = True
+        self.store_blocked = True
+        self.install_ok = True
+        self.clock = 1000.0
+
+    def step(self, name: str, *, result: object = True) -> object:
+        """Log *name* and return *result*."""
+        self.calls.append(name)
+        return result
+
+
+@pytest.fixture(name="world")
+def _world() -> Iterator[_World]:
+    """Patch the network/package collaborators with a call-recording world."""
+    w = _World()
+
+    def hosts(*, steam_open: bool = False) -> bool:
+        return bool(w.step(f"hosts:{'open' if steam_open else 'full'}"))
+
+    def install(package: str, *, masked_hooks: tuple[str, ...] = ()) -> bool:
+        w.step(f"install:{package}:{','.join(masked_hooks)}")
+        w.steam_installed = w.steam_installed or w.install_ok
+        return w.install_ok
+
+    with (
+        patch(f"{PKG}.refresh_total_block_hosts", side_effect=hosts),
+        patch(f"{PKG}.flush_dns_cache", side_effect=lambda: w.step("flush")),
+        patch(f"{PKG}.hosts_blocks_store", side_effect=lambda: w.store_blocked),
+        patch(f"{PKG}.unblock_store", side_effect=lambda: w.step("unblock-store")),
+        patch(f"{PKG}.total_block_chain_exists", side_effect=lambda: w.chain_present),
+        patch(
+            f"{PKG}.release_total_block_iptables",
+            side_effect=lambda: w.step("release-chain"),
+        ),
+        patch(f"{PKG}.is_steam_installed", side_effect=lambda: w.steam_installed),
+        patch(f"{PKG}.install_package", side_effect=install),
+        patch(f"{PKG}.monotonic", side_effect=lambda: w.clock),
+    ):
+        yield w
+
+
+class TestReconcile:
+    """One pass makes the network and the package match the open window."""
+
+    def test_first_pass_opens_everything_in_order(self, world: _World) -> None:
+        """Test that hosts, store, firewall, then the package are opened."""
+        net.reconcile_open_window(make_record())
+        assert world.calls == [
+            "hosts:open",
+            "flush",
+            "unblock-store",
+            "release-chain",
+            "install:steam:guard-lib-block-steam-block.hook",
+        ]
+
+    def test_converged_pass_does_no_work(self, world: _World) -> None:
+        """Test that a second pass over an open window forks nothing."""
+        net.reconcile_open_window(make_record())
+        world.calls.clear()
+        world.store_blocked = False
+        world.chain_present = False
+        net.reconcile_open_window(make_record())
+        assert world.calls == ["hosts:open"]
+
+    def test_store_lines_that_come_back_are_undone(self, world: _World) -> None:
+        """Test that the hourly hosts reinstall cannot re-close the store."""
+        net.reconcile_open_window(make_record())
+        world.calls.clear()
+        world.store_blocked = True
+        world.chain_present = False
+        net.reconcile_open_window(make_record())
+        assert world.calls == ["hosts:open", "unblock-store"]
+
+    def test_chain_that_comes_back_is_dropped(self, world: _World) -> None:
+        """Test that a rebuilt chain (e.g. by total-block-refresh) is removed."""
+        net.reconcile_open_window(make_record())
+        world.calls.clear()
+        world.store_blocked = False
+        world.chain_present = True
+        net.reconcile_open_window(make_record())
+        assert world.calls == ["hosts:open", "release-chain"]
+
+    def test_a_restarted_daemon_redoes_the_one_time_release(
+        self, world: _World
+    ) -> None:
+        """Test that the first pass of a fresh process is a full pass again."""
+        net.reconcile_open_window(make_record())
+        net.RELEASED.clear()
+        world.calls.clear()
+        net.reconcile_open_window(make_record())
+        assert "unblock-store" in world.calls
+        assert "flush" in world.calls
+
+    def test_failed_hosts_refresh_is_retried_next_pass(self, world: _World) -> None:
+        """Test that the one-time release is not marked done on failure."""
+        with patch(f"{PKG}.refresh_total_block_hosts", return_value=False):
+            net.reconcile_open_window(make_record())
+        assert not net.RELEASED
+        world.calls.clear()
+        net.reconcile_open_window(make_record())
+        assert world.calls[:3] == ["hosts:open", "flush", "unblock-store"]
+
+
+class TestSteamPackage:
+    """Reinstalling Steam, only when missing, never in a tight loop."""
+
+    def test_installed_steam_is_left_alone(self, world: _World) -> None:
+        """Test that an installed package is never reinstalled."""
+        world.steam_installed = True
+        assert net.ensure_steam_installed() is True
+        assert not [c for c in world.calls if c.startswith("install")]
+
+    def test_missing_steam_is_installed_without_the_block_hook(
+        self, world: _World
+    ) -> None:
+        """Test that only the steam-block hook is masked for the install."""
+        assert net.ensure_steam_installed() is True
+        assert world.calls == ["install:steam:guard-lib-block-steam-block.hook"]
+
+    def test_failed_install_backs_off(self, world: _World) -> None:
+        """Test that a failure waits before the next attempt."""
+        world.install_ok = False
+        assert net.ensure_steam_installed() is False
+        world.calls.clear()
+        world.clock += net.RETRY_AFTER_SECONDS - 1
+        assert net.ensure_steam_installed() is False
+        assert world.calls == []
+        world.clock += 2
+        assert net.ensure_steam_installed() is False
+        assert len(world.calls) == 1
+
+    def test_pacman_database_lock_defers_the_install(self, world: _World) -> None:
+        """Test that a running pacman transaction is not raced."""
+        net.DB_LOCK_FILE.parent.mkdir(parents=True)
+        net.DB_LOCK_FILE.write_text("", encoding="utf-8")
+        assert net.ensure_steam_installed() is False
+        assert world.calls == []
+
+    def test_hook_name_follows_the_package_block_name(self) -> None:
+        """Test that the masked hook is guard-lib's file for steam-block."""
+        assert net.block_hook_name() == "guard-lib-block-steam-block.hook"

steam_backlog_enforcer/tests/test_research_open.py

@@ -0,0 +1,182 @@
+"""Opening a research window: the 48 h cap, once per lock, fail closed."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+import os
+import time
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _research_record as store
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer.tests._research_helpers import (
+    HOUR,
+    LOCK_END,
+    LOCK_START,
+    NOW,
+    lock,
+    make_record,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from steam_backlog_enforcer._research_record import ResearchRecord
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+NOW_PATCH = "steam_backlog_enforcer._research_window.now"
+
+
+def _at(offset: float) -> datetime:
+    """The instant NOW + *offset* seconds, timezone-aware."""
+    return datetime.fromtimestamp(NOW + offset, tz=UTC)
+
+
+def _open(offset: float) -> ResearchRecord:
+    with patch(NOW_PATCH, return_value=NOW):
+        return window.open_research_window(_at(offset))
+
+
+@pytest.fixture(name="live")
+def _live(total_block_paths: Paths) -> None:
+    """An active 6-day lock, two days in."""
+    lock(total_block_paths)
+
+
+@pytest.mark.usefixtures("live")
+class TestOpen:
+    """The happy path and the bounds around it."""
+
+    def test_records_the_window_for_this_lock(self) -> None:
+        """Test that a valid expiry is stored with the lock's identity."""
+        record = _open(10 * HOUR)
+        assert record == window.read_record()
+        assert (record.lock_started_at, record.lock_until) == (LOCK_START, LOCK_END)
+        assert (record.opened_at, record.until) == (NOW, NOW + 10 * HOUR)
+        assert (record.closed, record.store_reblocked) == (False, False)
+
+    def test_exactly_48_hours_is_allowed(self) -> None:
+        """Test that the cap is inclusive."""
+        assert _open(48 * HOUR).until == NOW + 48 * HOUR
+
+    def test_more_than_48_hours_is_refused(self) -> None:
+        """Test that one second over the cap is refused and nothing written."""
+        with pytest.raises(window.ResearchRefusedError, match="48"):
+            _open(48 * HOUR + 1)
+        assert window.read_record() is None
+
+    @pytest.mark.parametrize("offset", [0.0, -1.0, -3 * HOUR])
+    def test_not_in_the_future_is_refused(self, offset: float) -> None:
+        """Test that now and the past are refused."""
+        with pytest.raises(window.ResearchRefusedError, match="future"):
+            _open(offset)
+        assert window.read_record() is None
+
+    def test_window_may_not_outlive_the_lock(self) -> None:
+        """Test that an expiry after the lock end is refused."""
+        with (
+            patch(NOW_PATCH, return_value=LOCK_END - HOUR),
+            pytest.raises(window.ResearchRefusedError, match="lock"),
+        ):
+            window.open_research_window(datetime.fromtimestamp(LOCK_END + 1, tz=UTC))
+        assert window.read_record() is None
+
+    def test_write_failure_is_a_refusal(self) -> None:
+        """Test that an unwritable state dir is reported, not a traceback."""
+        with (
+            patch.object(window, "write_record", side_effect=OSError("ro")),
+            pytest.raises(window.ResearchRefusedError, match="write"),
+        ):
+            _open(HOUR)
+
+
+class TestNoLock:
+    """Without an active total block there is nothing to research."""
+
+    def test_no_lock_file(self) -> None:
+        """Test that the command has no meaning without a block."""
+        with pytest.raises(window.ResearchRefusedError, match="total block"):
+            _open(HOUR)
+
+    def test_expired_lock(self, total_block_paths: Paths) -> None:
+        """Test that a lock that already ended cannot host a window."""
+        lock(total_block_paths)
+        with (
+            patch(NOW_PATCH, return_value=LOCK_END + 1),
+            pytest.raises(window.ResearchRefusedError, match="total block"),
+        ):
+            window.open_research_window(datetime.fromtimestamp(LOCK_END + 2, UTC))
+
+
+@pytest.mark.usefixtures("live")
+class TestOncePerLock:
+    """One window per lock, however it ended; reruns never extend."""
+
+    def test_second_open_is_refused_and_changes_nothing(self) -> None:
+        """Test that a rerun with a later expiry never extends the window."""
+        _open(2 * HOUR)
+        before = store.RESEARCH_FILE.read_bytes()
+        with pytest.raises(window.ResearchRefusedError, match="already"):
+            _open(40 * HOUR)
+        assert store.RESEARCH_FILE.read_bytes() == before
+
+    def test_used_window_stays_used_after_it_closed(self) -> None:
+        """Test that an expired, latched window is not a free second window."""
+        window.write_record(make_record(closed=True, store_reblocked=True))
+        with pytest.raises(window.ResearchRefusedError, match="already"):
+            _open(HOUR)
+
+    def test_a_new_lock_gets_its_own_window(self, total_block_paths: Paths) -> None:
+        """Test that last lock's record does not burn the next lock's window."""
+        window.write_record(make_record(lock_started_at=LOCK_START - 7 * 86400))
+        lock(total_block_paths)
+        assert _open(HOUR).lock_started_at == LOCK_START
+
+    def test_corrupt_state_refuses_instead_of_resetting(self) -> None:
+        """Test that corruption cannot be used to get a second window."""
+        store.RESEARCH_FILE.parent.mkdir(parents=True)
+        store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+        with pytest.raises(window.ResearchRefusedError, match="unreadable"):
+            _open(HOUR)
+        assert store.RESEARCH_FILE.read_text(encoding="utf-8") == "{oops"
+
+
+@pytest.fixture(name="warsaw")
+def _warsaw(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """Run the test in Europe/Warsaw local time, restoring the zone after."""
+    monkeypatch.setenv("TZ", "Europe/Warsaw")
+    time.tzset()
+    yield
+    monkeypatch.undo()
+    time.tzset()
+
+
+class TestParseUntil:
+    """--until is local time unless it carries an offset."""
+
+    @pytest.mark.usefixtures("warsaw")
+    def test_naive_means_local_time(self) -> None:
+        """Test that 2026-11-13T00:00 is midnight in the local zone."""
+        parsed = window.parse_until("2026-11-13T00:00")
+        assert parsed == datetime(2026, 11, 12, 23, 0, tzinfo=UTC)
+        assert os.environ["TZ"] == "Europe/Warsaw"
+
+    def test_offset_is_honoured(self) -> None:
+        """Test that an explicit offset is not reinterpreted."""
+        parsed = window.parse_until("2026-11-13T00:00:00+02:00")
+        assert parsed == datetime(2026, 11, 12, 22, 0, tzinfo=UTC)
+
+    def test_z_suffix(self) -> None:
+        """Test that Z means UTC."""
+        assert window.parse_until("2026-11-13T00:00:00Z") == datetime(
+            2026, 11, 13, tzinfo=UTC
+        )
+
+    @pytest.mark.parametrize("raw", ["", "tomorrow", "2026-13-40", "13:00"])
+    def test_garbage_is_a_value_error(self, raw: str) -> None:
+        """Test that an unparsable expiry is rejected."""
+        with pytest.raises(ValueError, match="until"):
+            window.parse_until(raw)

steam_backlog_enforcer/tests/test_research_record.py

@@ -0,0 +1,139 @@
+"""The research record: strict parse, fail-closed reads, window predicate."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from steam_backlog_enforcer import _research_record as store
+from steam_backlog_enforcer.tests._research_helpers import (
+    HOUR,
+    LOCK_END,
+    LOCK_START,
+    NOW,
+    make_record,
+)
+
+
+def _raw(**changes: object) -> str:
+    """The JSON of a valid record with *changes* applied (None drops a key)."""
+    data: dict[str, object] = {
+        "lock_started_at": LOCK_START,
+        "lock_until": LOCK_END,
+        "opened_at": NOW - 60,
+        "until": NOW + HOUR,
+        "closed": False,
+        "store_reblocked": False,
+    }
+    for key, value in changes.items():
+        if value is None:
+            del data[key]
+        else:
+            data[key] = value
+    return json.dumps(data)
+
+
+class TestParse:
+    """A record is either exactly valid or an error; there is no repair."""
+
+    def test_roundtrip(self) -> None:
+        """Test that to_json then parse gives the same record."""
+        record = make_record(closed=True, store_reblocked=True)
+        assert store.parse_record(record.to_json()) == record
+
+    def test_flags_default_to_false(self) -> None:
+        """Test that a record without the latches reads as not closed."""
+        parsed = store.parse_record(_raw(closed=None, store_reblocked=None))
+        assert (parsed.closed, parsed.store_reblocked) == (False, False)
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            "not json",
+            "[]",
+            "null",
+            "{}",
+            _raw(until=None),
+            _raw(lock_started_at=None),
+            _raw(until="2026-11-13"),
+            _raw(until=True),
+            _raw(opened_at=float("nan")),
+            _raw(until=float("inf")),
+            _raw(until=10**400),
+            _raw(until=NOW - 60),
+            _raw(until=NOW - 3600),
+            _raw(lock_until=LOCK_START),
+            _raw(until=LOCK_END + 1),
+            _raw(opened_at=LOCK_END - HOUR, until=LOCK_END + HOUR),
+            _raw(until=NOW - 60 + 48 * HOUR + 2),
+            _raw(closed="yes"),
+            _raw(store_reblocked=1),
+        ],
+    )
+    def test_invalid_is_an_error(self, raw: str) -> None:
+        """Test that every malformed or out-of-policy record is refused."""
+        with pytest.raises(store.ResearchStateError):
+            store.parse_record(raw)
+
+    def test_exactly_48_hours_is_valid(self) -> None:
+        """Test that the cap is inclusive."""
+        parsed = store.parse_record(_raw(until=NOW - 60 + 48 * HOUR))
+        assert parsed.until - parsed.opened_at == 48 * HOUR
+
+
+class TestReadWrite:
+    """Reading distinguishes never-used from unreadable."""
+
+    def test_absent_is_none(self) -> None:
+        """Test that no file means the window was never used."""
+        assert store.read_record() is None
+
+    def test_write_then_read(self) -> None:
+        """Test that a written record comes back, root-only on disk."""
+        record = make_record()
+        store.write_record(record)
+        assert store.read_record() == record
+        assert store.RESEARCH_FILE.stat().st_mode & 0o777 == 0o600
+        assert store.RESEARCH_FILE.parent.stat().st_mode & 0o777 == 0o700
+
+    def test_corrupt_raises(self) -> None:
+        """Test that a corrupt file is an error, not 'never used'."""
+        store.RESEARCH_FILE.parent.mkdir(parents=True)
+        store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+        with pytest.raises(store.ResearchStateError):
+            store.read_record()
+
+    def test_unreadable_raises(self) -> None:
+        """Test that an unreadable path (a directory) is an error."""
+        store.RESEARCH_FILE.mkdir(parents=True)
+        with pytest.raises(store.ResearchStateError):
+            store.read_record()
+
+
+class TestIsOpen:
+    """Open means opened_at <= now < until and not latched closed."""
+
+    def test_inside(self) -> None:
+        """Test that a time inside the window is open."""
+        assert make_record().is_open(NOW)
+
+    def test_opened_at_is_inclusive(self) -> None:
+        """Test that the first instant is open."""
+        assert make_record().is_open(NOW - 60)
+
+    def test_before_opening_is_closed(self) -> None:
+        """Test that a clock set back before opening does not open it."""
+        assert not make_record().is_open(NOW - 61)
+
+    def test_until_is_exclusive(self) -> None:
+        """Test that the deadline instant is already closed."""
+        assert not make_record().is_open(NOW + HOUR)
+
+    def test_after_deadline_is_closed(self) -> None:
+        """Test that a late clock is closed."""
+        assert not make_record().is_open(NOW + 5 * HOUR)
+
+    def test_latch_wins_over_the_clock(self) -> None:
+        """Test that a closed record stays closed even for an in-range clock."""
+        assert not make_record(closed=True).is_open(NOW)

steam_backlog_enforcer/tests/test_research_strict.py

@@ -0,0 +1,124 @@
+"""Strict mode: during research only runtimes survive besides the allowed game."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+from unittest.mock import patch
+
+from steam_backlog_enforcer._enforce_guards import guard_installed_games
+from steam_backlog_enforcer.enforcer import enforce_allowed_game
+from steam_backlog_enforcer.game_uninstall import (
+    is_infrastructure_app,
+    is_protected_app,
+)
+
+UNI = "steam_backlog_enforcer.game_uninstall"
+LIBRARIAN = 4197610
+EXCEPTION = 880940  # an approved whitelist exception
+ZERO_ACH = 555  # an owned game with no achievements
+PROTON = 1493710  # Proton Experimental, a runtime
+OTHER = 111
+
+
+def _world() -> tuple[object, object]:
+    """Patch the two exemption sources: approved exceptions and zero-ach games."""
+    return (
+        patch(f"{UNI}.get_approved_exception_ids", return_value={EXCEPTION}),
+        patch(f"{UNI}.zero_achievement_ids", return_value={ZERO_ACH}),
+    )
+
+
+class TestPredicates:
+    """Infrastructure is the strict subset of protected."""
+
+    def test_runtime_is_infrastructure(self) -> None:
+        """Test that Proton is protected in both modes."""
+        assert is_infrastructure_app(PROTON)
+        assert is_protected_app(PROTON)
+
+    def test_exceptions_are_protected_but_not_infrastructure(self) -> None:
+        """Test that the whitelist does not reach strict mode."""
+        exc, zero = _world()
+        with exc, zero:
+            assert is_protected_app(EXCEPTION)
+            assert is_protected_app(ZERO_ACH)
+            assert not is_infrastructure_app(EXCEPTION)
+            assert not is_infrastructure_app(ZERO_ACH)
+
+
+class TestKill:
+    """enforce_allowed_game with and without strict."""
+
+    RUNNING: ClassVar[dict[int, int]] = {
+        10: LIBRARIAN,
+        11: EXCEPTION,
+        12: ZERO_ACH,
+        13: PROTON,
+        14: 0,
+        15: OTHER,
+    }
+
+    def _violations(self, *, strict: bool) -> list[tuple[int, int]]:
+        exc, zero = _world()
+        with (
+            exc,
+            zero,
+            patch(
+                "steam_backlog_enforcer.enforcer.get_running_steam_game_pids",
+                return_value=self.RUNNING,
+            ),
+            patch("steam_backlog_enforcer.enforcer.kill_process"),
+        ):
+            return enforce_allowed_game(
+                {LIBRARIAN}, kill_unauthorized=True, strict=strict
+            )
+
+    def test_default_spares_exceptions(self) -> None:
+        """Test that normal enforcement is unchanged."""
+        assert self._violations(strict=False) == [(15, OTHER)]
+
+    def test_strict_kills_exceptions_and_zero_achievement_games(self) -> None:
+        """Test that only Librarian, runtimes and app 0 survive strict mode."""
+        assert self._violations(strict=True) == [
+            (11, EXCEPTION),
+            (12, ZERO_ACH),
+            (15, OTHER),
+        ]
+
+
+class TestUninstall:
+    """guard_installed_games with and without strict."""
+
+    INSTALLED: ClassVar[list[tuple[int, str]]] = [
+        (LIBRARIAN, "Librarian"),
+        (EXCEPTION, "Party game"),
+        (ZERO_ACH, "No achievements"),
+        (PROTON, "Proton Experimental"),
+        (OTHER, "Other"),
+    ]
+
+    def _removed(self, *, strict: bool) -> list[int]:
+        exc, zero = _world()
+        with (
+            exc,
+            zero,
+            patch(
+                "steam_backlog_enforcer._enforce_guards.get_installed_games",
+                return_value=self.INSTALLED,
+            ),
+            patch(
+                "steam_backlog_enforcer._enforce_guards.uninstall_game",
+                return_value=True,
+            ) as uninstall,
+            patch("steam_backlog_enforcer._enforce_guards.send_notification"),
+        ):
+            guard_installed_games({LIBRARIAN}, strict=strict)
+        return [call.args[0] for call in uninstall.call_args_list]
+
+    def test_default_spares_exceptions(self) -> None:
+        """Test that normal enforcement is unchanged."""
+        assert self._removed(strict=False) == [OTHER]
+
+    def test_strict_removes_everything_but_librarian_and_runtimes(self) -> None:
+        """Test that exceptions and zero-achievement games go in strict mode."""
+        assert self._removed(strict=True) == [EXCEPTION, ZERO_ACH, OTHER]

steam_backlog_enforcer/tests/test_research_tick.py

@@ -0,0 +1,108 @@
+"""One daemon pass under a total block: the window and the full block."""
+
+from __future__ import annotations
+
+from steam_backlog_enforcer import _research_record as store
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer import config as cfg
+from steam_backlog_enforcer.tests._research_fixtures import Mocks, run_pass
+from steam_backlog_enforcer.tests._research_helpers import HOUR, NOW, make_record
+
+
+class TestFullBlock:
+    """Without a live window the whole total block is enforced."""
+
+    def test_no_record_runs_the_full_tick(self, tm: Mocks) -> None:
+        """Test that the default is the full block."""
+        run_pass()
+        tm.block_tick.assert_called_once_with()
+        tm.reconcile.assert_not_called()
+        tm.enforce.assert_not_called()
+
+    def test_corrupt_record_runs_the_full_tick(self, tm: Mocks) -> None:
+        """Test that unreadable research state means fully blocked."""
+        store.RESEARCH_FILE.parent.mkdir(parents=True)
+        store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+        run_pass()
+        tm.block_tick.assert_called_once_with()
+        tm.reconcile.assert_not_called()
+
+    def test_clock_before_opening_blocks_without_latching(self, tm: Mocks) -> None:
+        """Test that a clock set back blocks, but does not burn the window."""
+        window.write_record(make_record())
+        run_pass(NOW - 3 * HOUR)
+        tm.block_tick.assert_called_once_with()
+        tm.reconcile.assert_not_called()
+        record = window.read_record()
+        assert record is not None
+        assert record.closed is False
+
+    def test_latched_and_reblocked_record_only_runs_the_block(self, tm: Mocks) -> None:
+        """Test that a finished window costs nothing but the block tick."""
+        window.write_record(make_record(closed=True, store_reblocked=True))
+        run_pass(NOW + 5 * HOUR)
+        tm.block_tick.assert_called_once_with()
+        tm.restore_store.assert_not_called()
+
+
+class TestOpenWindow:
+    """Inside the window: Steam is left alone, everything but Librarian is not."""
+
+    def test_reconciles_and_skips_the_full_block(self, tm: Mocks) -> None:
+        """Test that the open window replaces the block tick."""
+        window.write_record(make_record())
+        run_pass()
+        tm.reconcile.assert_called_once_with(make_record())
+        tm.block_tick.assert_not_called()
+
+    def test_steam_is_never_killed_uninstalled_or_purged(self, tm: Mocks) -> None:
+        """Test that the destructive half of the block stays off."""
+        window.write_record(make_record())
+        run_pass()
+        tm.steam_kill.assert_not_called()
+        tm.steam_uninstall.assert_not_called()
+        tm.steam_purge.assert_not_called()
+
+    def test_other_launchers_and_counted_games_are_still_killed(
+        self, tm: Mocks
+    ) -> None:
+        """Test that non-Steam games stay forbidden during the window."""
+        window.write_record(make_record())
+        run_pass()
+        tm.launchers.assert_called_once()
+        tm.counted.assert_called_once()
+
+    def test_only_librarian_is_allowed_and_strictly(self, tm: Mocks) -> None:
+        """Test that the allowed set is exactly Librarian, no exceptions."""
+        window.write_record(make_record())
+        run_pass()
+        tm.enforce.assert_called_once_with(
+            {window.LIBRARIAN_APP_ID}, kill_unauthorized=True, strict=True
+        )
+        tm.guard.assert_called_once_with({window.LIBRARIAN_APP_ID}, strict=True)
+
+    def test_killed_game_is_reported(self, tm: Mocks) -> None:
+        """Test that a killed non-Librarian game raises a notification."""
+        tm.enforce.return_value = [(4242, 12345)]
+        window.write_record(make_record())
+        run_pass()
+        tm.notify.assert_called_once()
+
+    def test_no_game_enforcement_while_steam_is_not_installed_yet(
+        self, tm: Mocks
+    ) -> None:
+        """Test that the install gap does not scan an absent library."""
+        tm.steam_installed.return_value = False
+        window.write_record(make_record())
+        run_pass()
+        tm.enforce.assert_not_called()
+        tm.guard.assert_not_called()
+
+    def test_real_assignment_is_never_touched(self, tm: Mocks) -> None:
+        """Test that state.json (the real backlog) is not read or written."""
+        cfg.STATE_FILE.write_text('{"current_app_id": 1366540}', encoding="utf-8")
+        before = cfg.STATE_FILE.read_bytes()
+        window.write_record(make_record())
+        run_pass()
+        assert cfg.STATE_FILE.read_bytes() == before
+        assert tm.enforce.call_args.args[0] == {window.LIBRARIAN_APP_ID}

steam_backlog_enforcer/tests/test_research_window.py

@@ -0,0 +1,145 @@
+"""The research window: lock binding, fail-closed live record, the clock."""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from steam_backlog_enforcer import _research_record as store
+from steam_backlog_enforcer import _research_window as window
+from steam_backlog_enforcer.tests._research_helpers import (
+    HOUR,
+    LOCK_END,
+    LOCK_START,
+    NOW,
+    lock,
+    make_record,
+)
+
+if TYPE_CHECKING:
+    import pytest
+
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+NOW_PATCH = "steam_backlog_enforcer._research_window.now"
+
+
+class TestLiveRecord:
+    """live_record is the daemon's fail-closed view of the file."""
+
+    def test_no_lock_means_no_window(self) -> None:
+        """Test that a record without an active total block is ignored."""
+        window.write_record(make_record())
+        assert window.live_record() is None
+
+    def test_no_record(self, total_block_paths: Paths) -> None:
+        """Test that a lock without a record has no window."""
+        lock(total_block_paths)
+        with patch(NOW_PATCH, return_value=NOW):
+            assert window.live_record() is None
+
+    def test_matching_record(self, total_block_paths: Paths) -> None:
+        """Test that a record for the live lock is returned."""
+        lock(total_block_paths)
+        window.write_record(make_record())
+        with patch(NOW_PATCH, return_value=NOW):
+            assert window.live_record() == make_record()
+
+    def test_other_locks_record_is_ignored(self, total_block_paths: Paths) -> None:
+        """Test that last month's record does not open this lock."""
+        lock(total_block_paths)
+        window.write_record(make_record(lock_started_at=LOCK_START - 99.0))
+        with patch(NOW_PATCH, return_value=NOW):
+            assert window.live_record() is None
+
+    def test_corrupt_record_is_no_window_and_logged_once(
+        self, total_block_paths: Paths, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test that corruption fails closed and does not spam every tick."""
+        lock(total_block_paths)
+        store.RESEARCH_FILE.parent.mkdir(parents=True)
+        store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+        with patch(NOW_PATCH, return_value=NOW), caplog.at_level(logging.ERROR):
+            assert window.live_record() is None
+            assert window.live_record() is None
+        assert len([r for r in caplog.records if "research" in r.message]) == 1
+
+    def test_corruption_is_logged_again_after_recovery(
+        self, total_block_paths: Paths, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Test that a record that heals and breaks again is reported again."""
+        lock(total_block_paths)
+        store.RESEARCH_FILE.parent.mkdir(parents=True)
+        store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+        with patch(NOW_PATCH, return_value=NOW), caplog.at_level(logging.ERROR):
+            window.live_record()
+            window.write_record(make_record())
+            assert window.live_record() is not None
+            store.RESEARCH_FILE.write_text("{oops", encoding="utf-8")
+            window.live_record()
+        assert len([r for r in caplog.records if "research" in r.message]) == 2
+
+
+class TestOpenNow:
+    """live_record plus is_open is the daemon's whole decision."""
+
+    @staticmethod
+    def _open_at(at: float) -> bool:
+        with patch(NOW_PATCH, return_value=at):
+            record = window.live_record()
+        return record is not None and record.is_open(at)
+
+    def test_open_window(self, total_block_paths: Paths) -> None:
+        """Test that an open window on the live lock is open."""
+        lock(total_block_paths)
+        window.write_record(make_record())
+        assert self._open_at(NOW) is True
+
+    def test_expired_window(self, total_block_paths: Paths) -> None:
+        """Test that a passed deadline is not open."""
+        lock(total_block_paths)
+        window.write_record(make_record())
+        assert self._open_at(NOW + 2 * HOUR) is False
+
+    def test_clock_in_the_past_is_not_open(self, total_block_paths: Paths) -> None:
+        """Test that a clock before opening is fully blocked."""
+        lock(total_block_paths)
+        window.write_record(make_record())
+        assert self._open_at(NOW - 3 * HOUR) is False
+
+    def test_closed_latch_is_not_open(self, total_block_paths: Paths) -> None:
+        """Test that a latched window never reopens."""
+        lock(total_block_paths)
+        window.write_record(make_record(closed=True))
+        assert self._open_at(NOW) is False
+
+    def test_missing_state_is_fully_blocked(self, total_block_paths: Paths) -> None:
+        """Test that no record at all is the blocked default."""
+        lock(total_block_paths)
+        assert self._open_at(NOW) is False
+
+    def test_lock_identity_tolerates_float_rounding(
+        self, total_block_paths: Paths
+    ) -> None:
+        """Test that sub-second noise still matches, whole seconds do not."""
+        lock(total_block_paths)
+        window.write_record(make_record(lock_started_at=LOCK_START + 0.4))
+        assert self._open_at(NOW) is True
+        window.write_record(make_record(lock_started_at=LOCK_START + 2.0))
+        assert self._open_at(NOW) is False
+
+    def test_lock_without_start_is_no_lock(self, total_block_paths: Paths) -> None:
+        """Test that a lock file lacking started_at cannot host a window."""
+        total_block_paths.lock_file.write_text(
+            json.dumps({"until": LOCK_END}), encoding="utf-8"
+        )
+        window.write_record(make_record())
+        assert self._open_at(NOW) is False
+
+    def test_now_is_the_wall_clock(self) -> None:
+        """Test that now() reads time.time()."""
+        with patch("steam_backlog_enforcer._research_window.time.time") as clock:
+            clock.return_value = 42.5
+            assert window.now() == 42.5

steam_backlog_enforcer/tests/test_store_restore.py

@@ -0,0 +1,77 @@
+"""restore_store_block: the normal store block back, whatever else blocks it."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+from steam_backlog_enforcer.store_blocker import restore_store_block
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.tests._total_block_paths import Paths
+
+PKG = "steam_backlog_enforcer.store_blocker"
+# The normal section after an unblock, plus the total block's own live copy of
+# the same domain: the live copy is what makes block_store() say "already
+# blocked" and leave the commented lines alone.
+HOSTS = (
+    "127.0.0.1 localhost\n"
+    "# Steam Store\n"
+    "# 0.0.0.0 store.steampowered.com\n"
+    "# 0.0.0.0 checkout.steampowered.com\n"
+    "# BEGIN steam-backlog-enforcer total-block\n"
+    "0.0.0.0 store.steampowered.com\n"
+    "# END steam-backlog-enforcer total-block\n"
+)
+
+
+def _restore(paths: Paths, *, iptables: bool = True) -> tuple[bool, str]:
+    paths.hosts_file.write_text(HOSTS, encoding="utf-8")
+
+    def write(content: str) -> None:
+        paths.hosts_file.write_text(content, encoding="utf-8")
+
+    with (
+        patch("steam_backlog_enforcer._hosts_protection.HOSTS_FILE", paths.hosts_file),
+        patch("steam_backlog_enforcer._hosts_protection._disable_hosts_protection"),
+        patch("steam_backlog_enforcer._hosts_protection._enable_hosts_protection"),
+        patch(
+            "steam_backlog_enforcer._hosts_protection._sudo_write_hosts",
+            side_effect=write,
+        ),
+        patch(f"{PKG}._block_store_iptables", return_value=iptables) as chain,
+        patch(f"{PKG}.flush_dns_cache") as flush,
+    ):
+        result = restore_store_block()
+    chain.assert_called_once_with()
+    flush.assert_called_once_with()
+    return result, paths.hosts_file.read_text(encoding="utf-8")
+
+
+class TestRestoreStoreBlock:
+    """Unlike block_store it never stops at 'already blocked'."""
+
+    def test_uncomments_the_normal_lines_despite_a_live_copy(
+        self, total_block_paths: Paths
+    ) -> None:
+        """Test that the commented normal lines come back."""
+        result, text = _restore(total_block_paths)
+        assert result is True
+        assert "\n0.0.0.0 store.steampowered.com\n0.0.0.0 checkout" in text
+        assert "# 0.0.0.0 store.steampowered.com" not in text
+        assert "# Steam Store\n" in text
+
+    def test_hosts_alone_is_enough(self, total_block_paths: Paths) -> None:
+        """Test that a failing iptables layer still counts as restored."""
+        result, _ = _restore(total_block_paths, iptables=False)
+        assert result is True
+
+    def test_nothing_restored_is_false(self, total_block_paths: Paths) -> None:
+        """Test that failing both layers is reported."""
+        with patch(f"{PKG}._reblock_hosts", return_value=False):
+            total_block_paths.hosts_file.write_text(HOSTS, encoding="utf-8")
+            with (
+                patch(f"{PKG}._block_store_iptables", return_value=False),
+                patch(f"{PKG}.flush_dns_cache"),
+            ):
+                assert restore_store_block() is False

steam_backlog_enforcer/tests/test_total_block_lifecycle.py

@@ -104,7 +104,7 @@ class TestEnforceTotalBlockTick:
             patch(f"{PKG}.is_steam_installed", return_value=True),
             patch(f"{PKG}.uninstall_steam_package") as mock_uninstall,
             patch(f"{PKG}.purge_steam_and_proton"),
-            patch(f"{PKG}.apply_total_block_hosts", return_value=True),
+            patch(f"{PKG}.refresh_total_block_hosts", return_value=True),
             patch(f"{PKG}.apply_total_block_iptables", return_value=True),
         ):
             enforce_total_block_tick()
@@ -117,7 +117,7 @@ class TestEnforceTotalBlockTick:
             patch(f"{PKG}.is_steam_installed", return_value=False),
             patch(f"{PKG}.uninstall_steam_package") as mock_uninstall,
             patch(f"{PKG}.purge_steam_and_proton"),
-            patch(f"{PKG}.apply_total_block_hosts", return_value=True),
+            patch(f"{PKG}.refresh_total_block_hosts", return_value=True),
             patch(f"{PKG}.apply_total_block_iptables", return_value=True),
         ):
             enforce_total_block_tick()
@@ -130,12 +130,30 @@ class TestEnforceTotalBlockTick:
             patch(f"{PKG}.is_steam_installed", return_value=False),
             patch(f"{PKG}.uninstall_steam_package"),
             patch(f"{PKG}.purge_steam_and_proton") as mock_purge,
-            patch(f"{PKG}.apply_total_block_hosts", return_value=True),
+            patch(f"{PKG}.refresh_total_block_hosts", return_value=True),
             patch(f"{PKG}.apply_total_block_iptables", return_value=True),
         ):
             enforce_total_block_tick()
         mock_purge.assert_called_once()
 
+    def test_hosts_are_reconciled_to_the_full_list_every_tick(self) -> None:
+        """Test that the tick restores Steam names after a research window.
+
+        The cheap apply leaves a present section alone, which would keep the
+        window's open section for the rest of the lock.
+        """
+        with (
+            patch(f"{PKG}.kill_steam_and_launchers", return_value=[]),
+            patch(f"{PKG}.is_steam_installed", return_value=False),
+            patch(f"{PKG}.purge_steam_and_proton"),
+            patch(f"{PKG}.apply_total_block_iptables", return_value=True),
+            patch(f"{PKG}.apply_total_block_hosts") as mock_apply,
+            patch(f"{PKG}.refresh_total_block_hosts", return_value=True) as refresh,
+        ):
+            enforce_total_block_tick()
+        refresh.assert_called_once_with()
+        mock_apply.assert_not_called()
+
 
 class TestEndTotalBlockCleanup:
     """Tests for end total block cleanup."""