Auto-detect completed manual picks before enforcing the pick cap

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

pyproject.toml

@@ -51,6 +51,12 @@ ignore = [
     "COM812", # formatter handles this
     "ISC001", # formatter may create these
     "S603",   # prone to false positives
+    # Conflicts with SIM117. A `pytest.raises` that needs patches in scope has
+    # no legal form: combine them into one `with` and PT012 objects to the
+    # extra context managers; nest `raises` inside and SIM117 objects to the
+    # nested `with`. SIM117's combined form is the style used across this repo,
+    # so PT012 is the one that loses.
+    "PT012",
 ]
 fixable = ["ALL"]
 unfixable = []

steam_backlog_enforcer/_cmd_done_finalize.py

@@ -16,6 +16,7 @@ from steam_backlog_enforcer._cmd_done import (
     _report_assigned_confidence,
 )
 from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
+from steam_backlog_enforcer._pick_completion import mark_finished
 from steam_backlog_enforcer.config import Config, State, load_snapshot
 from steam_backlog_enforcer.enforcer import (
     enforce_allowed_game,
@@ -48,7 +49,7 @@ def _finalize_completion(
 ) -> None:
     """Mark game complete, pick next, hide non-assigned games, notify."""
     _echo(f"\n  COMPLETED: {game_name}!")
-    state.finished_app_ids.append(app_id)
+    mark_finished(state, app_id)
 
     snapshot_data = load_snapshot()
     _echo("\nPicking next game...")

steam_backlog_enforcer/_enforce_loop.py

@@ -16,6 +16,10 @@ from steam_backlog_enforcer._enforce_steps import (
 # Re-exported: five modules and the test suite import get_all_owned_app_ids
 # 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
 from steam_backlog_enforcer._playtime_session import PlaytimeSession, new_session
 from steam_backlog_enforcer._steam_launch import steam_is_installed
@@ -139,7 +143,16 @@ def _enforce_loop_iteration(
     if not steam_is_installed():
         return
 
-    allowed = allowed_app_ids(state)
+    # Retire manual picks that have hit 100%, 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
     if not allowed:
         return
 
@@ -224,6 +237,10 @@ def do_enforce(config: Config, state: State, *, demo: bool = False) -> None:
             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
 
             _enforce_loop_iteration(config, state, session=session, demo=demo)
             time.sleep(ENFORCE_INTERVAL)

steam_backlog_enforcer/_mcp_actions.py

@@ -23,6 +23,9 @@ from steam_backlog_enforcer._mcp_server import (
     logger,
     mcp,
 )
+from steam_backlog_enforcer._pick_completion import (
+    retire_completed_manual_picks,
+)
 from steam_backlog_enforcer._total_block import start_total_block
 from steam_backlog_enforcer.config import Config, State
 
@@ -61,12 +64,16 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]:
             ),
             "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.
+    retired = retire_completed_manual_picks(config, state)
     refused = apply_manual_pick(
         state,
         app_id,
         game_name,
-        max_picks=Config.load().max_manual_picks,
+        max_picks=config.max_manual_picks,
     )
     if refused is not None:
         return {"ok": False, "reason": refused}
@@ -77,6 +84,9 @@ def pick_manual(app_id: int, *, confirm: bool = False) -> dict[str, Any]:
         "action": "pick_manual",
         "app_id": app_id,
         "game_name": game_name,
+        "retired_completed": [
+            {"app_id": r.app_id, "game_name": r.game_name} for r in retired if r.retired
+        ],
     }
 
 

steam_backlog_enforcer/_pick_completion.py

@@ -0,0 +1,249 @@
+"""Achievement-based retirement of finished 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.
+
+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``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+import time
+from typing import TYPE_CHECKING
+
+from steam_backlog_enforcer._allowed_games import active_manual_picks
+from steam_backlog_enforcer.game_install import _echo
+from steam_backlog_enforcer.steam_api import SteamAPIClient
+
+if TYPE_CHECKING:
+    from steam_backlog_enforcer.config import Config, State
+
+# How long the enforce daemon waits between achievement re-checks. That loop
+# ticks every 3s; re-checking every tick would be ~1200 Steam API calls an hour
+# for a fact that changes a few times a year. CLI commands do not throttle.
+MANUAL_PICK_RECHECK_TTL_SECONDS = 900
+
+logger = logging.getLogger(__name__)
+
+
+class _DaemonSweep:
+    """The enforce daemon's re-check gate, plus what it has retired.
+
+    An object rather than module-level values because rebinding those 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.
+    """
+
+    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.
+
+        Args:
+            now: Current ``time.monotonic()`` reading.
+            ttl: Minimum seconds between checks.
+
+        Returns:
+            Whether the caller should run the check.
+        """
+        if self.last is not None and now - self.last < ttl:
+            return False
+        self.last = now
+        return True
+
+
+# Process-local by design: the daemon is long-lived, and a restart re-checking
+# once immediately is harmless.
+daemon_sweep = _DaemonSweep()
+
+
+@dataclass(frozen=True)
+class PickProgress:
+    """Achievement progress for one active manual pick.
+
+    ``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".
+    """
+
+    app_id: int
+    game_name: str
+    unlocked: int
+    total: int
+    retired: bool
+    determinable: bool
+
+    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
+
+
+def retire_completed_manual_picks(
+    config: Config,
+    state: State,
+    *,
+    client: SteamAPIClient | None = None,
+) -> list[PickProgress]:
+    """Mark every 100%-complete manual pick finished, 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.
+
+    Args:
+        config: Enforcer configuration (for the Steam credentials).
+        state: The enforcer state to inspect, mutate and save.
+        client: Pre-built API client, for reuse and for tests.
+
+    Returns:
+        One entry per *active* pick, in pick order -- including the ones left
+        alone, so callers can report progress rather than only retirements.
+    """
+    picks = active_manual_picks(state)
+    if not picks:
+        return []
+
+    if client is None:
+        client = SteamAPIClient(config.steam_api_key, config.steam_id)
+
+    results: list[PickProgress] = []
+    retired_any = False
+    for pick in picks:
+        # active_manual_picks has already dropped entries without an app_id.
+        app_id = pick["app_id"]
+        name = pick.get("game_name", "") or f"AppID={app_id}"
+
+        game = client.refresh_single_game(app_id, name)
+        if game is None:
+            results.append(
+                PickProgress(app_id, name, 0, 0, retired=False, determinable=False)
+            )
+            continue
+
+        if game.is_complete and mark_finished(state, app_id):
+            retired_any = True
+            logger.info("Manual pick retired at 100%%: %s (AppID=%s)", name, app_id)
+        results.append(
+            PickProgress(
+                app_id,
+                name,
+                game.unlocked_achievements,
+                game.total_achievements,
+                retired=game.is_complete,
+                determinable=True,
+            )
+        )
+
+    if retired_any:
+        state.save()
+    return results
+
+
+def report_completion(config: Config, state: State) -> list[PickProgress]:
+    """Run the sweep and print a per-pick progress block.
+
+    Args:
+        config: Enforcer configuration.
+        state: The enforcer state to inspect, mutate and save.
+
+    Returns:
+        Only the picks whose slots this call freed.
+    """
+    results = retire_completed_manual_picks(config, state)
+    if not results:
+        return []
+    _echo("\nChecking your manual picks for completion...")
+    for result in results:
+        _echo(result.describe())
+    return [r for r in results if r.retired]
+
+
+def warn_stale_assignment(state: State, retired: list[PickProgress]) -> None:
+    """Warn when a just-retired pick is still the current assignment.
+
+    Only reachable from the ``pick-manual`` abort path: on the success path
+    ``apply_manual_pick`` overwrites ``current_app_id`` with the new pick.
+
+    Args:
+        state: The enforcer state (for ``current_app_id``).
+        retired: Picks whose slots were freed by this run.
+    """
+    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"
+                "\n      your next game."
+            )
+            return
+
+
+def retire_completed_manual_picks_throttled(
+    config: Config,
+    state: State,
+) -> 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.
+
+    Returns:
+        The per-pick progress list, or ``[]`` when 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

steam_backlog_enforcer/_scanning_assign.py

@@ -10,6 +10,7 @@ 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._scanning_candidates import (
     _pick_next_shortest_candidate,
     _sort_key,
@@ -84,7 +85,12 @@ def _assign_chosen_game(
     )
     _report_poll_confidence(chosen, games, state)
     if config.uninstall_other_games:
-        count = uninstall_other_games(chosen.app_id)
+        # The whole allowed set, not just the new assignment: passing a bare
+        # int made the `in` test raise TypeError, and would tear down a
+        # concurrent manual pick even if it did not. Matches the other four
+        # call sites (main/install.py, _cmd_done_finalize.py, main/picks.py,
+        # _enforce_steps.py). state.current_app_id is already `chosen` here.
+        count = uninstall_other_games(allowed_app_ids(state))
         if count:
             _echo(f"\n  Uninstalled {count} non-assigned games")
     if not is_game_installed(chosen.app_id):

steam_backlog_enforcer/_scanning_tampering.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import logging
 from typing import TYPE_CHECKING, Any
 
+from steam_backlog_enforcer._allowed_games import allowed_games
 from steam_backlog_enforcer.config import load_snapshot
 from steam_backlog_enforcer.enforcer import send_notification
 from steam_backlog_enforcer.game_install import _echo
@@ -22,6 +23,30 @@ logger = logging.getLogger(__name__)
 _TAMPER_CHECK_LIMIT = 3
 
 
+def _legitimately_played(state: State) -> set[int]:
+    """Return every app id the user was allowed to earn achievements on.
+
+    Wider than ``allowed_app_ids`` on purpose. A manual pick that has been
+    completed is no longer *allowed* -- it leaves the active set the moment it
+    lands in ``finished_app_ids`` -- but the achievements that finished it were
+    earned legitimately, so comparing it against a stale snapshot would report
+    the user's own completion as tampering. Every game ever manually picked,
+    plus every finished game, is therefore exempt.
+
+    Args:
+        state: Current enforcer state.
+
+    Returns:
+        App ids whose achievement progress must not be treated as suspicious.
+    """
+    exempt = {app_id for app_id, _ in allowed_games(state)}
+    exempt.update(state.finished_app_ids)
+    exempt.update(
+        pick["app_id"] for pick in state.manual_picks if pick.get("app_id") is not None
+    )
+    return exempt
+
+
 def _check_game_tampering(
     client: SteamAPIClient,
     entry: dict[str, Any],
@@ -38,7 +63,7 @@ def _check_game_tampering(
         Tuple of (name, app_id, diff) if tampering detected, else None.
     """
     app_id = entry["app_id"]
-    if app_id == state.current_app_id:
+    if app_id in _legitimately_played(state):
         return None
     if entry["unlocked_achievements"] >= entry["total_achievements"]:
         return None

steam_backlog_enforcer/config.py

@@ -52,6 +52,10 @@ def _atomic_write(path: Path, data: str, *, mode: int | None = None) -> None:
         os.write(fd, data.encode("utf-8"))
         if mode is not None:
             os.fchmod(fd, mode)
+        # Keep the owner, or a root daemon write locks the user's CLI out.
+        with contextlib.suppress(OSError):  # absent file / non-root: skip
+            original = path.stat()
+            os.fchown(fd, original.st_uid, original.st_gid)
         os.close(fd)
         tmp_path.replace(path)
     except BaseException:

steam_backlog_enforcer/docs/DOCS-manual-pick-design.md

@@ -60,7 +60,28 @@ releases only when every pick is finished or past its own 14-day deadline.
   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.
+  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`/`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
+  `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.
+  - 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.
+  - `mark_finished` is the single writer of `finished_app_ids`, so the sweep
+    and `done`/`check` cannot both record the same completion.
 
 **Deployment note:** the enforce daemon holds the allowed set in code, so
 `sudo systemctl restart steam-backlog-enforcer` is required after upgrading.

steam_backlog_enforcer/main/picks.py

@@ -17,6 +17,10 @@ from steam_backlog_enforcer._allowed_games import (
 )
 from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
 from steam_backlog_enforcer._hltb_types import load_hltb_cache
+from steam_backlog_enforcer._pick_completion import (
+    report_completion,
+    warn_stale_assignment,
+)
 from steam_backlog_enforcer.config import load_snapshot
 from steam_backlog_enforcer.game_install import (
     _echo,
@@ -177,6 +181,8 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
 
     _echo(f"\nGame found: {game_name} (AppID={app_id})")
 
+    retired = report_completion(config, state)
+
     existing = _report_pick_slots(config, state)
 
     _echo(
@@ -189,12 +195,18 @@ def cmd_pick_manual(config: Config, state: State, args: list[str]) -> None:
         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:
+        _echo(
+            f"  - Uninstall {done_pick.game_name} (completed, no longer an"
+            " allowed game)"
+        )
     _echo()
     confirm = input(
         f"Type YES to confirm you will play {game_name} until completion: "
     ).strip()
     if confirm != "YES":
         _echo("Aborted.")
+        warn_stale_assignment(state, retired)
         return
 
     # State mutation is the shared, stdout-free core (also used by the MCP

steam_backlog_enforcer/main/status.py

@@ -8,6 +8,7 @@ from steam_backlog_enforcer._actions import (
     active_manual_picks,
     manual_pick_age_days,
 )
+from steam_backlog_enforcer._pick_completion import report_completion
 from steam_backlog_enforcer._total_block import get_total_block_status
 from steam_backlog_enforcer.config import load_snapshot
 from steam_backlog_enforcer.game_install import (
@@ -23,7 +24,7 @@ if TYPE_CHECKING:
     from steam_backlog_enforcer.config import Config, State
 
 
-def cmd_status(_config: Config, state: State) -> None:
+def cmd_status(config: Config, state: State) -> None:
     """Show current status."""
     _echo("=== Steam Backlog Enforcer ===\n")
 
@@ -53,6 +54,16 @@ def cmd_status(_config: Config, state: State) -> None:
         is_assigned_installed = any(aid == state.current_app_id for aid, _ in installed)
         _echo(f"Assigned game installed: {is_assigned_installed}")
 
+    # Retire any pick that has hit 100% since we last looked, so the list
+    # below reflects reality. Deliberately non-fatal: status is the
+    # always-available diagnostic (it is exempt from both the total block and
+    # the manual-pick lock), so a Steam outage or an unwritable state file must
+    # degrade to a printed note, never to a failed status.
+    try:
+        report_completion(config, state)
+    except (OSError, RuntimeError, ValueError) as exc:
+        _echo(f"\n(Could not refresh pick completion: {exc})")
+
     picks = active_manual_picks(state)
     if picks:
         _echo(f"\nManual picks ({len(picks)}):")

steam_backlog_enforcer/scanning.py

@@ -10,6 +10,7 @@ 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._scanning_assign import (
     _NO_CONF_MSG,
     _assign_chosen_game,
@@ -205,6 +206,7 @@ def pick_next_game(
 
 def do_check(config: Config, state: State) -> None:
     """Check assigned game completion status; detect tampering."""
+    report_completion(config, state)
     if state.current_app_id is None:
         _echo("No game currently assigned. Run 'scan' first.")
         return
@@ -224,7 +226,7 @@ def do_check(config: Config, state: State) -> None:
 
     if game.is_complete:
         _echo(f"\n  COMPLETED: {state.current_game_name}!")
-        state.finished_app_ids.append(state.current_app_id)
+        mark_finished(state, state.current_app_id)
         send_notification(
             "Game Complete!",
             f"You finished {state.current_game_name}! Picking next game...",

steam_backlog_enforcer/tests/test_do_enforce.py

@@ -151,3 +151,32 @@ class TestDoEnforce:
         ):
             do_enforce(config, state)
             mock_iter.assert_not_called()
+
+
+class TestStateReload:
+    """The loop must pick up CLI/MCP state changes without a restart."""
+
+    def test_reloads_manual_picks_from_disk(self) -> None:
+        # The MCP pick_manual tool adds a *second* pick without touching
+        # current_app_id. A daemon that never reloaded manual_picks would drop
+        # it out of the allowed set and uninstall it as unauthorized.
+        state = State(current_app_id=1, current_game_name="G")
+        fresh = State(
+            current_app_id=1,
+            current_game_name="G",
+            finished_app_ids=[7],
+            manual_picks=[{"app_id": 900, "game_name": "AddedByMcp", "started_at": ""}],
+        )
+        with (
+            patch(f"{PKG}._enforce_setup"),
+            patch(f"{PKG}._echo"),
+            patch.object(State, "load", return_value=fresh),
+            patch(
+                f"{PKG}._enforce_loop_iteration",
+                side_effect=KeyboardInterrupt,
+            ),
+            patch(f"{PKG}.time.sleep"),
+        ):
+            do_enforce(Config(), state)
+        assert state.manual_picks == fresh.manual_picks
+        assert state.finished_app_ids == [7]

steam_backlog_enforcer/tests/test_main_abandon.py

@@ -213,6 +213,10 @@ class TestAbandonOneOfTwoPicks:
                 "steam_backlog_enforcer.main.status.get_installed_games",
                 return_value=[],
             ),
+            patch(
+                "steam_backlog_enforcer.main.status.report_completion",
+                return_value=[],
+            ),
             patch("steam_backlog_enforcer.main.status._echo") as mock_echo,
         ):
             cmd_status(Config(), two_pick_state())
@@ -234,6 +238,10 @@ class TestAbandonOneOfTwoPicks:
                 "steam_backlog_enforcer.main.status.get_installed_games",
                 return_value=[],
             ),
+            patch(
+                "steam_backlog_enforcer.main.status.report_completion",
+                return_value=[],
+            ),
             patch("steam_backlog_enforcer.main.status._echo") as mock_echo,
         ):
             cmd_status(Config(), state)

steam_backlog_enforcer/tests/test_main_pick_cap.py

@@ -4,6 +4,7 @@ from unittest.mock import patch
 
 import pytest
 
+from steam_backlog_enforcer._pick_completion import PickProgress
 from steam_backlog_enforcer.config import Config, State
 from steam_backlog_enforcer.main import (
     cmd_pick_manual,
@@ -20,6 +21,7 @@ class TestPickManualCap:
     def test_second_pick_is_added_not_replacing(self) -> None:
         state = locked_state(app_id=100)
         with (
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
             patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
             patch("builtins.input", return_value="YES"),
@@ -34,6 +36,7 @@ class TestPickManualCap:
     def test_third_pick_refused_at_cap(self) -> None:
         state = two_pick_state()
         with (
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
             patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
             pytest.raises(SystemExit) as exc,
@@ -49,6 +52,7 @@ class TestPickManualCap:
         # The cap has room, so this exercises the core's own refusal path.
         state = locked_state(app_id=489830)
         with (
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
             patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
             patch("builtins.input", return_value="YES"),
@@ -68,6 +72,7 @@ class TestPickManualCap:
         # not uninstall or hide the first.
         state = locked_state(app_id=100)
         with (
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
             patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
             patch("builtins.input", return_value="YES"),
@@ -84,6 +89,7 @@ class TestPickManualCap:
     def test_installs_every_missing_allowed_game(self) -> None:
         state = locked_state(app_id=100)
         with (
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
             patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
             patch("builtins.input", return_value="YES"),
@@ -96,3 +102,28 @@ class TestPickManualCap:
             cmd_pick_manual(Config(max_manual_picks=2), state, ["489830"])
         installed = {call.args[0] for call in mock_install.call_args_list}
         assert installed == {100, 489830}
+
+
+class TestPickManualRetirementNotices:
+    """A pick retired by the sweep is announced before the YES prompt."""
+
+    def test_warns_about_uninstall_and_stale_assignment(self) -> None:
+        state = locked_state(app_id=100)
+        state.current_app_id = 200
+        state.current_game_name = "Finished"
+        retired = [PickProgress(200, "Finished", 5, 5, retired=True, determinable=True)]
+        with (
+            patch(f"{PKG}.report_completion", return_value=retired),
+            patch(f"{PKG}._echo") as mock_echo,
+            # warn_stale_assignment prints from the completion module.
+            patch("steam_backlog_enforcer._pick_completion._echo") as mock_note,
+            patch(f"{PKG}._resolve_game_name", return_value="Skyrim SE"),
+            patch("builtins.input", return_value="no"),
+        ):
+            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 "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)
+        assert "still" in note

steam_backlog_enforcer/tests/test_main_status.py

@@ -29,6 +29,7 @@ class TestCmdStatus:
         with (
             patch(f"{PKG}.is_store_blocked", return_value=True),
             patch(f"{PKG}.get_installed_games", return_value=[(440, "TF2")]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
         ):
             cmd_status(Config(), state)
@@ -37,6 +38,7 @@ class TestCmdStatus:
         with (
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
         ):
             cmd_status(Config(), State())
@@ -48,6 +50,7 @@ class TestCmdList:
     def test_no_snapshot(self) -> None:
         with (
             patch(f"{PKG}.load_snapshot", return_value=None),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_list(Config(), State())
@@ -62,6 +65,7 @@ class TestCmdList:
         state = State(current_app_id=1)
         with (
             patch(f"{PKG}.load_snapshot", return_value=snapshot),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo"),
         ):
             cmd_list(Config(), state)
@@ -70,6 +74,7 @@ class TestCmdList:
         snapshot = [snap(i, f"Game{i}") for i in range(60)]
         with (
             patch(f"{PKG}.load_snapshot", return_value=snapshot),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_list(Config(), State())
@@ -82,6 +87,7 @@ class TestCmdStatusLockHint:
         with (
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_status(Config(), state)
@@ -92,6 +98,7 @@ class TestCmdStatusLockHint:
         with (
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_status(Config(), State())
@@ -120,6 +127,7 @@ class TestCmdStatusTotalBlock:
             patch(f"{PKG}.get_total_block_status", return_value=ACTIVE_STATUS),
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_status(Config(), State())
@@ -132,6 +140,7 @@ class TestCmdStatusTotalBlock:
             patch(f"{PKG}.get_total_block_status", return_value=INACTIVE_STATUS),
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_status(Config(), State())
@@ -146,6 +155,7 @@ class TestCmdStatusTotalBlock:
             patch(f"{PKG}.get_total_block_status", return_value=status),
             patch(f"{PKG}.is_store_blocked", return_value=False),
             patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}.report_completion", return_value=[]),
             patch(f"{PKG}._echo") as mock_echo,
         ):
             cmd_status(Config(), State())
@@ -157,3 +167,19 @@ class TestCmdStatusTotalBlock:
 # ──────────────────────────────────────────────────────────────
 # main() dispatch to the daily-gaming-budget commands
 # ──────────────────────────────────────────────────────────────
+
+
+class TestCmdStatusCompletionSweepIsNonFatal:
+    """status stays usable when the completion sweep cannot run."""
+
+    def test_reports_failure_and_still_renders(self) -> None:
+        with (
+            patch(f"{PKG}.report_completion", side_effect=OSError("state read-only")),
+            patch(f"{PKG}.is_store_blocked", return_value=True),
+            patch(f"{PKG}.get_installed_games", return_value=[]),
+            patch(f"{PKG}._echo") as mock_echo,
+        ):
+            cmd_status(Config(), State(current_app_id=440, current_game_name="TF2"))
+        output = " ".join(str(c) for c in mock_echo.call_args_list)
+        assert "Could not refresh pick completion: state read-only" in output
+        assert "Store blocked" in output

steam_backlog_enforcer/tests/test_pick_completion.py

@@ -0,0 +1,230 @@
+"""Tests for achievement-based retirement of finished manual picks."""
+
+from unittest.mock import MagicMock, patch
+
+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.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:
+    """Build a GameInfo with the given achievement progress.
+
+    Args:
+        app_id: Steam application id.
+        unlocked: Unlocked achievement count.
+        total: Total achievement count.
+
+    Returns:
+        A GameInfo carrying that progress.
+    """
+    return GameInfo(
+        app_id=app_id,
+        name="G",
+        total_achievements=total,
+        unlocked_achievements=unlocked,
+        playtime_minutes=0,
+    )
+
+
+def client_returning(*games: GameInfo | None) -> MagicMock:
+    """Build a fake SteamAPIClient yielding *games* in order.
+
+    Args:
+        games: What each ``refresh_single_game`` call should return.
+
+    Returns:
+        A MagicMock standing in for SteamAPIClient.
+    """
+    fake = MagicMock()
+    fake.refresh_single_game.side_effect = list(games)
+    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:
+        line = PickProgress(1, "G", 5, 5, retired=True, determinable=True).describe()
+        assert "COMPLETE, freeing its slot" 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
+
+    def test_in_progress_with_zero_total(self) -> None:
+        # Defensive: determinable but no achievements must not divide by zero.
+        line = PickProgress(1, "G", 0, 0, retired=False, determinable=True).describe()
+        assert "(0%)" in line
+
+
+class TestRetireCompletedManualPicks:
+    def test_no_picks_makes_no_api_call(self) -> None:
+        fake = client_returning()
+        assert retire_completed_manual_picks(Config(), State(), client=fake) == []
+        fake.refresh_single_game.assert_not_called()
+
+    def test_complete_pick_is_retired_and_saved(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, 5, 5))
+            )
+        assert [r.retired for r in results] == [True]
+        assert state.finished_app_ids == [100]
+        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:
+            results = retire_completed_manual_picks(
+                Config(), state, client=client_returning(game(100, 1, 5))
+            )
+        assert [r.retired for r in results] == [False]
+        assert state.finished_app_ids == []
+        mock_save.assert_not_called()
+
+    def test_unreadable_pick_stays_locked(self) -> None:
+        # No achievements at all, or Steam unreachable: both surface as 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(None)
+            )
+        assert results == [
+            PickProgress(100, "TestGame", 0, 0, retired=False, determinable=False)
+        ]
+        assert state.finished_app_ids == []
+        mock_save.assert_not_called()
+
+    def test_reports_every_pick_not_only_retirements(self) -> None:
+        state = two_pick_state()
+        with patch.object(State, "save"):
+            results = retire_completed_manual_picks(
+                Config(),
+                state,
+                client=client_returning(game(100, 1, 5), game(200, 5, 5)),
+            )
+        assert [(r.app_id, r.retired) for r in results] == [(100, False), (200, True)]
+
+    def test_pick_without_app_id_is_skipped(self) -> None:
+        state = State(manual_picks=[{"game_name": "Broken", "started_at": ""}])
+        fake = client_returning()
+        assert retire_completed_manual_picks(Config(), state, client=fake) == []
+        fake.refresh_single_game.assert_not_called()
+
+    def test_unnamed_pick_falls_back_to_app_id(self) -> None:
+        state = State(manual_picks=[{"app_id": 100, "game_name": "", "started_at": ""}])
+        results = retire_completed_manual_picks(
+            Config(), state, client=client_returning(game(100, 1, 5))
+        )
+        assert results[0].game_name == "AppID=100"
+
+    def test_builds_its_own_client_when_none_given(self) -> None:
+        state = locked_state(app_id=100)
+        fake = client_returning(game(100, 1, 5))
+        with patch(f"{PKG}.SteamAPIClient", return_value=fake) as mock_cls:
+            retire_completed_manual_picks(
+                Config(steam_api_key="k", steam_id="i"), state
+            )
+        mock_cls.assert_called_once_with("k", "i")
+
+
+class TestReportCompletion:
+    def test_prints_nothing_without_picks(self) -> None:
+        with patch(f"{PKG}._echo") as mock_echo:
+            assert report_completion(Config(), State()) == []
+        mock_echo.assert_not_called()
+
+    def test_prints_each_pick_and_returns_only_retirements(self) -> None:
+        state = two_pick_state()
+        with (
+            patch.object(State, "save"),
+            patch(
+                f"{PKG}.SteamAPIClient",
+                return_value=client_returning(game(100, 1, 5), game(200, 5, 5)),
+            ),
+            patch(f"{PKG}._echo") as mock_echo,
+        ):
+            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
+
+
+class TestWarnStaleAssignment:
+    def test_warns_when_retired_pick_is_current(self) -> None:
+        state = State(current_app_id=200, current_game_name="SecondGame")
+        retired = [
+            PickProgress(200, "SecondGame", 5, 5, retired=True, determinable=True)
+        ]
+        with patch(f"{PKG}._echo") as mock_echo:
+            warn_stale_assignment(state, retired)
+        assert "still" in " ".join(str(c) for c in mock_echo.call_args_list)
+
+    def test_silent_when_current_is_something_else(self) -> None:
+        state = State(current_app_id=999)
+        retired = [PickProgress(200, "G", 5, 5, retired=True, determinable=True)]
+        with patch(f"{PKG}._echo") as mock_echo:
+            warn_stale_assignment(state, retired)
+        mock_echo.assert_not_called()
+
+
+class TestDaemonSweep:
+    def test_due_then_throttled(self) -> None:
+        sweep = _DaemonSweep()
+        assert sweep.due(1000.0, 900.0) is True
+        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:
+        state = locked_state(app_id=100)
+        daemon_sweep.last = None
+        daemon_sweep.retired.clear()
+        try:
+            with (
+                patch.object(State, "save"),
+                patch(
+                    f"{PKG}.SteamAPIClient",
+                    return_value=client_returning(game(100, 5, 5)),
+                ),
+            ):
+                results = retire_completed_manual_picks_throttled(Config(), state)
+                assert [r.retired for r in results] == [True]
+                assert daemon_sweep.retired == {100}
+                # 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

steam_backlog_enforcer/tests/test_scanning_part2.py

@@ -38,6 +38,48 @@ class TestCheckGameTampering:
         result = _check_game_tampering(MagicMock(), _entry(app_id=1), state)
         assert result is None
 
+    def test_finished_manual_pick_is_not_tampering(self) -> None:
+        # The completion that retired the pick is legitimate progress; the
+        # snapshot is simply stale. Reported as TAMPERING before this exemption.
+        client = MagicMock()
+        game = MagicMock()
+        game.unlocked_achievements = 16
+        client.refresh_single_game.return_value = game
+        state = State(
+            current_app_id=475150,
+            finished_app_ids=[588730],
+            manual_picks=[
+                {"app_id": 588730, "game_name": "Majotori", "started_at": ""}
+            ],
+        )
+        entry = _entry(app_id=588730, name="Majotori", total=16, unlocked=13)
+        assert _check_game_tampering(client, entry, state) is None
+
+    def test_active_second_pick_is_not_tampering(self) -> None:
+        # A second manual pick is allowed but is not current_app_id.
+        client = MagicMock()
+        game = MagicMock()
+        game.unlocked_achievements = 9
+        client.refresh_single_game.return_value = game
+        state = State(
+            current_app_id=475150,
+            manual_picks=[
+                {"app_id": 412830, "game_name": "STEINS;GATE", "started_at": ""}
+            ],
+        )
+        entry = _entry(app_id=412830, name="STEINS;GATE", total=42, unlocked=5)
+        assert _check_game_tampering(client, entry, state) is None
+
+    def test_unrelated_game_still_flagged(self) -> None:
+        # The detector must keep working for games never allowed.
+        client = MagicMock()
+        game = MagicMock()
+        game.unlocked_achievements = 8
+        client.refresh_single_game.return_value = game
+        state = State(current_app_id=475150, finished_app_ids=[588730])
+        entry = _entry(app_id=813780, name="AoE2", unlocked=5)
+        assert _check_game_tampering(client, entry, state) == ("AoE2", 813780, 3)
+
     def test_already_complete_skipped(self) -> None:
         state = State()
         result = _check_game_tampering(

steam_backlog_enforcer/tests/test_scanning_pick_next_game.py

@@ -2,13 +2,18 @@
 
 from __future__ import annotations
 
+from typing import Any
 from unittest.mock import patch
 
+import pytest
+
 from steam_backlog_enforcer.config import Config, State
+from steam_backlog_enforcer.game_uninstall import uninstall_other_games
 from steam_backlog_enforcer.scanning import (
     pick_next_game,
 )
 from steam_backlog_enforcer.steam_api import GameInfo
+from steam_backlog_enforcer.tests._main_helpers import STARTED_AT
 
 
 def _game(
@@ -156,3 +161,75 @@ class TestPickNextGame:
         ):
             pick_next_game([g1], state, config)
             mock_install.assert_called_once()
+
+
+class TestAssignUninstallsAgainstTheAllowedSet:
+    """_assign_chosen_game must pass the allowed *set*, never a bare app id."""
+
+    def test_passes_set_including_concurrent_manual_pick(self) -> None:
+        g = _game(app_id=2, name="Short", hours=10.0)
+        config = Config(steam_api_key="k", steam_id="i")
+        state = State(
+            manual_picks=[
+                {"app_id": 900, "game_name": "Pick", "started_at": STARTED_AT}
+            ],
+        )
+        with (
+            patch(
+                "steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
+                side_effect=lambda c: c[0] if c else None,
+            ),
+            patch("steam_backlog_enforcer.scanning._echo"),
+            patch("steam_backlog_enforcer._scanning_confidence._echo"),
+            patch(
+                "steam_backlog_enforcer._scanning_assign.is_game_installed",
+                return_value=True,
+            ),
+            patch(
+                "steam_backlog_enforcer._scanning_assign.uninstall_other_games",
+                return_value=0,
+            ) as mock_uninstall,
+            patch("builtins.input", return_value="1"),
+        ):
+            pick_next_game([g], state, config)
+        (allowed,) = mock_uninstall.call_args.args
+        assert isinstance(allowed, set)
+        # The new assignment survives, and so does the untouched manual pick.
+        assert allowed == {2, 900}
+
+    def test_real_uninstall_accepts_that_argument(self) -> None:
+        """Guards the contract the mock above cannot: a bare int raises.
+
+        ``uninstall_other_games`` membership-tests every installed game against
+        its argument, so an int made it raise TypeError as soon as anything was
+        installed. The suite hid that because no games are installed by default.
+        """
+        with (
+            patch(
+                "steam_backlog_enforcer.game_uninstall.get_installed_games",
+                return_value=[(2, "Short"), (900, "Pick"), (55, "Other")],
+            ),
+            patch(
+                "steam_backlog_enforcer.game_uninstall.is_protected_app",
+                return_value=False,
+            ),
+            patch(
+                "steam_backlog_enforcer.game_uninstall.uninstall_game",
+                return_value=True,
+            ) as mock_rm,
+        ):
+            assert uninstall_other_games({2, 900}) == 1
+        assert [c.args[0] for c in mock_rm.call_args_list] == [55]
+
+        # Typed as Any so the wrong-type call is a runtime check, not a
+        # suppressed type error.
+        bare_app_id: Any = 2
+        with (
+            patch(
+                "steam_backlog_enforcer.game_uninstall.get_installed_games",
+                return_value=[(2, "Short")],
+            ),
+            patch("steam_backlog_enforcer.game_uninstall.uninstall_game"),
+            pytest.raises(TypeError),
+        ):
+            uninstall_other_games(bare_app_id)