kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-10 19:07:05 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Release a game after one new achievement instead of 100% completion

Browse Source 
Requiring 100% locked the user onto one game for weeks, which proved
ineffective. The assignment and every manual pick now release as soon as one
achievement is unlocked after the game was assigned (Steam unlock_time vs the
pick's started_at / State.current_assigned_at).

- finished_app_ids still means 100% complete; a game released below 100% gets
  released_at plus a 7-day cooldown and returns to the pool.
- Next pick: least recently assigned first, then shortest HLTB time.
- Releasing never evicts: released/expired picks stay installed and playable
  until the user explicitly chooses another game (new pick-manual, or
  accepting the next assignment); with no replacement the current game stays.
- Tampering detection judges a released game by unlocks after its release.
- The daemon reloads every State field each tick so its sweep save cannot
  overwrite cooldowns or assignment times written by the CLI.
- State.load backfills assignment times once and persists them; load-time
  migrations moved to _state_migrations.py.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PhzaTAWtRfJk8gre6NLSka
This commit is contained in:
Krzysztof Rudnicki 2026-09-26 16:46:03 +02:00
parent bbc3853c67
commit a191c8e459
27 changed files with 1025 additions and 250 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
README.md
View File

@ -1,10 +1,16 @@
# Steam Backlog Enforcer
Forces you to 100% complete one Steam game at a time before moving on.
Forces you to play one Steam game at a time: earn at least one new
achievement in the assigned game before moving on to the next.
Requiring 100% completion kept you locked onto one game for weeks, which proved
ineffective. A game released after one new achievement stays in the backlog
(it is not counted as finished) and comes round again later.
## Features
- **Achievement tracking**: Picks the next game by shortest HLTB completionist time
- **Achievement tracking**: Picks the least-recently-assigned game next, then the
shortest HLTB completionist time; a released game sits out a 7-day cooldown
- **Store blocking**: Blocks `store.steampowered.com` via `/etc/hosts`
- **Game uninstalling**: Removes all installed games except the assigned one
- **Process enforcement**: Kills unauthorized game processes
@ -22,7 +28,7 @@ python -m python_pkg.steam_backlog_enforcer.main setup
| Command | Description |
| ----------- | ------------------------------------------ |
| `scan` | Scan library, fetch HLTB data, assign game |
| `check` | Check if assigned game is complete |
| `check` | Check assigned game for a new achievement |
| `status` | Show current assignment and blocking |
| `list` | List incomplete games from snapshot |
| `skip` | Skip the currently assigned game |

22
steam_backlog_enforcer/_actions.py
View File

@ -15,7 +15,9 @@ from typing import TYPE_CHECKING, Any
from steam_backlog_enforcer._allowed_games import (
active_manual_picks,
allowed_games,
drop_inactive_picks,
)
from steam_backlog_enforcer._assignment_progress import record_assignment
# Marks these two as an intentional re-export (they're imported from
# _allowed_games above, not defined here) -- without this, mypy's
@ -51,7 +53,7 @@ def is_manual_pick_locked(state: State) -> bool:
"""Return ``True`` if any manual pick is currently holding the lock.
With several picks allowed at once the lock releases only when every one
of them is finished or expired.
of them is released (one new achievement), finished or expired.
Args:
state: The loaded enforcer state.
@ -118,8 +120,8 @@ def apply_manual_pick(
its interactive ``YES`` confirmation. Keeping this state-only means an
automated caller (the MCP server) can never wipe installed games.
Finished and expired entries are dropped on the way through so the stored
list does not grow without bound.
Picking is an explicit choice of another game, so released, finished and
expired picks are dropped here (and enforcement may then remove them).
Args:
state: The enforcer state to mutate and save.
@ -144,14 +146,10 @@ def apply_manual_pick(
)
now = datetime.now(UTC).isoformat()
# Rewriting from `active` also prunes finished/expired entries.
state.manual_picks = [
*active,
{"app_id": app_id, "game_name": game_name, "started_at": now},
]
state.current_app_id = app_id
state.current_game_name = game_name
if not state.enforcement_started_at:
state.enforcement_started_at = now
drop_inactive_picks(state)
state.manual_picks.append(
{"app_id": app_id, "game_name": game_name, "started_at": now}
)
record_assignment(state, app_id, game_name, at=now)
state.save()
return None

43
steam_backlog_enforcer/_allowed_games.py
View File

@ -21,10 +21,15 @@ MANUAL_LOCK_DAYS = 14
def _pick_is_active(state: State, pick: dict[str, Any]) -> bool:
"""Return ``True`` if *pick* still holds the lock.
"""Return ``True`` if *pick* still holds the lock and a pick slot.
A pick stops being active once its game is finished (100% achievements) or
once ``MANUAL_LOCK_DAYS`` have elapsed. A missing or malformed timestamp
An inactive pick is *not* evicted: it stays in ``state.manual_picks``, and
so installed and playable, until the user explicitly chooses another game
(:func:`drop_inactive_picks`).
A pick stops being active once it earned a new achievement since it was
picked (``released_at``), once its game is 100% finished, or once
``MANUAL_LOCK_DAYS`` have elapsed. A missing or malformed timestamp
keeps it active: with no deadline to evaluate, the safe answer for an
enforcement tool is "still locked".
@ -38,6 +43,8 @@ def _pick_is_active(state: State, pick: dict[str, Any]) -> bool:
app_id = pick.get("app_id")
if app_id is None or app_id in state.finished_app_ids:
return False
if pick.get("released_at"):
return False
started_at = pick.get("started_at") or ""
if started_at:
@ -58,15 +65,39 @@ def active_manual_picks(state: State) -> list[dict[str, Any]]:
state: The loaded enforcer state.
Returns:
The subset of ``state.manual_picks`` that is neither finished nor past
its own deadline.
The subset of ``state.manual_picks`` that is neither released,
finished nor past its own deadline.
"""
return [p for p in state.manual_picks if _pick_is_active(state, p)]
def drop_inactive_picks(state: State) -> list[dict[str, Any]]:
"""Forget every pick that no longer holds the lock; return the dropped ones.
Released, finished and expired picks stay installed and playable for as
long as the user wants. Only an explicit choice of another game (a new
manual pick, or accepting the next assignment) calls this and so lets
enforcement remove them. Does not save.
Args:
state: The enforcer state to mutate.
Returns:
The dropped pick entries, oldest first.
"""
active = active_manual_picks(state)
dropped = [p for p in state.manual_picks if p not in active]
state.manual_picks = active
return dropped
def allowed_games(state: State) -> list[tuple[int, str]]:
"""Return ``(app_id, name)`` for every game the enforcer must keep.
That is the current assignment plus every stored manual pick, including
released and expired ones: a game is only removed once the user has
explicitly chosen something else.
Args:
state: The loaded enforcer state.
@ -77,7 +108,7 @@ def allowed_games(state: State) -> list[tuple[int, str]]:
games: list[tuple[int, str]] = []
if state.current_app_id is not None:
games.append((state.current_app_id, state.current_game_name))
for pick in active_manual_picks(state):
for pick in state.manual_picks:
app_id = pick.get("app_id")
if app_id is not None and all(app_id != aid for aid, _ in games):
games.append((app_id, pick.get("game_name", "")))

162
steam_backlog_enforcer/_assignment_progress.py Normal file
View File

@ -0,0 +1,162 @@
"""The "earn one new achievement" release rule, and the bookkeeping behind it.
A game (the current assignment or a manual pick) no longer has to reach 100%
achievements before the user may move on: one achievement unlocked after the
game was assigned is enough. Requiring 100% kept the user stuck on one game for
weeks, which proved ineffective.
``State.finished_app_ids`` still means *100% complete* and nothing else, so the
stats built on it stay honest. A game released below 100% goes back into the
pool behind a cooldown and is ordered least-recently-assigned-first, so it
comes round again only after the rest of the backlog has had a turn.
Leaf, stdout-free module: safe for the MCP server and importable from
``config.py`` without a cycle.
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from steam_backlog_enforcer._allowed_games import active_manual_picks
if TYPE_CHECKING:
from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.steam_api import AchievementInfo, GameInfo
# Days a released game stays out of the auto-assignment pool; matches the
# "n = skip" answer of the done prompt.
RELEASE_COOLDOWN_DAYS = 7
def _now_iso() -> str:
return datetime.now(UTC).isoformat()
def iso_to_epoch(iso: str) -> float | None:
"""Return *iso* as epoch seconds, or ``None`` when empty or malformed."""
if not iso:
return None
try:
return datetime.fromisoformat(iso).timestamp()
except ValueError:
return None
def record_assignment(
state: State, app_id: int, game_name: str, at: str | None = None
) -> None:
"""Make *app_id* the current assignment, stamping when it happened.
The single writer of ``current_assigned_at`` and ``last_assigned_at``, so
every assignment path starts the "new achievement" clock the same way.
Does not save.
Args:
state: The enforcer state to mutate.
app_id: The newly assigned game.
game_name: Its display name.
at: ISO timestamp of the assignment; defaults to now.
"""
stamp = at or _now_iso()
state.current_app_id = app_id
state.current_game_name = game_name
state.current_assigned_at = stamp
state.last_assigned_at[str(app_id)] = stamp
if not state.enforcement_started_at:
state.enforcement_started_at = stamp
def last_assigned_epoch(state: State, app_id: int) -> float:
"""Return when *app_id* was last assigned, ``0.0`` if never (sorts first)."""
return iso_to_epoch(state.last_assigned_at.get(str(app_id), "")) or 0.0
def assignment_baseline(state: State, app_id: int) -> str:
"""Return the ISO time after which a new achievement on *app_id* counts.
An active manual pick counts from its own ``started_at`` (a released or
expired one must not: its old unlocks would release a later assignment of
the same game at once); otherwise the current
assignment counts from ``current_assigned_at``. Anything else has no
baseline (``""``), which never releases.
"""
for pick in active_manual_picks(state):
if pick.get("app_id") == app_id and pick.get("started_at"):
return str(pick["started_at"])
if app_id == state.current_app_id:
return state.current_assigned_at
return ""
def newest_unlock_since(game: GameInfo, since_iso: str) -> AchievementInfo | None:
"""Return the most recent achievement unlocked after *since_iso*, if any.
An empty or malformed baseline returns ``None``: for an enforcement tool,
an unknown start must never read as progress.
"""
since = iso_to_epoch(since_iso)
if since is None:
return None
fresh = [a for a in game.achievements if a.achieved and a.unlock_time > since]
return max(fresh, key=lambda a: a.unlock_time) if fresh else None
def release_verdict(state: State, game: GameInfo) -> tuple[bool, str]:
"""Judge *game* against the release rule: 100%, or one new achievement.
Args:
state: The enforcer state (for the assignment baseline).
game: Freshly refreshed achievement data for the game.
Returns:
Whether the user may move on, and a one-line reason for the CLI.
"""
baseline = assignment_baseline(state, game.app_id)
newest = newest_unlock_since(game, baseline)
if game.is_complete:
return True, f"COMPLETED: {game.name} (100% achievements)!"
if newest is not None:
return True, f"NEW ACHIEVEMENT on {game.name}: '{newest.display_name}'!"
since = baseline[:10] if baseline else "its assignment"
return False, (
f"No new achievement since {since}. Earn at least one to move on. Keep going!"
)
def mark_finished(state: State, app_id: int) -> bool:
"""Record *app_id* as 100% finished, without duplicating an existing record.
The single place a completion is written, so the manual-pick sweep and the
``done``/``check`` paths cannot both append the same id.
Args:
state: The enforcer state to mutate (not saved here).
app_id: The completed app id.
Returns:
Whether this call actually added the id.
"""
if app_id in state.finished_app_ids:
return False
state.finished_app_ids.append(app_id)
return True
def release_game(state: State, game: GameInfo) -> None:
"""Let the user move on from *game*, which has met the release rule.
Marks it finished only when it really is at 100%, frees any manual-pick
slot it holds, remembers when it was released (for tampering detection),
and puts it on the cooldown so the next pick is a different game. Does not
save and does not reassign.
"""
now = _now_iso()
if game.is_complete:
mark_finished(state, game.app_id)
for pick in state.manual_picks:
if pick.get("app_id") == game.app_id and not pick.get("released_at"):
pick["released_at"] = now
state.released_at[str(game.app_id)] = now
state.skip_for_days(game.app_id, RELEASE_COOLDOWN_DAYS)

48
steam_backlog_enforcer/_cmd_done_finalize.py
View File

@ -1,4 +1,4 @@
"""Finalising a completed game and re-enforcing afterwards.
"""Moving on from a game that earned a new achievement, or re-enforcing.
Split out of :mod:`steam_backlog_enforcer._cmd_done` to keep both files
under the 250-line cap.
@ -10,6 +10,10 @@ import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._assignment_progress import (
release_game,
release_verdict,
)
from steam_backlog_enforcer._cmd_done import (
_apply_cached_hours_to_games,
_prompt_keep_or_skip,
@ -18,7 +22,6 @@ from steam_backlog_enforcer._cmd_done import (
)
from steam_backlog_enforcer._hltb_cached import fetch_hltb_times_cached
from steam_backlog_enforcer._hltb_types import load_hltb_cache
from steam_backlog_enforcer._pick_completion import mark_finished
from steam_backlog_enforcer._snapshot import load_snapshot
from steam_backlog_enforcer._steam_state import is_game_fully_installed
from steam_backlog_enforcer.enforcer import (
@ -45,19 +48,20 @@ logger = logging.getLogger(__name__)
def _finalize_completion(
config: Config,
state: State,
game_name: str,
app_id: int,
game: GameInfo,
) -> None:
"""Mark game complete, pick next, hide non-assigned games, notify."""
_echo(f"\n COMPLETED: {game_name}!")
mark_finished(state, app_id)
"""Release the game, pick next, hide non-assigned games, notify.
The release (and its cooldown) is recorded before the next pick so the
game just played cannot be handed straight back.
"""
game_name = game.name
release_game(state, game)
snapshot_data = load_snapshot()
_echo("\nPicking next game...")
if not snapshot_data:
_echo(" No snapshot found. Run 'scan' first.")
state.current_app_id = None
state.current_game_name = ""
state.save()
return
@ -68,8 +72,9 @@ def _finalize_completion(
_apply_cached_hours_to_games(games, hltb_cache)
pick_next_game(games, state, config, on_select=_prompt_keep_or_skip)
if state.current_app_id is None:
_echo(" No more games to assign!")
if state.current_app_id in {None, game.app_id}:
# Nothing new was accepted, so the released game stays assigned.
_echo(f" No new game assigned; {game_name} stays playable.")
return
hide_others_and_report(config, state)
@ -92,8 +97,8 @@ def _finalize_completion(
)
send_notification(
"Game Complete!",
f"Finished {game_name}! Now playing: {state.current_game_name}",
"Moving on!",
f"New achievement in {game_name}! Now playing: {state.current_game_name}",
)
_echo(f"\nAll done! Go play {state.current_game_name}!")
@ -136,11 +141,12 @@ def _enforce_on_done(config: Config, state: State) -> None:
def cmd_done(config: Config, state: State) -> None:
"""Check completion, pick next game, uninstall & hide.
"""Check for a new achievement, pick next game, uninstall & hide.
All-in-one command for after finishing a game:
1. Verify 100% achievements on Steam.
2. Pick the next game (shortest HLTB leisure+dlc time).
All-in-one command for after making progress in a game:
1. Verify on Steam that at least one achievement was unlocked since the
game was assigned (or that it is at 100%).
2. Pick the next game (least recently assigned, then shortest HLTB time).
3. Uninstall all non-assigned games.
4. Hide all non-assigned games in the Steam library.
5. Install the newly assigned game.
@ -173,10 +179,10 @@ def cmd_done(config: Config, state: State) -> None:
_echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours")
_report_assigned_confidence(app_id, state)
if not game.is_complete:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!")
released, verdict = release_verdict(state, game)
_echo(f"\n {verdict}")
if not released:
_enforce_on_done(config, state)
return
_finalize_completion(config, state, game_name, app_id)
_finalize_completion(config, state, game)

26
steam_backlog_enforcer/_enforce_loop.py
View File

@ -17,7 +17,6 @@ from steam_backlog_enforcer._enforce_steps import (
# from here, so splitting it into _owned_apps_cache stays invisible to them.
from steam_backlog_enforcer._owned_apps_cache import get_all_owned_app_ids
from steam_backlog_enforcer._pick_completion import (
daemon_sweep,
retire_completed_manual_picks_throttled,
)
from steam_backlog_enforcer._playtime import playtime_tick
@ -141,16 +140,13 @@ def _enforce_loop_iteration(
if not steam_is_installed():
return
# Retire manual picks that have hit 100%, at most once every
# Release manual picks that earned a new achievement, at most once every
# MANUAL_PICK_RECHECK_TTL_SECONDS - this loop ticks every 3s.
retire_completed_manual_picks_throttled(config, state)
# Record, but do not evict. A pick the daemon just retired is no longer in
# allowed_app_ids, so steps A and B below would kill the running process
# and uninstall the game - possibly seconds after the final achievement
# popped, mid-session. Keeping those ids allowed defers eviction to a
# user-invoked done/check/pick-manual, which is where it was before.
allowed = allowed_app_ids(state) | daemon_sweep.retired
# Releasing never evicts: a released pick stays in allowed_app_ids until
# the user explicitly chooses another game, so this is safe mid-session.
allowed = allowed_app_ids(state)
if not allowed:
return
@ -232,13 +228,13 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
logger.warning("Failed to reload state: %s", exc)
time.sleep(ENFORCE_INTERVAL)
continue
state.current_app_id = fresh.current_app_id
state.current_game_name = fresh.current_game_name
state.finished_app_ids = fresh.finished_app_ids
# Manual picks too: the MCP pick_manual tool adds a *second* pick
# without touching current_app_id, so a daemon that never reloaded
# this list would uninstall that pick as unauthorized.
state.manual_picks = fresh.manual_picks
# Every field, not a hand-picked few: the daemon's pick sweep saves
# this object, and any field left stale here (cooldowns, pick
# release times, assignment times) would overwrite what the CLI
# or the MCP server wrote. The MCP pick_manual tool also adds a
# *second* pick without touching current_app_id, which a daemon
# that never reloaded manual_picks would uninstall.
vars(state).update(vars(fresh))
_enforce_loop_iteration(config, state, session=session, demo=demo)
time.sleep(ENFORCE_INTERVAL)

27
steam_backlog_enforcer/_manual_pick_lifecycle.py
View File

@ -17,6 +17,7 @@ from steam_backlog_enforcer._allowed_games import (
active_manual_picks,
allowed_games,
)
from steam_backlog_enforcer._assignment_progress import record_assignment
from steam_backlog_enforcer._total_block import get_total_block_status
from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app
from steam_backlog_enforcer.store_blocker import is_store_blocked
@ -67,7 +68,8 @@ def manual_pick_age_days(state: State, app_id: int) -> float | None:
def abandon_manual_pick(state: State, app_id: int) -> bool:
"""Drop one manual pick and persist ``state``.
A pick may be abandoned at any time, however long ago it was made.
A pick may be abandoned at any time, however long ago it was made, and
also after it was released or expired (it is still installed then).
Only the named pick is dropped: any other active pick keeps its own lock
and deadline. The abandoned app id goes onto the existing skip cooldown so
@ -83,21 +85,30 @@ def abandon_manual_pick(state: State, app_id: int) -> bool:
``True`` if the pick was abandoned, ``False`` if it is not an active
pick, in which case ``state`` is untouched.
"""
if find_manual_pick(state, app_id) is None:
if all(p.get("app_id") != app_id for p in state.manual_picks):
return False
state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS)
state.manual_picks = [
p for p in active_manual_picks(state) if p.get("app_id") != app_id
]
state.manual_picks = [p for p in state.manual_picks if p.get("app_id") != app_id]
# The abandoned pick may also have been the current assignment; hand the
# assignment to a surviving pick so the enforcer keeps guarding it, or
# clear it so 'scan' can reassign.
if state.current_app_id == app_id:
survivor = state.manual_picks[-1] if state.manual_picks else None
state.current_app_id = survivor["app_id"] if survivor else None
state.current_game_name = survivor["game_name"] if survivor else ""
survivors = active_manual_picks(state) or state.manual_picks
survivor = survivors[-1] if survivors else None
if survivor is None:
state.current_app_id = None
state.current_game_name = ""
state.current_assigned_at = ""
else:
# The survivor has been assigned since it was picked.
record_assignment(
state,
survivor["app_id"],
survivor["game_name"],
at=survivor.get("started_at") or None,
)
state.save()
return True

13
steam_backlog_enforcer/_mcp_actions.py
View File

@ -39,9 +39,9 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]:
With ``confirm=False`` (the default) this performs **no** mutation and
returns a preview of what confirming would do. Call again with
``confirm=True`` to apply. Applying overrides the current assignment and
locks all other commands for ``MANUAL_LOCK_DAYS`` days (or until 100%
achievements). Unlike the CLI's ``pick-manual``, this mutates **state only**
— it never uninstalls, installs, or hides games.
locks all other commands for ``MANUAL_LOCK_DAYS`` days (or until one new
achievement is earned in it). Unlike the CLI's ``pick-manual``, this
mutates **state only** — it never uninstalls, installs, or hides games.
Args:
app_id: The Steam app id to lock in (must exist in the last snapshot).
@ -62,14 +62,15 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]:
"game_name": game_name,
"effect": (
"Overrides the current assignment and locks all other commands "
f"for {MANUAL_LOCK_DAYS} days or until 100% achievements."
f"for {MANUAL_LOCK_DAYS} days or until you earn one new "
"achievement in it."
),
"confirm_required": True,
}
config = Config.load()
state = State.load()
# Free the slots of picks already at 100%, so the cap below cannot refuse
# on behalf of a game that is finished. Matches the CLI's pick-manual.
# Free the slots of picks that already earned a new achievement, so the
# cap below cannot refuse on behalf of them. Matches the CLI's pick-manual.
retired = retire_completed_manual_picks(config, state)
refused = apply_configured_pick(config, state, app_id, game_name)
if refused is not None:

104
steam_backlog_enforcer/_pick_completion.py
View File

@ -1,15 +1,17 @@
"""Achievement-based retirement of finished manual picks.
"""Achievement-based retirement of manual picks.
A manual pick stops holding its slot once its app id is in
``State.finished_app_ids`` -- that is what ``_allowed_games._pick_is_active``
reads. Historically only ``State.current_app_id`` was ever checked for
completion (``cmd_done``, ``do_check``), so a manual pick that was *not* the
current assignment had no achievement-based release path at all: it sat on a
slot until the 14-day lock expired or the user abandoned it by hand.
A manual pick stops holding its slot once it earned one achievement since it
was picked (``released_at`` on the pick) or reached 100% -- that is what
``_allowed_games._pick_is_active`` reads. Historically only
``State.current_app_id`` was ever checked (``cmd_done``, ``do_check``), so a
manual pick that was *not* the current assignment had no achievement-based
release path at all: it sat on a slot until the 14-day lock expired or the
user abandoned it by hand.
This module closes that gap by checking every active pick and retiring the ones
at 100%. It deliberately stops there -- freeing a slot is not a request for a
replacement game, so nothing here calls ``pick_next_game``.
that met the release rule (:mod:`._assignment_progress`). It deliberately stops
there -- freeing a slot is not a request for a replacement game, so nothing
here calls ``pick_next_game``.
"""
from __future__ import annotations
@ -20,6 +22,11 @@ import time
from typing import TYPE_CHECKING
from steam_backlog_enforcer._allowed_games import active_manual_picks
from steam_backlog_enforcer._assignment_progress import (
assignment_baseline,
newest_unlock_since,
release_game,
)
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.steam_api import SteamAPIClient
@ -35,21 +42,18 @@ logger = logging.getLogger(__name__)
class _DaemonSweep:
"""The enforce daemon's re-check gate, plus what it has retired.
"""The enforce daemon's re-check gate.
An object rather than module-level values because rebinding those would
An object rather than a module-level value because rebinding that would
need a ``global`` statement; attribute writes do not.
``retired`` exists so the daemon can *record* a completion without
*evicting* the game: retiring drops a pick out of ``allowed_app_ids``, and
the daemon would then kill and uninstall it mid-session, seconds after the
last achievement popped. Unioning these ids back into the allowed set
defers that to a user-invoked command; the freed slot is visible at once.
Releasing a pick never evicts it (it stays in ``allowed_app_ids`` until
the user explicitly chooses another game), so the daemon may release
mid-session without killing the game.
"""
def __init__(self) -> None:
self.last: float | None = None
self.retired: set[int] = set()
def due(self, now: float, ttl: float) -> bool:
"""Return whether a check is due, recording *now* when it is.
@ -78,7 +82,8 @@ class PickProgress:
``determinable`` is ``False`` when Steam returned nothing for this app (no
achievements, or the call failed); ``unlocked``/``total`` are 0 then, and
callers must read it as "unknown", never as "complete".
callers must read it as "unknown", never as "released". ``new_achievement``
names the unlock that released it below 100%.
"""
app_id: int
@ -87,40 +92,20 @@ class PickProgress:
total: int
retired: bool
determinable: bool
new_achievement: str = ""
def describe(self) -> str:
"""Return a one-line human summary for CLI output."""
if not self.determinable:
return f" {self.game_name}: progress unavailable (no achievements)"
if self.retired:
return (
f" {self.game_name}: {self.unlocked}/{self.total} (100%)"
" - COMPLETE, freeing its slot"
)
pct = (self.unlocked / self.total) * 100.0 if self.total else 0.0
return (
f" {self.game_name}: {self.unlocked}/{self.total}"
f" ({pct:.0f}%) - still in progress"
)
def mark_finished(state: State, app_id: int) -> bool:
"""Record *app_id* as finished, without duplicating an existing record.
The single place a completion is written, so the manual-pick sweep and the
``done``/``check`` paths cannot both append the same id.
Args:
state: The enforcer state to mutate (not saved here).
app_id: The completed app id.
Returns:
Whether this call actually added the id.
"""
if app_id in state.finished_app_ids:
return False
state.finished_app_ids.append(app_id)
return True
progress = f" {self.game_name}: {self.unlocked}/{self.total} ({pct:.0f}%)"
kept = "slot freed, stays playable until you choose another game"
if self.retired and self.unlocked >= self.total:
return f"{progress} - COMPLETE, {kept}"
if self.retired:
return f"{progress} - earned '{self.new_achievement}', {kept}"
return f"{progress} - no new achievement since it was picked yet"
def retire_completed_manual_picks(
@ -129,14 +114,14 @@ def retire_completed_manual_picks(
*,
client: SteamAPIClient | None = None,
) -> list[PickProgress]:
"""Mark every 100%-complete manual pick finished, freeing its slot.
"""Release every manual pick that earned a new achievement, freeing its slot.
Mutates and saves ``state`` only when something was actually retired. Does
**not** assign a replacement game, uninstall anything, or touch
``current_app_id``: the caller decides what a freed slot means.
A pick whose achievements cannot be read stays locked: for an enforcement
tool, Steam being unreachable must never look like completion.
tool, Steam being unreachable must never look like progress.
Args:
config: Enforcer configuration (for the Steam credentials).
@ -168,17 +153,21 @@ def retire_completed_manual_picks(
)
continue
if game.is_complete and mark_finished(state, app_id):
newest = newest_unlock_since(game, assignment_baseline(state, app_id))
released = game.is_complete or newest is not None
if released:
release_game(state, game)
retired_any = True
logger.info("Manual pick retired at 100%%: %s (AppID=%s)", name, app_id)
logger.info("Manual pick released: %s (AppID=%s)", name, app_id)
results.append(
PickProgress(
app_id,
name,
game.unlocked_achievements,
game.total_achievements,
retired=game.is_complete,
retired=released,
determinable=True,
new_achievement=newest.display_name if newest else "",
)
)
@ -200,7 +189,7 @@ def report_completion(config: Config, state: State) -> list[PickProgress]:
results = retire_completed_manual_picks(config, state)
if not results:
return []
_echo("\nChecking your manual picks for completion...")
_echo("\nChecking your manual picks for a new achievement...")
for result in results:
_echo(result.describe())
return [r for r in results if r.retired]
@ -219,8 +208,8 @@ def warn_stale_assignment(state: State, retired: list[PickProgress]) -> None:
for pick in retired:
if pick.app_id == state.current_app_id:
_echo(
f"\nNote: {pick.game_name} is recorded as complete but is still"
"\n your current assignment. Run './run.sh done' to get"
f"\nNote: {pick.game_name} is released but is still your"
"\n current assignment. Run './run.sh done' to get"
"\n your next game."
)
return
@ -232,9 +221,6 @@ def retire_completed_manual_picks_throttled(
) -> list[PickProgress]:
"""Rate-limited :func:`retire_completed_manual_picks` for the enforce loop.
Records what it retired in :data:`daemon_sweep` so the caller can keep
those games in the allowed set rather than evicting them unattended.
Args:
config: Enforcer configuration.
state: The enforcer state to inspect, mutate and save.
@ -244,6 +230,4 @@ def retire_completed_manual_picks_throttled(
"""
if not daemon_sweep.due(time.monotonic(), MANUAL_PICK_RECHECK_TTL_SECONDS):
return []
results = retire_completed_manual_picks(config, state)
daemon_sweep.retired.update(r.app_id for r in results if r.retired)
return results
return retire_completed_manual_picks(config, state)

38
steam_backlog_enforcer/_scanning_assign.py
View File

@ -6,11 +6,15 @@ the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``.
from __future__ import annotations
from datetime import UTC, datetime
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._allowed_games import drop_inactive_picks
from steam_backlog_enforcer._assignment_progress import (
last_assigned_epoch,
record_assignment,
)
from steam_backlog_enforcer._scanning_candidates import (
_pick_next_shortest_candidate,
_sort_key,
@ -67,11 +71,13 @@ def _assign_chosen_game(
state: State,
config: Config,
) -> None:
"""Save assignment, announce it, and handle install/uninstall."""
state.current_app_id = chosen.app_id
state.current_game_name = chosen.name
if not state.enforcement_started_at:
state.enforcement_started_at = datetime.now(UTC).isoformat()
"""Save assignment, announce it, and handle install/uninstall.
Accepting a new game is the explicit choice that lets enforcement remove
released picks, so they are dropped here.
"""
drop_inactive_picks(state)
record_assignment(state, chosen.app_id, chosen.name)
state.save()
hours_str = (
f" (~{chosen.completionist_hours:.1f}h leisure+dlc)"
@ -101,10 +107,12 @@ def _assign_chosen_game(
def _clear_assignment(state: State, message: str) -> None:
"""Say why nothing could be assigned and leave the slot empty."""
"""Say why nothing could be assigned; keep the current game, if any.
With no replacement chosen, the game already assigned stays installed and
playable: only an explicit choice of another game may remove it.
"""
_echo(message)
state.current_app_id = None
state.current_game_name = ""
state.save()
@ -116,11 +124,19 @@ def _no_pick_message(confidence_skipped: int, linux_skipped: int) -> str:
def _open_candidates(games: list[GameInfo], state: State) -> list[GameInfo]:
"""Unfinished, unskipped games, shortest first, with cached confidence."""
"""Unfinished, unskipped games with cached confidence, in pick order.
Least-recently-assigned first (never-assigned games lead), then shortest
HLTB time. A game released after one new achievement is still unfinished,
so without the recency key the same short game would come straight back
and the user would be locked onto it again.
"""
skip = set(state.finished_app_ids) | state.active_skipped_ids()
candidates = [g for g in games if not g.is_complete and g.app_id not in skip]
if candidates:
candidates.sort(key=_sort_key)
candidates.sort(
key=lambda g: (last_assigned_epoch(state, g.app_id), *_sort_key(g))
)
_apply_cached_confidence_to_candidates(candidates)
return candidates

13
steam_backlog_enforcer/_scanning_tampering.py
View File

@ -10,6 +10,7 @@ import logging
from typing import TYPE_CHECKING, Any
from steam_backlog_enforcer._allowed_games import allowed_games
from steam_backlog_enforcer._assignment_progress import iso_to_epoch
from steam_backlog_enforcer._snapshot import load_snapshot
from steam_backlog_enforcer.enforcer import send_notification
from steam_backlog_enforcer.game_install import _echo
@ -72,8 +73,18 @@ def _check_game_tampering(
game = client.refresh_single_game(
app_id, entry["name"], entry.get("playtime_minutes", 0)
)
if game and game.unlocked_achievements > entry["unlocked_achievements"]:
if game is None:
return None
released = iso_to_epoch(state.released_at.get(str(app_id), ""))
if released is not None:
# Released below 100%: the snapshot predates the unlocks earned while
# it was assigned, so only unlocks after the release are suspicious.
diff = sum(
1 for a in game.achievements if a.achieved and a.unlock_time > released
)
else:
diff = game.unlocked_achievements - entry["unlocked_achievements"]
if diff > 0:
return (entry["name"], app_id, diff)
return None

84
steam_backlog_enforcer/_state_migrations.py Normal file
View File

@ -0,0 +1,84 @@
"""Load-time upgrades of an older ``state.json``.
Each function mutates a freshly loaded ``State`` in place and reports whether
it changed anything, so ``State.load`` can persist an upgrade exactly once.
Split out of :mod:`steam_backlog_enforcer.config` to keep it under the
250-line cap.
"""
from __future__ import annotations
from datetime import UTC, datetime
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from steam_backlog_enforcer.config import State
def migrate_legacy_manual_pick(state: State) -> bool:
"""Fold a pre-multi-pick single manual pick into ``manual_picks``.
A live lock written by the old single-slot code must survive the
upgrade, so the legacy fields are read once, converted, and cleared.
Args:
state: The loaded enforcer state to mutate.
Returns:
Whether a legacy pick was migrated.
"""
if state.manual_pick_app_id is None or state.manual_picks:
return False
state.manual_picks = [
{
"app_id": state.manual_pick_app_id,
"game_name": state.manual_pick_game_name,
"started_at": state.manual_pick_started_at,
}
]
state.manual_pick_app_id = None
state.manual_pick_game_name = ""
state.manual_pick_started_at = ""
return True
def backfill_assignment_times(state: State) -> bool:
"""Stamp assignments made before assignment times were recorded.
The current game gets "now": achievements earned before this upgrade
cannot be shown to come after the assignment, so they do not count.
Manual picks already carry ``started_at``, which is reused, including for
a current game that is also a pick.
Args:
state: The loaded enforcer state to mutate.
Returns:
Whether anything changed. The caller must persist it, or every load
would re-stamp "now" and no achievement could ever count.
"""
changed = False
pick_started = {
p["app_id"]: p["started_at"]
for p in state.manual_picks
if p.get("app_id") is not None and p.get("started_at")
}
for app_id, started_at in pick_started.items():
changed |= _stamp_if_missing(state, app_id, started_at)
current = state.current_app_id
if current is not None:
if not state.current_assigned_at:
# A current game that is also a pick was assigned when picked.
state.current_assigned_at = pick_started.get(current) or (
datetime.now(UTC).isoformat()
)
changed = True
changed |= _stamp_if_missing(state, current, state.current_assigned_at)
return changed
def _stamp_if_missing(state: State, app_id: int, stamp: str) -> bool:
if str(app_id) in state.last_assigned_at:
return False
state.last_assigned_at[str(app_id)] = stamp
return True

43
steam_backlog_enforcer/config.py
View File

@ -12,6 +12,11 @@ from pathlib import Path
import tempfile
from typing import Any
from steam_backlog_enforcer._state_migrations import (
backfill_assignment_times,
migrate_legacy_manual_pick,
)
CONFIG_DIR = Path.home() / ".config" / "steam_backlog_enforcer"
CONFIG_FILE = CONFIG_DIR / "config.json"
STATE_FILE = CONFIG_DIR / "state.json"
@ -166,12 +171,18 @@ class State:
blanked once migrated and are never written again.
"""
manual_picks: list[dict[str, Any]] = field(default_factory=list)
"""Active manual picks, newest last.
"""Manual picks, newest last; inactive ones stay until a new choice.
Each entry is ``{"app_id": int, "game_name": str, "started_at": iso}``.
Plain dicts rather than a dataclass because ``save`` serialises
``self.__dict__`` straight to JSON.
Each entry is ``{"app_id": int, "game_name": str, "started_at": iso}``,
plus ``"released_at": iso`` once it earned a new achievement. Plain dicts
because ``save`` serialises ``self.__dict__`` straight to JSON.
"""
current_assigned_at: str = ""
"""ISO time the current game was assigned; unlocks after it release it."""
last_assigned_at: dict[str, str] = field(default_factory=dict)
"""``str(app_id)`` → ISO time of its latest assignment (pick ordering)."""
released_at: dict[str, str] = field(default_factory=dict)
"""``str(app_id)`` → ISO time it was released below 100% (tampering)."""
def skip_for_days(self, app_id: int, days: int) -> None:
"""Mark ``app_id`` as skipped for ``days`` days from now (UTC)."""
@ -218,26 +229,8 @@ class State:
logger.warning("Corrupt state file, using defaults.")
return cls()
state = cls(**{k: v for k, v in data.items() if k in _field_names(cls)})
state._migrate_legacy_manual_pick()
migrated = migrate_legacy_manual_pick(state)
if backfill_assignment_times(state) or migrated:
state.save()
return state
return cls()
def _migrate_legacy_manual_pick(self) -> None:
"""Fold a pre-multi-pick single manual pick into ``manual_picks``.
A live lock written by the old single-slot code must survive the
upgrade, so the legacy fields are read once, converted, and cleared.
Migration happens in memory; the next ``save`` persists it.
"""
if self.manual_pick_app_id is None or self.manual_picks:
return
self.manual_picks = [
{
"app_id": self.manual_pick_app_id,
"game_name": self.manual_pick_game_name,
"started_at": self.manual_pick_started_at,
}
]
self.manual_pick_app_id = None
self.manual_pick_game_name = ""
self.manual_pick_started_at = ""

66
steam_backlog_enforcer/docs/DOCS-manual-pick-design.md
View File

@ -3,17 +3,17 @@ Picking a game should work like this:
1. user invokes script with specific flag for picking a game manually
2. user provides game steam id (in testing use 489830)
3. Script shows what game it believes this id means (in this case it should show The Elder Scrolls V: Skyrim Special Edition)
4. user confirms that this is the game they want to pick and confirm that they will not be able to use the script for up to 2 weeks or until the game is completed (100% achievements)
4. user confirms that this is the game they want to pick and confirm that they will not be able to use the script for up to 2 weeks or until they earn one new achievement in it (originally: 100% achievements; see "Release rule" below)
When user picks a game manually this should override the current pick if it exists
After picking manually backlog enforcer should make a note of that and very aggressively disallow user to do anything else
for a period of 2 weeks or until user completes a given game
for a period of 2 weeks or until user earns a new achievement in the given game
Logic should be as follows:
1. backlog checks if a user picked game manually
a. if not -> continue as before
2. if yes check if the game is completed
2. if yes check if the game earned a new achievement since it was picked
a. if yes -> continue as before
3. if NOT show info that user picked a specific game manually and they have to finish it before using ANY OTHER functionality of backlog enforcer
3. if NOT show info that user picked a specific game manually and they have to earn a new achievement in it before using ANY OTHER functionality of backlog enforcer
test the functionality with 489830 (The Elder Scrolls V: Skyrim Special Edition)
as always first write full functionality confirm that it works alone and with the user and only AFTER that write tests and coverage and fix issues
@ -43,7 +43,8 @@ the `pick-manual` warning mentions it up front.
`Config.max_manual_picks` (default 2) is how many games may be locked in at
once. All active picks stay installed, visible and un-killed; the lock
releases only when every pick is finished or past its own 14-day deadline.
releases only when every pick is released (one new achievement), finished or
past its own 14-day deadline.
- `State.manual_picks` is the list of `{app_id, game_name, started_at}`
entries. The old single-slot `manual_pick_*` fields are still read on load,
@ -58,31 +59,60 @@ releases only when every pick is finished or past its own 14-day deadline.
tears down an earlier one.
- `abandon-pick <app_id>` drops only that pick; a survivor inherits the
assignment and keeps the lock.
- `cmd_done` still finishes `current_app_id` and auto-picks a replacement, so
the remaining picks stay queued. A finished pick leaves the active set
automatically via `finished_app_ids` — no pruning needed — but *something*
has to put the id there in the first place; see the next point.
- **Completion is checked per pick, not just for the assignment**
(`_pick_completion.py`). `finished_app_ids` used to gain ids only from
- `cmd_done` still releases `current_app_id` and auto-picks a replacement, so
the remaining picks stay queued. A released pick leaves the active set
automatically via its `released_at` stamp (or `finished_app_ids` at 100%) —
no pruning needed — but *something* has to stamp it; see the next point.
- **The release rule is checked per pick, not just for the assignment**
(`_pick_completion.py`). Picks used to be released only from
`cmd_done`/`do_check`, and both look at `current_app_id` alone — so a pick
that was not the current assignment had no achievement-based release path
and sat on its slot until the 14-day expiry. `retire_completed_manual_picks`
now re-checks every active pick and records the ones at 100%, so the cap can
never refuse on behalf of a game that is already finished. It runs from
now re-checks every active pick and releases the ones with a new
achievement, so the cap can never refuse on behalf of one. It runs from
`pick-manual`, `status`, `check`, the MCP `pick_manual` tool, and the enforce
loop (throttled to once every `MANUAL_PICK_RECHECK_TTL_SECONDS`, since that
loop ticks every 3s).
- A pick whose achievements cannot be read — the game has none, or Steam is
unreachable — stays locked. Unknown must never look like complete.
unreachable — stays locked. Unknown must never look like progress.
- Retiring frees the slot and nothing more: it never calls `pick_next_game`,
so freeing a slot cannot hand you an unrequested extra game.
- In the daemon it records without evicting. A retired pick leaves
`allowed_app_ids`, which would otherwise make the loop kill and uninstall
the game mid-session, seconds after the last achievement popped; the ids in
`daemon_sweep.retired` stay allowed until a user-invoked command acts.
- Releasing never evicts. A released, finished or expired pick stays in
`State.manual_picks`, and therefore in `allowed_app_ids`, installed and
playable for as long as the user wants. Only an explicit choice of another
game — a new `pick-manual`, or accepting the next assignment in
`done`/`check`/`pick`/`scan` — calls `drop_inactive_picks`, after which
enforcement may remove them. `abandon-pick` works on them too. When no
replacement is accepted, the current game stays assigned.
- `mark_finished` is the single writer of `finished_app_ids`, so the sweep
and `done`/`check` cannot both record the same completion.
## Release rule: one new achievement (changed 2026-09-26)
Requiring 100% achievements locked the user onto one game for weeks, which
proved ineffective. Now the assignment and every manual pick release as soon as
**one achievement is unlocked after the game was assigned**
(`_assignment_progress.py`).
- "After" is judged by Steam's per-achievement `unlock_time` against a
baseline: an active pick's `started_at`, otherwise
`State.current_assigned_at`. `record_assignment` is the only writer of that
field and of `last_assigned_at`. An unknown baseline never releases.
- On upgrade, `State.load` backfills the baseline once and saves it: a pick
keeps its `started_at`; a plain assignment gets "now", so older unlocks do
not count.
- `finished_app_ids` still means **100% complete** and nothing else. A game
released below 100% gets `released_at` (on the pick and in
`State.released_at`) and a 7-day `skipped_until` cooldown, then returns to
the pool.
- Next pick order: least recently assigned first (never-assigned games lead),
then shortest HLTB time, so the same short game does not come straight back.
- Tampering detection compares a released game by unlock time after
`released_at`, because its snapshot row predates the legitimate unlocks.
- The enforce daemon now reloads every `State` field each tick. It saves state
from its pick sweep, and a stale copy of any field would overwrite what the
CLI wrote.
**Deployment note:** the enforce daemon holds the allowed set in code, so
`sudo systemctl restart steam-backlog-enforcer` is required after upgrading.
A pre-upgrade daemon only knows `current_app_id` and will uninstall the other

4
steam_backlog_enforcer/main/__init__.py
View File

@ -119,7 +119,7 @@ __all__ = [
COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"scan": ("Scan library & assign a game", do_scan),
"check": ("Check assigned game completion", do_check),
"check": ("Check assigned game for a new achievement", do_check),
"status": ("Show current status", cmd_status),
"list": ("List games from snapshot", cmd_list),
"install": ("Install the assigned game", cmd_install),
@ -131,7 +131,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = {
"installed": ("List installed games", cmd_installed),
"uninstall": ("Uninstall all non-assigned games", cmd_uninstall),
"setup": ("Run first-time setup", cmd_setup),
"done": ("Finish game, open HLTB, pick next", cmd_done),
"done": ("Move on after a new achievement, pick next", cmd_done),
"pick": ("Manually pick your next game from candidates", cmd_pick),
"stats": ("Show backlog completion-time estimates", cmd_stats),
"gaming-status": ("Show today's gaming time and block state", cmd_gaming_status),

6
steam_backlog_enforcer/main/_shared.py
View File

@ -158,9 +158,9 @@ def _show_manual_pick_lock_message(state: State) -> None:
any_in_grace |= _describe_pick(state, pick)
_echo(
"\nYou CANNOT use any other feature until you finish these games"
"\n(100% achievements) or their 2-week deadlines pass."
"\n\nTo release the lock: finish them, then run 'done' or 'check'."
"\nYou CANNOT use any other feature until you earn at least one new"
"\nachievement in each of these games or their 2-week deadlines pass."
"\n\nTo release the lock: earn one, then run 'done' or 'check'."
)
# 'abandon-pick' is dropped from the allowed list once no pick is still

3
steam_backlog_enforcer/main/misc.py
View File

@ -103,6 +103,9 @@ def cmd_reset(config: Config, state: State) -> None:
state.manual_pick_game_name = ""
state.manual_pick_started_at = ""
state.manual_picks = []
state.current_assigned_at = ""
state.last_assigned_at = {}
state.released_at = {}
state.save()
_echo("State reset. Store unblocked.")

22
steam_backlog_enforcer/main/picks.py
View File

@ -140,7 +140,9 @@ def _apply_allowed_set(config: Config, state: State) -> None:
def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
"""Manually pick a game by Steam app_id, locking the enforcer for 2 weeks.
"""Manually pick a game by Steam app_id, locking the enforcer.
The lock lasts 2 weeks, or until one new achievement is earned in it.
Args:
config: Enforcer configuration.
@ -175,19 +177,21 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
f"\n - Add it to your allowed games ({len(existing) + 1} of"
f" {config.max_manual_picks} slot(s) used)"
f"\n - Lock ALL other commands for {_MANUAL_LOCK_DAYS} DAYS or until"
f"\n you reach 100% achievements on every pick"
f"\n you earn at least one new achievement on every pick"
f"\n - Leave only these commands usable:"
f"\n {', '.join(sorted(_MANUAL_LOCK_EXEMPT_COMMANDS))}"
f"\n - Stay undoable at any time via 'abandon-pick {app_id}'"
)
for done_pick in retired:
active = active_manual_picks(state)
for old_pick in (p for p in state.manual_picks if p not in active):
_echo(
f" - Uninstall {done_pick.game_name} (completed, no longer an"
" allowed game)"
f" - Replace {old_pick['game_name']} (released or expired; it stays"
" playable until you confirm)"
)
_echo()
confirm = input(
f"Type YES to confirm you will play {game_name} until completion: "
f"Type YES to confirm you will play {game_name} until you earn a new"
" achievement: "
).strip()
if confirm != "YES":
_echo("Aborted.")
@ -202,8 +206,10 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
sys.exit(1)
_echo(f"\nManual pick confirmed: {game_name} (AppID={app_id})")
_echo(f"Lock active from now until 100% achievements or {_MANUAL_LOCK_DAYS} days.")
_echo("Run 'done' or 'check' once you have 100% to release the lock.\n")
_echo(
f"Lock active from now until one new achievement or {_MANUAL_LOCK_DAYS} days."
)
_echo("Run 'done' or 'check' once you earn one to release the lock.\n")
# Post-assignment: mirror what _assign_chosen_game + cmd_pick do, but for
# the whole allowed set so an earlier pick is not torn down by a later one.

27
steam_backlog_enforcer/scanning.py
View File

@ -6,12 +6,16 @@ import logging
import time
from typing import TYPE_CHECKING
from steam_backlog_enforcer._assignment_progress import (
release_game,
release_verdict,
)
from steam_backlog_enforcer._hltb_cached import fetch_hltb_times_cached
from steam_backlog_enforcer._hltb_types import (
load_hltb_count_comp_cache,
load_hltb_polls_cache,
)
from steam_backlog_enforcer._pick_completion import mark_finished, report_completion
from steam_backlog_enforcer._pick_completion import report_completion
from steam_backlog_enforcer._scanning_assign import (
_NO_CONF_MSG,
_assign_chosen_game,
@ -151,7 +155,8 @@ def pick_next_game(
) -> None:
"""Present a ranked list of eligible games and let the user pick one.
Games are ranked by shortest completionist time first. Games with
Games are ranked least-recently-assigned first, then by shortest
completionist time. Games with
silver-or-worse ProtonDB ratings (or gold trending downward) are
excluded as unplayable on Linux.
@ -186,7 +191,7 @@ def pick_next_game(
def do_check(config: Config, state: State) -> None:
"""Check assigned game completion status; detect tampering."""
"""Check the assigned game for a new achievement; detect tampering."""
report_completion(config, state)
if state.current_app_id is None:
_echo("No game currently assigned. Run 'scan' first.")
@ -205,12 +210,13 @@ def do_check(config: Config, state: State) -> None:
f" ({game.completion_pct:.1f}%)"
)
if game.is_complete:
_echo(f"\n COMPLETED: {state.current_game_name}!")
mark_finished(state, state.current_app_id)
released, verdict = release_verdict(state, game)
_echo(f"\n {verdict}")
if released:
release_game(state, game)
send_notification(
"Game Complete!",
f"You finished {state.current_game_name}! Picking next game...",
"Moving on!",
f"New achievement in {state.current_game_name}! Picking next game...",
)
# Load snapshot and pick next.
@ -219,13 +225,8 @@ def do_check(config: Config, state: State) -> None:
games = [GameInfo.from_snapshot(d) for d in snapshot_data]
pick_next_game(games, state, config)
else:
state.current_app_id = None
state.current_game_name = ""
state.save()
_echo(" Run 'scan' to pick the next game.")
else:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f" {remaining} achievements remaining. Keep going!")
# Tampering detection on snapshot.
detect_tampering(config, state)

20
steam_backlog_enforcer/tests/_main_helpers.py
View File

@ -12,6 +12,7 @@ from typing import Any
from steam_backlog_enforcer._allowed_games import MANUAL_LOCK_DAYS
from steam_backlog_enforcer._total_block import TotalBlockStatus
from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.steam_api import GameInfo
STARTED_AT = (datetime.now(UTC) - timedelta(days=1)).isoformat()
@ -141,3 +142,22 @@ def two_pick_state() -> State:
state.current_app_id = 200
state.current_game_name = "SecondGame"
return state
def done_game(app_id: int = 1, name: str = "G") -> GameInfo:
"""Build a 100%-complete GameInfo, as ``done`` sees a finished game.
Args:
app_id: Steam application id.
name: Game name.
Returns:
A GameInfo with every achievement unlocked.
"""
return GameInfo(
app_id=app_id,
name=name,
total_achievements=10,
unlocked_achievements=10,
playtime_minutes=60,
)

181
steam_backlog_enforcer/tests/test_assignment_progress.py Normal file
View File

@ -0,0 +1,181 @@
"""Tests for the "earn one new achievement" release rule."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from steam_backlog_enforcer._assignment_progress import (
RELEASE_COOLDOWN_DAYS,
assignment_baseline,
iso_to_epoch,
last_assigned_epoch,
mark_finished,
newest_unlock_since,
record_assignment,
release_game,
release_verdict,
)
from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.steam_api import AchievementInfo, GameInfo
ASSIGNED = datetime(2026, 9, 20, tzinfo=UTC)
BEFORE = int((ASSIGNED - timedelta(days=1)).timestamp())
AFTER = int((ASSIGNED + timedelta(days=1)).timestamp())
LATER = int((ASSIGNED + timedelta(days=2)).timestamp())
def game(*unlocks: tuple[str, int], total: int = 5, app_id: int = 1) -> GameInfo:
"""Build a GameInfo whose achievements unlocked at the given epochs.
Args:
unlocks: ``(display_name, unlock_time)`` per unlocked achievement.
total: Total achievement count (locked ones fill the rest).
app_id: Steam application id.
Returns:
The GameInfo.
"""
achievements = [
AchievementInfo(n, n, achieved=True, unlock_time=t) for n, t in unlocks
]
achievements += [
AchievementInfo(f"L{i}", f"L{i}", achieved=False, unlock_time=0)
for i in range(total - len(unlocks))
]
return GameInfo(
app_id=app_id,
name="G",
total_achievements=total,
unlocked_achievements=len(unlocks),
playtime_minutes=60,
achievements=achievements,
)
def assigned_state(app_id: int = 1) -> State:
"""State whose current game *app_id* was assigned at ``ASSIGNED``."""
return State(
current_app_id=app_id,
current_game_name="G",
current_assigned_at=ASSIGNED.isoformat(),
)
class TestIsoToEpoch:
def test_parses(self) -> None:
assert iso_to_epoch(ASSIGNED.isoformat()) == ASSIGNED.timestamp()
def test_empty_and_malformed_are_unknown(self) -> None:
assert iso_to_epoch("") is None
assert iso_to_epoch("not a date") is None
class TestNewestUnlockSince:
def test_picks_the_latest_unlock_after_the_baseline(self) -> None:
g = game(("old", BEFORE), ("a", AFTER), ("b", LATER))
newest = newest_unlock_since(g, ASSIGNED.isoformat())
assert newest is not None
assert newest.display_name == "b"
def test_unlocks_before_the_baseline_do_not_count(self) -> None:
assert newest_unlock_since(game(("old", BEFORE)), ASSIGNED.isoformat()) is None
def test_unknown_baseline_never_counts(self) -> None:
assert newest_unlock_since(game(("a", AFTER)), "") is None
class TestAssignmentBaseline:
def test_active_pick_uses_its_started_at(self) -> None:
state = assigned_state()
state.manual_picks = [{"app_id": 1, "game_name": "G", "started_at": "P"}]
assert assignment_baseline(state, 1) == "P"
def test_released_pick_falls_back_to_the_assignment(self) -> None:
state = assigned_state()
state.manual_picks = [
{"app_id": 1, "game_name": "G", "started_at": "P", "released_at": "R"}
]
assert assignment_baseline(state, 1) == ASSIGNED.isoformat()
def test_unassigned_game_has_no_baseline(self) -> None:
assert assignment_baseline(assigned_state(), 2) == ""
class TestReleaseVerdict:
def test_new_achievement_releases(self) -> None:
released, why = release_verdict(assigned_state(), game(("Win", AFTER)))
assert released is True
assert "NEW ACHIEVEMENT on G: 'Win'" in why
def test_complete_releases_even_without_a_new_unlock(self) -> None:
released, why = release_verdict(assigned_state(), game(("a", BEFORE), total=1))
assert released is True
assert "100%" in why
def test_no_new_achievement_keeps_the_game(self) -> None:
released, why = release_verdict(assigned_state(), game(("a", BEFORE)))
assert released is False
assert "No new achievement since 2026-09-20" in why
def test_no_baseline_is_named_generically(self) -> None:
released, why = release_verdict(State(), game())
assert released is False
assert "since its assignment" in why
class TestRecordAssignment:
def test_stamps_current_and_history(self) -> None:
state = State()
record_assignment(state, 7, "Seven", at="T")
assert (state.current_app_id, state.current_game_name) == (7, "Seven")
assert state.current_assigned_at == "T"
assert state.last_assigned_at == {"7": "T"}
assert state.enforcement_started_at == "T"
def test_keeps_the_first_enforcement_start(self) -> None:
state = State(enforcement_started_at="FIRST")
record_assignment(state, 7, "Seven")
assert state.enforcement_started_at == "FIRST"
assert iso_to_epoch(state.current_assigned_at) is not None
class TestLastAssignedEpoch:
def test_never_assigned_sorts_first(self) -> None:
assert last_assigned_epoch(State(), 1) == 0.0
def test_reads_the_stamp(self) -> None:
state = State(last_assigned_at={"1": ASSIGNED.isoformat()})
assert last_assigned_epoch(state, 1) == ASSIGNED.timestamp()
class TestMarkFinished:
def test_records_once(self) -> None:
state = State()
assert mark_finished(state, 7) is True
assert mark_finished(state, 7) is False
assert state.finished_app_ids == [7]
class TestReleaseGame:
def test_below_100_is_not_finished_but_cools_down(self) -> None:
state = assigned_state()
state.manual_picks = [{"app_id": 1, "game_name": "G", "started_at": "P"}]
release_game(state, game(("a", AFTER)))
assert state.finished_app_ids == []
assert state.manual_picks[0]["released_at"]
assert "1" in state.released_at
expiry = datetime.fromisoformat(state.skipped_until["1"])
assert expiry > datetime.now(UTC) + timedelta(days=RELEASE_COOLDOWN_DAYS - 1)
def test_complete_game_is_finished(self) -> None:
state = assigned_state()
release_game(state, game(("a", AFTER), total=1))
assert state.finished_app_ids == [1]
def test_keeps_an_earlier_pick_release_time(self) -> None:
state = assigned_state()
state.manual_picks = [
{"app_id": 1, "game_name": "G", "started_at": "P", "released_at": "R"}
]
release_game(state, game(("a", AFTER)))
assert state.manual_picks[0]["released_at"] == "R"

16
steam_backlog_enforcer/tests/test_main_part2.py
View File

@ -7,6 +7,7 @@ from unittest.mock import (
from steam_backlog_enforcer._cmd_done_finalize import _finalize_completion
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.tests._main_helpers import done_game
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
PKG = "steam_backlog_enforcer.main"
@ -60,7 +61,7 @@ class TestFinalizeCompletion:
s.current_game_name = "NewGame"
mock_pick.side_effect = set_next
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))
assert 1 in state.finished_app_ids
def test_hide_skipped_when_steam_unreachable(self) -> None:
@ -98,7 +99,7 @@ class TestFinalizeCompletion:
s.current_game_name = "NewGame"
mock_pick.side_effect = set_next
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))
assert "skipped (update in progress)" in " ".join(
str(c) for c in mock_echo.call_args_list
)
@ -111,8 +112,9 @@ class TestFinalizeCompletion:
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=None),
patch.object(State, "save"),
):
_finalize_completion(config, state, "G", 1)
assert state.current_app_id is None
_finalize_completion(config, state, done_game(1, "G"))
# No replacement was chosen, so the released game stays assigned.
assert state.current_app_id == 1
def test_no_next_game(self) -> None:
config = Config()
@ -134,7 +136,7 @@ class TestFinalizeCompletion:
s.current_app_id = None
mock_pick.side_effect = set_none
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))
def test_no_owned_ids(self) -> None:
config = Config()
@ -162,7 +164,7 @@ class TestFinalizeCompletion:
s.current_game_name = "Next"
mock_pick.side_effect = set_2
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))
def test_hide_returns_zero(self) -> None:
config = Config()
@ -194,4 +196,4 @@ class TestFinalizeCompletion:
s.current_game_name = "Next"
mock_pick.side_effect = set_2
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))

