From a191c8e459df1eff3d2530ec5b23095eab9d12c7 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Sat, 26 Sep 2026 16:46:03 +0200 Subject: [PATCH] Release a game after one new achievement instead of 100% completion 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 Claude-Session: https://claude.ai/code/session_01PhzaTAWtRfJk8gre6NLSka --- README.md | 12 +- steam_backlog_enforcer/_actions.py | 22 +- steam_backlog_enforcer/_allowed_games.py | 43 +++- .../_assignment_progress.py | 162 ++++++++++++++ steam_backlog_enforcer/_cmd_done_finalize.py | 48 ++-- steam_backlog_enforcer/_enforce_loop.py | 26 +-- .../_manual_pick_lifecycle.py | 27 ++- steam_backlog_enforcer/_mcp_actions.py | 13 +- steam_backlog_enforcer/_pick_completion.py | 104 ++++----- steam_backlog_enforcer/_scanning_assign.py | 38 +++- steam_backlog_enforcer/_scanning_tampering.py | 13 +- steam_backlog_enforcer/_state_migrations.py | 84 +++++++ steam_backlog_enforcer/config.py | 43 ++-- .../docs/DOCS-manual-pick-design.md | 66 ++++-- steam_backlog_enforcer/main/__init__.py | 4 +- steam_backlog_enforcer/main/_shared.py | 6 +- steam_backlog_enforcer/main/misc.py | 3 + steam_backlog_enforcer/main/picks.py | 22 +- steam_backlog_enforcer/scanning.py | 27 +-- steam_backlog_enforcer/tests/_main_helpers.py | 20 ++ .../tests/test_assignment_progress.py | 181 +++++++++++++++ .../tests/test_main_part2.py | 16 +- .../tests/test_main_part2c.py | 7 +- .../tests/test_main_pick_cap.py | 11 +- .../tests/test_pick_completion.py | 62 ++--- .../tests/test_release_lifecycle.py | 211 ++++++++++++++++++ ...test_scanning_pick_next_game_sequential.py | 4 +- 27 files changed, 1025 insertions(+), 250 deletions(-) create mode 100644 steam_backlog_enforcer/_assignment_progress.py create mode 100644 steam_backlog_enforcer/_state_migrations.py create mode 100644 steam_backlog_enforcer/tests/test_assignment_progress.py create mode 100644 steam_backlog_enforcer/tests/test_release_lifecycle.py diff --git a/README.md b/README.md index fd50a85..49f029e 100644 --- a/README.md +++ b/README.md @@ -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 | diff --git a/steam_backlog_enforcer/_actions.py b/steam_backlog_enforcer/_actions.py index fd08956..e7eedfd 100644 --- a/steam_backlog_enforcer/_actions.py +++ b/steam_backlog_enforcer/_actions.py @@ -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 diff --git a/steam_backlog_enforcer/_allowed_games.py b/steam_backlog_enforcer/_allowed_games.py index 3368602..f2658d5 100644 --- a/steam_backlog_enforcer/_allowed_games.py +++ b/steam_backlog_enforcer/_allowed_games.py @@ -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", ""))) diff --git a/steam_backlog_enforcer/_assignment_progress.py b/steam_backlog_enforcer/_assignment_progress.py new file mode 100644 index 0000000..c315135 --- /dev/null +++ b/steam_backlog_enforcer/_assignment_progress.py @@ -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) diff --git a/steam_backlog_enforcer/_cmd_done_finalize.py b/steam_backlog_enforcer/_cmd_done_finalize.py index 0afbd63..215efb3 100644 --- a/steam_backlog_enforcer/_cmd_done_finalize.py +++ b/steam_backlog_enforcer/_cmd_done_finalize.py @@ -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) diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py index aeb7f58..9670ee2 100644 --- a/steam_backlog_enforcer/_enforce_loop.py +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -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) diff --git a/steam_backlog_enforcer/_manual_pick_lifecycle.py b/steam_backlog_enforcer/_manual_pick_lifecycle.py index 8bc399a..d8ca824 100644 --- a/steam_backlog_enforcer/_manual_pick_lifecycle.py +++ b/steam_backlog_enforcer/_manual_pick_lifecycle.py @@ -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 diff --git a/steam_backlog_enforcer/_mcp_actions.py b/steam_backlog_enforcer/_mcp_actions.py index d11a280..de2e9f9 100644 --- a/steam_backlog_enforcer/_mcp_actions.py +++ b/steam_backlog_enforcer/_mcp_actions.py @@ -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: diff --git a/steam_backlog_enforcer/_pick_completion.py b/steam_backlog_enforcer/_pick_completion.py index b318842..666078e 100644 --- a/steam_backlog_enforcer/_pick_completion.py +++ b/steam_backlog_enforcer/_pick_completion.py @@ -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) diff --git a/steam_backlog_enforcer/_scanning_assign.py b/steam_backlog_enforcer/_scanning_assign.py index 1801d5e..f247447 100644 --- a/steam_backlog_enforcer/_scanning_assign.py +++ b/steam_backlog_enforcer/_scanning_assign.py @@ -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 diff --git a/steam_backlog_enforcer/_scanning_tampering.py b/steam_backlog_enforcer/_scanning_tampering.py index 5f4e804..59df8c4 100644 --- a/steam_backlog_enforcer/_scanning_tampering.py +++ b/steam_backlog_enforcer/_scanning_tampering.py @@ -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 diff --git a/steam_backlog_enforcer/_state_migrations.py b/steam_backlog_enforcer/_state_migrations.py new file mode 100644 index 0000000..eb8d38f --- /dev/null +++ b/steam_backlog_enforcer/_state_migrations.py @@ -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 diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index 1214321..31a06f4 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -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 = "" diff --git a/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md b/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md index a474841..2784830 100644 --- a/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md +++ b/steam_backlog_enforcer/docs/DOCS-manual-pick-design.md @@ -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 ` 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 diff --git a/steam_backlog_enforcer/main/__init__.py b/steam_backlog_enforcer/main/__init__.py index 9087e42..694860e 100644 --- a/steam_backlog_enforcer/main/__init__.py +++ b/steam_backlog_enforcer/main/__init__.py @@ -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), diff --git a/steam_backlog_enforcer/main/_shared.py b/steam_backlog_enforcer/main/_shared.py index 9fa196e..9a107dc 100644 --- a/steam_backlog_enforcer/main/_shared.py +++ b/steam_backlog_enforcer/main/_shared.py @@ -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 diff --git a/steam_backlog_enforcer/main/misc.py b/steam_backlog_enforcer/main/misc.py index e00f3e5..cd1a0ab 100644 --- a/steam_backlog_enforcer/main/misc.py +++ b/steam_backlog_enforcer/main/misc.py @@ -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.") diff --git a/steam_backlog_enforcer/main/picks.py b/steam_backlog_enforcer/main/picks.py index d77570d..10387a2 100644 --- a/steam_backlog_enforcer/main/picks.py +++ b/steam_backlog_enforcer/main/picks.py @@ -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. diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index bdb1fb0..ab71579 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -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) diff --git a/steam_backlog_enforcer/tests/_main_helpers.py b/steam_backlog_enforcer/tests/_main_helpers.py index 5bdcf8e..4206c69 100644 --- a/steam_backlog_enforcer/tests/_main_helpers.py +++ b/steam_backlog_enforcer/tests/_main_helpers.py @@ -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, + ) diff --git a/steam_backlog_enforcer/tests/test_assignment_progress.py b/steam_backlog_enforcer/tests/test_assignment_progress.py new file mode 100644 index 0000000..1477370 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_assignment_progress.py @@ -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" diff --git a/steam_backlog_enforcer/tests/test_main_part2.py b/steam_backlog_enforcer/tests/test_main_part2.py index 3867728..0e16e42 100644 --- a/steam_backlog_enforcer/tests/test_main_part2.py +++ b/steam_backlog_enforcer/tests/test_main_part2.py @@ -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")) diff --git a/steam_backlog_enforcer/tests/test_main_part2c.py b/steam_backlog_enforcer/tests/test_main_part2c.py index da5a0ed..203f116 100644 --- a/steam_backlog_enforcer/tests/test_main_part2c.py +++ b/steam_backlog_enforcer/tests/test_main_part2c.py @@ -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() diff --git a/steam_backlog_enforcer/tests/test_main_pick_cap.py b/steam_backlog_enforcer/tests/test_main_pick_cap.py index c064568..3c98076 100644 --- a/steam_backlog_enforcer/tests/test_main_pick_cap.py +++ b/steam_backlog_enforcer/tests/test_main_pick_cap.py @@ -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) diff --git a/steam_backlog_enforcer/tests/test_pick_completion.py b/steam_backlog_enforcer/tests/test_pick_completion.py index ccdc18f..143f4ec 100644 --- a/steam_backlog_enforcer/tests/test_pick_completion.py +++ b/steam_backlog_enforcer/tests/test_pick_completion.py @@ -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 diff --git a/steam_backlog_enforcer/tests/test_release_lifecycle.py b/steam_backlog_enforcer/tests/test_release_lifecycle.py new file mode 100644 index 0000000..85d0ff2 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_release_lifecycle.py @@ -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() diff --git a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py index 773dd6d..0250a3a 100644 --- a/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py +++ b/steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py @@ -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."""