fix(steam_backlog_enforcer): reload state in enforce loop, use steam:// protocol for installs

The enforce daemon loaded state once at startup and never reloaded it.
When the CLI reassigned a game (e.g. via 'done'), the daemon kept
enforcing the old assignment and deleted the newly assigned game every
3 seconds as 'unauthorized'.

Fix: reload state from disk at the top of each enforce loop iteration
so CLI changes take effect within one cycle.

Also add steam://install protocol handler for interactive installs
(via xdg-open) so Steam determines the correct installdir from its
own metadata, avoiding 'Missing game executable' errors from guessed
directory names in fabricated appmanifests.
This commit is contained in:
Krzysztof kuhy Rudnicki 2026-03-17 21:48:33 +01:00
parent 36064c848d
commit 6e3040ed84
3 changed files with 73 additions and 11 deletions

View File

@ -61,6 +61,26 @@ PROTECTED_APP_IDS = {
STEAMAPPS_PATH = Path("~/.local/share/Steam/steamapps").expanduser() 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.
Returns True if the URI handler was invoked successfully.
"""
xdg_open = shutil.which("xdg-open") or "/usr/bin/xdg-open"
try:
subprocess.run(
[xdg_open, f"steam://install/{app_id}"],
capture_output=True,
timeout=15,
check=False,
)
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
return False
else:
logger.info("Triggered Steam install for %s via protocol handler", label)
return True
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# Game install management # Game install management
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
@ -142,22 +162,34 @@ def _ensure_steam_running() -> None:
logger.exception("Steam executable not found") logger.exception("Steam executable not found")
def install_game(app_id: int, game_name: str, steam_id: str) -> bool: def install_game(
"""Install a game by writing an appmanifest that triggers Steam's download. app_id: int,
game_name: str,
steam_id: str,
*,
use_steam_protocol: bool = False,
) -> bool:
"""Install a game by triggering a Steam download.
Creates a minimal appmanifest with StateFlags=1026 (UpdateRequired | When *use_steam_protocol* is True the ``steam://install`` URI handler
UpdateStarted) in the steamapps directory. The running Steam client is used, which lets Steam determine the correct install directory from
detects the new manifest and automatically queues the download no its own metadata. This avoids mismatches between the display name and
dialog or user interaction required. the canonical ``installdir`` that can cause "Missing game executable"
errors. Falls back to writing a fabricated appmanifest if the URI
handler is unavailable.
If Steam is not running it will be started in silent mode first. When *use_steam_protocol* is False (the default) a minimal
appmanifest with StateFlags=1026 is written directly. This is
suitable for non-interactive / daemon contexts where opening a Steam
dialog is undesirable.
Args: Args:
app_id: Steam application ID. app_id: Steam application ID.
game_name: Human-readable game name. game_name: Human-readable game name.
steam_id: Steam64 ID of the account that owns the game. steam_id: Steam64 ID of the account that owns the game.
use_steam_protocol: Prefer the ``steam://install`` URI handler.
Returns True if the manifest was written successfully. Returns True if the install was triggered successfully.
""" """
label = game_name or f"AppID={app_id}" label = game_name or f"AppID={app_id}"
@ -165,6 +197,12 @@ def install_game(app_id: int, game_name: str, steam_id: str) -> bool:
logger.info("Game already installed: %s", label) logger.info("Game already installed: %s", label)
return True return True
if use_steam_protocol:
_ensure_steam_running()
if _trigger_steam_install(app_id, label):
return True
logger.debug("steam:// protocol failed; falling back to manifest")
# Build a minimal appmanifest. StateFlags 1026 = UpdateRequired (2) + # Build a minimal appmanifest. StateFlags 1026 = UpdateRequired (2) +
# UpdateStarted (1024), which tells Steam "this app needs downloading". # UpdateStarted (1024), which tells Steam "this app needs downloading".
manifest_content = ( manifest_content = (

View File

@ -236,7 +236,12 @@ def cmd_install(config: Config, state: State) -> None:
return return
_echo(f"Installing {state.current_game_name} (AppID={state.current_app_id})...") _echo(f"Installing {state.current_game_name} (AppID={state.current_app_id})...")
if install_game(state.current_app_id, state.current_game_name, config.steam_id): if install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
):
_echo("Done!") _echo("Done!")
else: else:
_echo("Failed to create install manifest.") _echo("Failed to create install manifest.")
@ -382,6 +387,7 @@ def _enforce_on_done(config: Config, state: State) -> None:
state.current_app_id, state.current_app_id,
state.current_game_name, state.current_game_name,
config.steam_id, config.steam_id,
use_steam_protocol=True,
) )

View File

@ -199,7 +199,12 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None:
if not is_game_installed(chosen.app_id): if not is_game_installed(chosen.app_id):
_echo(f"\n Auto-installing {chosen.name}...") _echo(f"\n Auto-installing {chosen.name}...")
install_game(chosen.app_id, chosen.name, config.steam_id) install_game(
chosen.app_id,
chosen.name,
config.steam_id,
use_steam_protocol=True,
)
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
@ -404,7 +409,12 @@ def _enforce_auto_install(config: Config, state: State) -> None:
return return
if not is_game_installed(app_id): if not is_game_installed(app_id):
_echo(f" Auto-installing {state.current_game_name}...") _echo(f" Auto-installing {state.current_game_name}...")
if install_game(app_id, state.current_game_name, config.steam_id): if install_game(
app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
):
send_notification( send_notification(
"Game Installing", "Game Installing",
f"{state.current_game_name} is being downloaded.", f"{state.current_game_name} is being downloaded.",
@ -495,6 +505,14 @@ def do_enforce(config: Config, state: State) -> None:
_echo(" Press Ctrl+C to stop.\n") _echo(" Press Ctrl+C to stop.\n")
try: try:
while True: while True:
# Reload state from disk so CLI changes (e.g. new game
# assignment via ``done`` / ``scan``) take effect immediately
# without needing to restart the daemon.
fresh = State.load()
state.current_app_id = fresh.current_app_id
state.current_game_name = fresh.current_game_name
state.finished_app_ids = fresh.finished_app_ids
_enforce_loop_iteration(config, state) _enforce_loop_iteration(config, state)
time.sleep(ENFORCE_INTERVAL) time.sleep(ENFORCE_INTERVAL)
except KeyboardInterrupt: except KeyboardInterrupt: