fix(enforcer): defer Steam restart while a game update is in progress

Restarting Steam (steam -shutdown + relaunch for the CDP port) suspends any
in-flight game update. Repeatedly interrupting a large update — including
during the commit-to-disk phase — leaves a partially written install that
Steam's "verify integrity" still reports as OK, but the game's anti-tamper
detects and crashes on. This corrupted Age of Empires II DE (deterministic
0xc0000005 at AoE2DE_s.exe+0x29B9E3) once commit 00a08de made the enforcer's
CDP restart actually work and it began bouncing Steam mid-update.

Add steam_update_in_progress() in a new _steam_state leaf module (now the
single source of truth for STEAMAPPS_PATH, shared by game_install and
library_hider without an import cycle). It scans appmanifests for unfinished
download/stage bytes. ensure_steam_debug_port() now raises
SteamUpdateInProgressError (a SteamUnavailableError subclass, so the enforce
loop already skips + retries) and restart_steam() skips, when an update runs.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Vw1sLrSjDksvH5zjU2RMYB

steam_backlog_enforcer/_steam_state.py

@@ -0,0 +1,61 @@
+"""Read-only helpers for Steam's on-disk state (install/update progress).
+
+Kept in its own leaf module (no intra-package imports) so both
+``game_install`` and ``library_hider`` can use it without forming an import
+cycle. ``STEAMAPPS_PATH`` lives here as the single source of truth for the
+Steam library location.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import re
+
+STEAMAPPS_PATH = Path("~/.local/share/Steam/steamapps").expanduser()
+
+
+def _manifest_transfer_active(manifest: Path) -> bool:
+    """Return True if *manifest* shows unfinished download or staging bytes.
+
+    Args:
+        manifest: Path to an ``appmanifest_*.acf`` file.
+
+    Returns:
+        True if the app still has bytes left to download or stage to disk.
+    """
+    try:
+        content = manifest.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return False
+
+    def _field(name: str) -> int:
+        match = re.search(rf'"{name}"\s+"(\d+)"', content)
+        return int(match.group(1)) if match else 0
+
+    # BytesTo* are the totals for the in-flight update; BytesDownloaded/Staged
+    # track progress. Progress strictly below the total means the update (or
+    # its on-disk commit) is still running.
+    return _field("BytesToDownload") > _field("BytesDownloaded") or _field(
+        "BytesToStage"
+    ) > _field("BytesStaged")
+
+
+def steam_update_in_progress() -> bool:
+    """Return True if any installed Steam app is mid-update.
+
+    Steam records live transfer progress in each ``appmanifest_*.acf``: a game
+    is actively updating while it still has bytes left to download or commit to
+    disk. Restarting Steam during that window suspends the transfer and can
+    leave a partially-written install — corrupting it. The enforcer checks this
+    before bouncing Steam so it never interrupts an in-flight update (the
+    failure that corrupted Age of Empires II DE into a deterministic launch
+    crash).
+
+    Returns:
+        True if at least one appmanifest shows an unfinished download/stage.
+    """
+    try:
+        manifests = list(STEAMAPPS_PATH.glob("appmanifest_*.acf"))
+    except OSError:
+        return False
+    return any(_manifest_transfer_active(m) for m in manifests)

steam_backlog_enforcer/game_install.py

@@ -13,6 +13,7 @@ import subprocess
 import sys
 import time
 
+from steam_backlog_enforcer._steam_state import STEAMAPPS_PATH
 from steam_backlog_enforcer._whitelist import get_approved_exception_ids
 from steam_backlog_enforcer.library_hider import steam_is_installed
 
@@ -95,8 +96,6 @@ PROTECTED_APP_IDS = {
     1826330,  # Proton EasyAntiCheat Runtime
 }
 
-STEAMAPPS_PATH = Path("~/.local/share/Steam/steamapps").expanduser()
-
 
 def _trigger_steam_install(app_id: int, label: str) -> bool:
     """Ask Steam to install a game via the ``steam://install`` URI.

steam_backlog_enforcer/library_hider.py

@@ -29,6 +29,8 @@ import time
 import requests
 import websockets
 
+from steam_backlog_enforcer._steam_state import steam_update_in_progress
+
 logger = logging.getLogger(__name__)
 
 _CDP_PORT = 9222