7
steam_backlog_enforcer/tests/test_main_part2c.py
View File

@ -7,6 +7,7 @@ from unittest.mock import (
from steam_backlog_enforcer._cmd_done_finalize import _finalize_completion
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.tests._main_helpers import done_game
if TYPE_CHECKING:
from steam_backlog_enforcer.steam_api import GameInfo
@ -68,7 +69,7 @@ class TestFinalizeCompletionRetries:
patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=capture_pick),
patch.object(State, "save"),
):
_finalize_completion(config, state, "G", 1)
_finalize_completion(config, state, done_game(1, "G"))
assert seen[2] == 20.05
assert seen[3] == 18.81
@ -106,7 +107,7 @@ class TestFinalizeCompletionRetries:
patch(f"{CMD_DONE_PKG}.send_notification"),
patch.object(State, "save"),
):
_finalize_completion(config, state, "DoneGame", 1)
_finalize_completion(config, state, done_game(1, "DoneGame"))
mock_install.assert_called_once_with(2, "Next", "sid", use_steam_protocol=True)
@ -142,6 +143,6 @@ class TestFinalizeCompletionRetries:
patch(f"{CMD_DONE_PKG}.send_notification"),
patch.object(State, "save"),
):
_finalize_completion(config, state, "DoneGame", 1)
_finalize_completion(config, state, done_game(1, "DoneGame"))
mock_install.assert_not_called()

11
steam_backlog_enforcer/tests/test_main_pick_cap.py
View File

@ -10,6 +10,7 @@ from steam_backlog_enforcer.main import (
cmd_pick_manual,
)
from steam_backlog_enforcer.tests._main_helpers import (
STARTED_AT,
locked_state,
two_pick_state,
)
@ -126,6 +127,14 @@ class TestPickManualRetirementNotices:
state = locked_state(app_id=100)
state.current_app_id = 200
state.current_game_name = "Finished"
state.manual_picks.append(
{
"app_id": 200,
"game_name": "Finished",
"started_at": STARTED_AT,
"released_at": STARTED_AT,
}
)
retired = [PickProgress(200, "Finished", 5, 5, retired=True, determinable=True)]
with (
patch(f"{PKG}.report_completion", return_value=retired),
@ -137,7 +146,7 @@ class TestPickManualRetirementNotices:
):
cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"])
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "Uninstall Finished (completed" in output
assert "Replace Finished (released or expired" in output
assert "Aborted." in output
# Abort leaves the finished game as the assignment, so say so.
note = " ".join(str(c) for c in mock_note.call_args_list)

62
steam_backlog_enforcer/tests/test_pick_completion.py
View File

@ -1,42 +1,50 @@
"""Tests for achievement-based retirement of finished manual picks."""
"""Tests for achievement-based release of manual picks."""
from datetime import UTC, datetime
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._pick_completion import (
MANUAL_PICK_RECHECK_TTL_SECONDS,
PickProgress,
_DaemonSweep,
daemon_sweep,
mark_finished,
report_completion,
retire_completed_manual_picks,
retire_completed_manual_picks_throttled,
warn_stale_assignment,
)
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.steam_api import GameInfo
from steam_backlog_enforcer.steam_api import AchievementInfo, GameInfo
from steam_backlog_enforcer.tests._main_helpers import locked_state, two_pick_state
PKG = "steam_backlog_enforcer._pick_completion"
def game(app_id: int, unlocked: int, total: int) -> GameInfo:
def game(app_id: int, unlocked: int, total: int, *, new: bool = False) -> GameInfo:
"""Build a GameInfo with the given achievement progress.
Args:
app_id: Steam application id.
unlocked: Unlocked achievement count.
total: Total achievement count.
new: Whether one of the unlocks happened just now (after any pick).
Returns:
A GameInfo carrying that progress.
"""
now = int(datetime.now(UTC).timestamp())
achievements = [
AchievementInfo(f"a{i}", f"Ach{i}", i < unlocked, now if new and i == 0 else 1)
for i in range(total)
]
return GameInfo(
app_id=app_id,
name="G",
total_achievements=total,
unlocked_achievements=unlocked,
playtime_minutes=0,
achievements=achievements,
)
@ -54,30 +62,24 @@ def client_returning(*games: GameInfo | None) -> MagicMock:
return fake
class TestMarkFinished:
def test_records_once(self) -> None:
state = State()
assert mark_finished(state, 7) is True
assert state.finished_app_ids == [7]
def test_does_not_duplicate(self) -> None:
state = State(finished_app_ids=[7])
assert mark_finished(state, 7) is False
assert state.finished_app_ids == [7]
class TestPickProgressDescribe:
def test_undeterminable(self) -> None:
line = PickProgress(1, "G", 0, 0, retired=False, determinable=False).describe()
assert "progress unavailable" in line
def test_retired(self) -> None:
def test_retired_at_100(self) -> None:
line = PickProgress(1, "G", 5, 5, retired=True, determinable=True).describe()
assert "COMPLETE, freeing its slot" in line
assert "COMPLETE, slot freed, stays playable" in line
def test_released_by_new_achievement(self) -> None:
line = PickProgress(
1, "G", 2, 5, retired=True, determinable=True, new_achievement="Win"
).describe()
assert "2/5 (40%) - earned 'Win', slot freed, stays playable" in line
def test_in_progress(self) -> None:
line = PickProgress(1, "G", 1, 4, retired=False, determinable=True).describe()
assert "1/4 (25%) - still in progress" in line
assert "1/4 (25%) - no new achievement since it was picked yet" in line
def test_in_progress_with_zero_total(self) -> None:
# Defensive: determinable but no achievements must not divide by zero.
@ -101,6 +103,18 @@ class TestRetireCompletedManualPicks:
assert state.finished_app_ids == [100]
mock_save.assert_called_once()
def test_new_achievement_releases_without_finishing(self) -> None:
state = locked_state(app_id=100)
with patch.object(State, "save") as mock_save:
results = retire_completed_manual_picks(
Config(), state, client=client_returning(game(100, 1, 5, new=True))
)
assert [(r.retired, r.new_achievement) for r in results] == [(True, "Ach0")]
assert state.finished_app_ids == []
assert state.manual_picks[0]["released_at"]
assert 100 in state.active_skipped_ids()
mock_save.assert_called_once()
def test_incomplete_pick_is_left_alone(self) -> None:
state = locked_state(app_id=100)
with patch.object(State, "save") as mock_save:
@ -176,8 +190,8 @@ class TestReportCompletion:
retired = report_completion(Config(), state)
assert [r.app_id for r in retired] == [200]
output = " ".join(str(c) for c in mock_echo.call_args_list)
assert "still in progress" in output
assert "COMPLETE, freeing its slot" in output
assert "no new achievement" in output
assert "COMPLETE, slot freed" in output
class TestWarnStaleAssignment:
@ -205,10 +219,9 @@ class TestDaemonSweep:
assert sweep.due(1100.0, 900.0) is False
assert sweep.due(2000.0, 900.0) is True
def test_throttled_call_records_retirements_without_evicting(self) -> None:
def test_throttled_call_releases_without_evicting(self) -> None:
state = locked_state(app_id=100)
daemon_sweep.last = None
daemon_sweep.retired.clear()
try:
with (
patch.object(State, "save"),
@ -219,12 +232,11 @@ class TestDaemonSweep:
):
results = retire_completed_manual_picks_throttled(Config(), state)
assert [r.retired for r in results] == [True]
assert daemon_sweep.retired == {100}
assert 100 in allowed_app_ids(state)
# Second call inside the TTL must not re-query Steam.
assert retire_completed_manual_picks_throttled(Config(), state) == []
finally:
daemon_sweep.last = None
daemon_sweep.retired.clear()
def test_ttl_is_well_clear_of_the_three_second_loop(self) -> None:
assert MANUAL_PICK_RECHECK_TTL_SECONDS >= 300

211
steam_backlog_enforcer/tests/test_release_lifecycle.py Normal file
View File

@ -0,0 +1,211 @@
"""A released game stays playable until the user explicitly picks another.
Covers the state upgrade, the allowed set, abandoning, pick order, tampering
and the ``check`` command under the "one new achievement" rule.
"""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import json
from unittest.mock import MagicMock, patch
from steam_backlog_enforcer import config as config_mod
from steam_backlog_enforcer._actions import (
allowed_app_ids,
apply_manual_pick,
is_manual_pick_locked,
)
from steam_backlog_enforcer._allowed_games import drop_inactive_picks
from steam_backlog_enforcer._manual_pick_lifecycle import abandon_manual_pick
from steam_backlog_enforcer._scanning_assign import _open_candidates
from steam_backlog_enforcer._scanning_tampering import _check_game_tampering
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.scanning import do_check
from steam_backlog_enforcer.steam_api import AchievementInfo, GameInfo
from steam_backlog_enforcer.tests._main_helpers import EXPIRED_AT, STARTED_AT
RELEASED = datetime(2026, 9, 20, tzinfo=UTC)
def pick(app_id: int, **extra: str) -> dict[str, object]:
"""Build one manual-pick entry started a day ago."""
return {
"app_id": app_id,
"game_name": f"G{app_id}",
"started_at": STARTED_AT,
} | dict(extra.items())
def game(app_id: int, *unlock_times: int, hours: float = 5.0) -> GameInfo:
"""Build a GameInfo with one unlocked achievement per epoch given."""
return GameInfo(
app_id=app_id,
name=f"G{app_id}",
total_achievements=10,
unlocked_achievements=len(unlock_times),
playtime_minutes=60,
completionist_hours=hours,
achievements=[
AchievementInfo(f"a{t}", f"A{t}", achieved=True, unlock_time=t)
for t in unlock_times
],
)
def write_state(data: dict[str, object]) -> None:
"""Write raw JSON to the (conftest-redirected) state file."""
config_mod.STATE_FILE.write_text(json.dumps(data), encoding="utf-8")
class TestLoadBackfill:
def test_plain_assignment_is_stamped_once_and_persisted(self) -> None:
write_state({"current_app_id": 5, "current_game_name": "G5"})
first = State.load()
assert first.current_assigned_at
assert State.load().current_assigned_at == first.current_assigned_at
assert first.last_assigned_at == {"5": first.current_assigned_at}
def test_current_pick_keeps_its_pick_time(self) -> None:
write_state({"current_app_id": 5, "manual_picks": [pick(5), pick(6)]})
state = State.load()
assert state.current_assigned_at == STARTED_AT
assert state.last_assigned_at == {"5": STARTED_AT, "6": STARTED_AT}
def test_legacy_pick_migration_is_persisted(self) -> None:
write_state({"manual_pick_app_id": 9, "manual_pick_started_at": ""})
State.load()
saved = json.loads(config_mod.STATE_FILE.read_text(encoding="utf-8"))
assert saved["manual_picks"][0]["app_id"] == 9
assert saved["manual_pick_app_id"] is None
def test_up_to_date_state_is_not_rewritten(self) -> None:
write_state({"current_app_id": None})
with patch.object(State, "save") as mock_save:
State.load()
mock_save.assert_not_called()
class TestReleasedPicksStayPlayable:
def test_released_and_expired_picks_stay_allowed_but_unlocked(self) -> None:
state = State(
current_app_id=1,
manual_picks=[
pick(2, released_at=STARTED_AT),
pick(3) | {"started_at": EXPIRED_AT},
],
)
assert allowed_app_ids(state) == {1, 2, 3}
assert is_manual_pick_locked(state) is False
def test_explicit_choice_drops_only_inactive_picks(self) -> None:
state = State(manual_picks=[pick(2, released_at=STARTED_AT), pick(3)])
dropped = drop_inactive_picks(state)
assert [p["app_id"] for p in dropped] == [2]
assert [p["app_id"] for p in state.manual_picks] == [3]
def test_new_manual_pick_replaces_released_ones(self) -> None:
state = State(current_app_id=2, manual_picks=[pick(2, released_at=STARTED_AT)])
with patch.object(State, "save"):
assert apply_manual_pick(state, 4, "G4", max_picks=1) is None
assert allowed_app_ids(state) == {4}
assert state.current_assigned_at == state.manual_picks[0]["started_at"]
def test_released_pick_can_be_abandoned(self) -> None:
state = State(
current_app_id=2, manual_picks=[pick(2, released_at="R"), pick(3)]
)
with patch.object(State, "save"):
assert abandon_manual_pick(state, 2) is True
# The active pick inherits the assignment, timed from its pick.
assert (state.current_app_id, state.current_assigned_at) == (3, STARTED_AT)
def test_abandoning_the_last_pick_clears_the_assignment(self) -> None:
state = State(current_app_id=2, current_assigned_at="T", manual_picks=[pick(2)])
with patch.object(State, "save"):
abandon_manual_pick(state, 2)
assert (state.current_app_id, state.current_assigned_at) == (None, "")
class TestPickOrder:
def test_least_recently_assigned_first_then_shortest(self) -> None:
state = State(
last_assigned_at={
"1": "2026-09-01T00:00:00+00:00",
"2": "2026-08-01T00:00:00+00:00",
}
)
games = [game(1, hours=1.0), game(2, hours=2.0), game(3, hours=9.0)]
with patch(
"steam_backlog_enforcer._scanning_assign"
"._apply_cached_confidence_to_candidates"
):
order = [g.app_id for g in _open_candidates(games, state)]
assert order == [3, 2, 1]
class TestTamperingAfterRelease:
def entry(self) -> dict[str, object]:
return {
"app_id": 7,
"name": "G7",
"total_achievements": 10,
"unlocked_achievements": 0,
"playtime_minutes": 60,
}
def check(self, *unlock_times: int) -> tuple[str, int, int] | None:
client = MagicMock()
client.refresh_single_game.return_value = game(7, *unlock_times)
state = State(current_app_id=1, released_at={"7": RELEASED.isoformat()})
return _check_game_tampering(client, self.entry(), state)
def test_unlocks_earned_while_assigned_are_legitimate(self) -> None:
before = int((RELEASED - timedelta(hours=1)).timestamp())
assert self.check(before, before) is None
def test_unlocks_after_release_are_flagged(self) -> None:
before = int((RELEASED - timedelta(hours=1)).timestamp())
after = int((RELEASED + timedelta(hours=1)).timestamp())
assert self.check(before, after) == ("G7", 7, 1)
def test_unreadable_game_is_not_flagged(self) -> None:
client = MagicMock()
client.refresh_single_game.return_value = None
assert _check_game_tampering(client, self.entry(), State()) is None
class TestCheckCommand:
def run_check(self, state: State, refreshed: GameInfo) -> MagicMock:
client = MagicMock()
client.refresh_single_game.return_value = refreshed
with (
patch(
"steam_backlog_enforcer.scanning.SteamAPIClient", return_value=client
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer.scanning.report_completion"),
patch("steam_backlog_enforcer.scanning.detect_tampering"),
patch("steam_backlog_enforcer.scanning.send_notification"),
patch("steam_backlog_enforcer.scanning.load_snapshot", return_value=[]),
patch("steam_backlog_enforcer.scanning.pick_next_game") as mock_pick,
patch.object(State, "save"),
):
do_check(Config(), state)
return mock_pick
def test_new_achievement_releases_and_keeps_the_game(self) -> None:
state = State(current_app_id=7, current_assigned_at=RELEASED.isoformat())
after = int((RELEASED + timedelta(hours=1)).timestamp())
self.run_check(state, game(7, after))
assert "7" in state.released_at
assert state.finished_app_ids == []
# No snapshot, so no replacement: the game stays assigned.
assert state.current_app_id == 7
def test_no_new_achievement_changes_nothing(self) -> None:
state = State(current_app_id=7, current_assigned_at=RELEASED.isoformat())
before = int((RELEASED - timedelta(hours=1)).timestamp())
mock_pick = self.run_check(state, game(7, before))
assert state.released_at == {}
mock_pick.assert_not_called()

4
steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py
View File

@ -96,14 +96,14 @@ class TestPickNextGameSequential:
assert any("Skipped G1 for 7 days" in line for line in echoed)
def test_on_select_no_candidates(self) -> None:
"""Sequential branch with no candidates clears state."""
"""Sequential branch with no candidates keeps the current game."""
complete = _game(app_id=1, hours=1.0, total=10, unlocked=10)
config = Config(steam_api_key="k", steam_id="i")
state = State(current_app_id=99, current_game_name="X")
echoed: list[str] = []
with self._common_patches(echoed):
pick_next_game([complete], state, config, on_select=lambda _g: True)
assert state.current_app_id is None
assert state.current_app_id == 99
def test_on_select_no_playable_branch(self) -> None:
"""Sequential branch when all candidates lack Linux compatibility."""