@@ -63,6 +65,16 @@ class SteamUnavailableError(RuntimeError):
     """
 
 
+class SteamUpdateInProgressError(SteamUnavailableError):
+    """Raised to defer a Steam restart while a game update is in flight.
+
+    Subclasses :class:`SteamUnavailableError` so existing callers already
+    degrade gracefully (skip this pass, retry next loop). Restarting Steam
+    mid-update suspends and can corrupt the transfer, so the enforcer waits
+    for updates to finish before bouncing Steam to open the CDP port.
+    """
+
+
 def steam_is_installed() -> bool:
     """Return True if the real Steam client binary is present."""
     return Path(_STEAM_BINARY).exists()
@@ -260,6 +272,18 @@ def ensure_steam_debug_port() -> None:
 
     logger.info("Steam CDP port not available — (re)starting Steam...")
     if _is_steam_running():
+        # Never bounce a running Steam while a game update is downloading or
+        # committing: the shutdown suspends it and can leave a partially
+        # written install (the root cause of the AoE2 launch crash). Defer and
+        # retry on the next enforce pass.
+        if steam_update_in_progress():
+            msg = (
+                "Deferring Steam restart: a game update is in progress. "
+                "Restarting now would interrupt and can corrupt it; will "
+                "retry once the update settles."
+            )
+            logger.info(msg)
+            raise SteamUpdateInProgressError(msg)
         _shutdown_steam()
 
     _launch_steam_with_debug()
@@ -403,7 +427,18 @@ def unhide_all_games(owned_app_ids: list[int]) -> int:
 
 
 def restart_steam() -> None:
-    """Gracefully restart the Steam client with CEF debugging enabled."""
+    """Gracefully restart the Steam client with CEF debugging enabled.
+
+    Skips the restart if a game update is downloading or committing, so the
+    update is not interrupted (interrupting it can corrupt the install).
+    """
+    if steam_update_in_progress():
+        logger.warning(
+            "Skipping Steam restart — a game update is in progress; "
+            "restarting now could corrupt it.",
+        )
+        return
+
     logger.info("Restarting Steam client with debug port...")
     _shutdown_steam()
     _launch_steam_with_debug()

steam_backlog_enforcer/tests/conftest.py

@@ -56,7 +56,13 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
             "steam_backlog_enforcer.config.SNAPSHOT_FILE",
             fake_config / "snapshot.json",
         ),
-        # Steam game manifests / install dirs
+        # Steam game manifests / install dirs. STEAMAPPS_PATH is canonical in
+        # _steam_state; game_install imports it, so both bindings need the
+        # patch (same gotcha as HOSTS_FILE below).
+        patch(
+            "steam_backlog_enforcer._steam_state.STEAMAPPS_PATH",
+            fake_steamapps,
+        ),
         patch(
             "steam_backlog_enforcer.game_install.STEAMAPPS_PATH",
             fake_steamapps,

steam_backlog_enforcer/tests/test_update_restart_guard.py

@@ -0,0 +1,142 @@
+"""Tests for the 'defer Steam restart while a game update is running' guard.
+
+Covers the manifest transfer scan in ``_steam_state`` and the two restart
+paths in ``library_hider`` (the daemon's ``ensure_steam_debug_port`` and the
+interactive ``restart_steam``) that must not bounce Steam mid-update.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer._steam_state import (
+    _manifest_transfer_active,
+    steam_update_in_progress,
+)
+from steam_backlog_enforcer.library_hider import (
+    SteamUpdateInProgressError,
+    ensure_steam_debug_port,
+    restart_steam,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+SS = "steam_backlog_enforcer._steam_state"
+LH = "steam_backlog_enforcer.library_hider"
+
+
+def _manifest(*, to_download: int, downloaded: int, to_stage: int, staged: int) -> str:
+    return (
+        '"AppState"\n{\n'
+        '\t"appid"\t\t"813780"\n'
+        f'\t"BytesToDownload"\t\t"{to_download}"\n'
+        f'\t"BytesDownloaded"\t\t"{downloaded}"\n'
+        f'\t"BytesToStage"\t\t"{to_stage}"\n'
+        f'\t"BytesStaged"\t\t"{staged}"\n'
+        "}\n"
+    )
+
+
+_DOWNLOADING = _manifest(to_download=100, downloaded=40, to_stage=100, staged=100)
+_STAGING = _manifest(to_download=100, downloaded=100, to_stage=100, staged=40)
+_IDLE = _manifest(to_download=0, downloaded=0, to_stage=0, staged=0)
+_COMPLETE = _manifest(to_download=100, downloaded=100, to_stage=100, staged=100)
+
+
+def _write(steamapps: Path, app_id: int, body: str) -> Path:
+    steamapps.mkdir(parents=True, exist_ok=True)
+    manifest = steamapps / f"appmanifest_{app_id}.acf"
+    manifest.write_text(body, encoding="utf-8")
+    return manifest
+
+
+class TestManifestTransferActive:
+    def test_downloading_is_active(self, tmp_path: Path) -> None:
+        assert _manifest_transfer_active(_write(tmp_path, 1, _DOWNLOADING)) is True
+
+    def test_staging_is_active(self, tmp_path: Path) -> None:
+        assert _manifest_transfer_active(_write(tmp_path, 1, _STAGING)) is True
+
+    def test_complete_is_inactive(self, tmp_path: Path) -> None:
+        assert _manifest_transfer_active(_write(tmp_path, 1, _COMPLETE)) is False
+
+    def test_idle_is_inactive(self, tmp_path: Path) -> None:
+        assert _manifest_transfer_active(_write(tmp_path, 1, _IDLE)) is False
+
+    def test_missing_fields_inactive(self, tmp_path: Path) -> None:
+        manifest = _write(tmp_path, 1, '"AppState"\n{\n\t"appid"\t\t"1"\n}\n')
+        assert _manifest_transfer_active(manifest) is False
+
+    def test_unreadable_manifest_inactive(self, tmp_path: Path) -> None:
+        # Never created -> read_text raises OSError -> treated as inactive.
+        assert _manifest_transfer_active(tmp_path / "appmanifest_gone.acf") is False
+
+
+class TestSteamUpdateInProgress:
+    def test_true_when_any_app_updating(self, tmp_path: Path) -> None:
+        steamapps = tmp_path / "steamapps"
+        _write(steamapps, 1, _IDLE)
+        _write(steamapps, 2, _DOWNLOADING)
+        with patch(f"{SS}.STEAMAPPS_PATH", steamapps):
+            assert steam_update_in_progress() is True
+
+    def test_false_when_all_idle(self, tmp_path: Path) -> None:
+        steamapps = tmp_path / "steamapps"
+        _write(steamapps, 1, _IDLE)
+        _write(steamapps, 2, _COMPLETE)
+        with patch(f"{SS}.STEAMAPPS_PATH", steamapps):
+            assert steam_update_in_progress() is False
+
+    def test_false_when_no_manifests(self, tmp_path: Path) -> None:
+        steamapps = tmp_path / "empty_steamapps"
+        steamapps.mkdir()
+        with patch(f"{SS}.STEAMAPPS_PATH", steamapps):
+            assert steam_update_in_progress() is False
+
+    def test_glob_oserror_is_false(self) -> None:
+        with patch(f"{SS}.STEAMAPPS_PATH") as fake_path:
+            fake_path.glob.side_effect = OSError("boom")
+            assert steam_update_in_progress() is False
+
+
+class TestEnsureSteamDebugPortDefersUpdate:
+    def test_defers_and_does_not_shut_down(self) -> None:
+        with (
+            patch(f"{LH}.steam_is_installed", return_value=True),
+            patch(f"{LH}._steam_has_debug_port", return_value=False),
+            patch(f"{LH}._is_steam_running", return_value=True),
+            patch(f"{LH}.steam_update_in_progress", return_value=True),
+            patch(f"{LH}._shutdown_steam") as shutdown,
+            patch(f"{LH}._launch_steam_with_debug") as launch,
+        ):
+            with pytest.raises(SteamUpdateInProgressError):
+                ensure_steam_debug_port()
+            shutdown.assert_not_called()
+            launch.assert_not_called()
+
+
+class TestRestartSteamGuard:
+    def test_skips_when_update_in_progress(self) -> None:
+        with (
+            patch(f"{LH}.steam_update_in_progress", return_value=True),
+            patch(f"{LH}._shutdown_steam") as shutdown,
+            patch(f"{LH}._launch_steam_with_debug") as launch,
+        ):
+            restart_steam()
+            shutdown.assert_not_called()
+            launch.assert_not_called()
+
+    def test_restarts_when_no_update(self) -> None:
+        with (
+            patch(f"{LH}.steam_update_in_progress", return_value=False),
+            patch(f"{LH}._shutdown_steam") as shutdown,
+            patch(f"{LH}._launch_steam_with_debug") as launch,
+            patch(f"{LH}._wait_for_cdp_ready", return_value=True),
+        ):
+            restart_steam()
+            shutdown.assert_called_once()
+            launch.assert_called_once()