restart safety: own scope for launched Steam, hold the day's budget, re-arm warnings after a raise

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

steam_backlog_enforcer/_budget_resolve.py

@@ -19,21 +19,16 @@ bonus must not interfere with the workout" means in code.
 same as a "no". The difference is only in what gets reported: an unreadable
 LeetCode ledger raises an incident, an honest "not solved yet" does not.
 
-**Rising, in normal use -- but not guaranteed.** Every earner only ever goes
-false->true within a day, so in the ordinary case the budget starts at the floor
-and rises. That is a property of the *inputs*, not something enforced here:
-nothing persists a per-day high-water mark, so anything that changes a resolved
-answer mid-day -- changing the registry, or deploying a change onto a day
-already in progress -- lowers the budget immediately and re-prices time already
-spent. Seconds already accrued are then measured against the new, smaller
-budget, and the cutoff fires on the next tick with no warning first, because
-``warned_seconds`` records thresholds by seconds *remaining* and remaining has
-already gone negative.
-
-That was a deliberate call (2026-08-29): the alternative is a persisted
-``granted_budget_seconds`` on ``PlaytimeState`` taking ``max(resolved,
-granted)``. If a mid-day drop ever bites, that is the fix -- not a nudge to
-these numbers.
+**Rising, and held there.** Every earner only ever goes false->true within a
+day, so in the ordinary case the budget starts at the floor and rises. This
+resolution alone does not guarantee it: earners reset at calendar midnight
+while the gaming day runs to 05:59, and a cold restart resolves with empty
+answer caches, so a live answer can come in lower than one given hours ago.
+That bit twice (2026-10-03 and 2026-10-04 00:00, 8h -> 4h, unwarned cutoff).
+``PlaytimeState.budget_seconds`` is the day's high-water mark, and
+``rules_for`` applies ``max(resolved + carry, held)`` via
+``_gaming_days.held_today`` -- so the daemon, ``/api/budget`` and MCP all see
+the held figure. Only the 06:00 roll-over lowers it.
 
 **One seam.** :func:`resolve_budget` is called from ``rules_for`` and nowhere
 else, so the enforcing daemon and the read-only HTTP/MCP views resolve the same

steam_backlog_enforcer/_gaming_days.py

@@ -145,3 +145,23 @@ def carry_into(stored: PlaytimeState | None, now: datetime) -> float:
         spent=stored.seconds,
         today=gaming_day_key(now),
     )
+
+
+def held_today(stored: PlaytimeState | None, now: datetime) -> float:
+    """Return the budget already granted earlier in *now*'s gaming day.
+
+    Earners reset at calendar midnight but the gaming day runs to 05:59, and a
+    cold restart starts with empty answer caches, so a live re-resolution can
+    come in lower than what was granted hours ago. ``rules_for`` holds the day
+    at this high-water mark so neither can cut a session off unwarned.
+
+    Args:
+        stored: The persisted production state, or ``None`` if unreadable.
+        now: Local timestamp for this tick.
+
+    Returns:
+        The recorded high-water budget; 0 when the state is another day's.
+    """
+    if stored is None or stored.day_key != gaming_day_key(now):
+        return 0.0
+    return stored.budget_seconds

steam_backlog_enforcer/_playtime.py

@@ -219,6 +219,12 @@ def _policy(
         reconcile(should_block=False)
         if state.is_blocked():
             state = _release_after_raise(state, rules)
+        # A raise that never reached the cutoff lifts remaining back above
+        # thresholds already fired; re-arm them or the new end comes unwarned.
+        remaining = rules.budget_seconds - state.seconds
+        rearmed = [t for t in state.warned_seconds if t >= remaining]
+        if rearmed != state.warned_seconds:
+            state = replace(state, warned_seconds=rearmed)
         return _warn(state, rules)
 
     if not state.is_blocked():

steam_backlog_enforcer/_playtime_state.py

@@ -14,7 +14,7 @@ import logging
 from typing import TYPE_CHECKING
 
 from steam_backlog_enforcer._budget_resolve import resolve_budget
-from steam_backlog_enforcer._gaming_days import carry_into
+from steam_backlog_enforcer._gaming_days import carry_into, held_today
 from steam_backlog_enforcer._whitelist_locking import (
     _try_set_immutable,
     unlock_for_write,
@@ -151,20 +151,19 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
     Returns:
         The rules governing this tick.
     """
-    # Production budget is resolved, not read straight off config: it is a
-    # floor plus a bonus per earner. Resolving it *here* rather than at each
-    # caller is what stops the daemon (which holds one Config for its whole
-    # life) and _budget_view (which reloads Config per HTTP request) from
-    # reporting different budgets.
+    # Resolved *here*, not per caller, so the daemon (one Config for life) and
+    # _budget_view (Config per request) can never report different budgets.
     resolved = None if demo else resolve_budget(config)
-    # Carry sits on top of the 8h cap: it was earned and not spent, so it
-    # comes with no strings attached.
+    # Carry sits on top of the 8h cap (earned, not spent: no strings). The
+    # day never re-prices below what it already granted -- see held_today.
     now = datetime.now().astimezone()
-    carry = 0.0 if demo else carry_into(load_state(demo=False), now)
+    stored = None if demo else load_state(demo=False)
+    carry = 0.0 if demo else carry_into(stored, now)
+    live = 0.0 if resolved is None else resolved.seconds + carry
+    held = held_today(stored, now)
+    held_note = f"; held at {held / 3600:.1f}h (earned earlier today)"
     return PlaytimeRules(
-        budget_seconds=_DEMO_BUDGET_SECONDS
-        if resolved is None
-        else resolved.seconds + carry,
+        budget_seconds=_DEMO_BUDGET_SECONDS if resolved is None else max(live, held),
         warn_at=_DEMO_WARN_AT if demo else _WARN_AT,
         sigkill_after=(_DEMO_SIGKILL_AFTER_SECONDS if demo else _SIGKILL_AFTER_SECONDS),
         count_launchers=config.count_launcher_processes,
@@ -181,6 +180,7 @@ def rules_for(config: Config, *, demo: bool) -> PlaytimeRules:
             if resolved is None
             else resolved.reason
             + (f", +{carry / 60:.0f}m carried over" if carry > 0 else "")
+            + (held_note if held > live else "")
         ),
         carry_seconds=carry,
     )

steam_backlog_enforcer/_steam_process.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import logging
 import os
+import shutil
 import subprocess
 
 from steam_backlog_enforcer._desktop_env import desktop_user_cmd
@@ -25,7 +26,14 @@ def spawn_detached(cmd: list[str]) -> subprocess.Popen[bytes]:
     Fire-and-forget on purpose: the process must outlive the calling scope
     (Steam itself, a desktop notifier), so it is polled or reaped later rather
     than waited on in a ``with`` block here.
+
+    The unit is KillMode=control-group, so a Steam started from here used to
+    stay in the service's cgroup and a daemon restart SIGTERMed it together
+    with the running game (2026-07-26, 2026-10-02). A transient scope moves it
+    out; ``--scope`` execs in place, so the returned handle is still the child.
     """
+    if os.geteuid() == 0 and shutil.which("systemd-run"):
+        cmd = ["systemd-run", "--scope", "--quiet", "--collect", *cmd]
     return subprocess.Popen(
         cmd,
         stdout=subprocess.DEVNULL,

steam_backlog_enforcer/tests/_no_systemd_run.py

@@ -0,0 +1,30 @@
+"""Autouse: the host's ``systemd-run`` never leaks into launch argv.
+
+``spawn_detached`` wraps root launches in a transient scope when
+``systemd-run`` is on PATH. That is a fact about the machine, so every test
+that asserts a launch command would pass or fail by host. Pinned absent here;
+``test_steam_process_scope`` patches it back in to test the wrapper itself.
+
+Split out of ``conftest.py``, whose isolation ``with`` is already at Python's
+static nesting limit; ``conftest`` imports the fixture by name to register it.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@pytest.fixture(autouse=True)
+def _no_systemd_run() -> Iterator[None]:
+    """Report ``systemd-run`` missing to ``_steam_process`` for one test."""
+    with patch(
+        "steam_backlog_enforcer._steam_process.shutil",
+        **{"which.return_value": None},
+    ):
+        yield

steam_backlog_enforcer/tests/conftest.py

@@ -26,6 +26,7 @@ from steam_backlog_enforcer.tests._isolate_playtime import _isolate_playtime
 from steam_backlog_enforcer.tests._isolate_reading import _isolate_reading
 from steam_backlog_enforcer.tests._no_fast_install import _no_fast_install
 from steam_backlog_enforcer.tests._no_subprocess import _block_real_subprocesses
+from steam_backlog_enforcer.tests._no_systemd_run import _no_systemd_run
 from steam_backlog_enforcer.tests._no_workout_http import _no_workout_http
 
 # Re-exported so ruff --fix does not delete the imports above: pytest
@@ -36,6 +37,7 @@ __all__ = [
     "_isolate_playtime",
     "_isolate_reading",
     "_no_fast_install",
+    "_no_systemd_run",
     "_no_workout_http",
 ]
 

steam_backlog_enforcer/tests/test_budget_view.py

@@ -33,6 +33,7 @@ from steam_backlog_enforcer.tests._no_workout_http import earned_budget, fixed_b
 # fold whatever Fri-Mon carry it holds today into RULES.
 with (
     patch("steam_backlog_enforcer._playtime_state.carry_into", return_value=0.0),
+    patch("steam_backlog_enforcer._playtime_state.held_today", return_value=0.0),
     patch(
         "steam_backlog_enforcer._playtime_state.resolve_budget",
         side_effect=earned_budget,

steam_backlog_enforcer/tests/test_gaming_days.py

@@ -6,7 +6,12 @@ from dataclasses import replace
 from datetime import datetime, timedelta, timezone
 from unittest.mock import patch
 
-from steam_backlog_enforcer._gaming_days import carry_due, carry_into, pass_on
+from steam_backlog_enforcer._gaming_days import (
+    carry_due,
+    carry_into,
+    held_today,
+    pass_on,
+)
 from steam_backlog_enforcer._playtime_budget import roll_over
 from steam_backlog_enforcer._playtime_state import PlaytimeState, rules_for
 from steam_backlog_enforcer.config import Config
@@ -116,3 +121,41 @@ class TestRulesCarry:
             "steam_backlog_enforcer._playtime_state.carry_into", return_value=1200.0
         ):
             assert rules_for(Config(), demo=True).carry_seconds == 0.0
+
+
+class TestHeldToday:
+    """The day's high-water budget, and only for the day it was recorded."""
+
+    NOW = datetime(2026, 10, 5, 0, 30, tzinfo=LOCAL)  # still gaming day SUN
+
+    def test_no_state_holds_nothing(self) -> None:
+        assert held_today(None, self.NOW) == 0.0
+
+    def test_same_gaming_day_returns_the_high_water(self) -> None:
+        stored = PlaytimeState(day_key=SUN, budget_seconds=8 * HOUR)
+        assert held_today(stored, self.NOW) == 8 * HOUR
+
+    def test_yesterdays_record_holds_nothing(self) -> None:
+        stored = PlaytimeState(day_key=SAT, budget_seconds=8 * HOUR)
+        assert held_today(stored, self.NOW) == 0.0
+
+
+class TestRulesHold:
+    """rules_for never prices the day below what it already granted."""
+
+    def test_midnight_reset_is_held_and_says_so(self) -> None:
+        with patch(
+            "steam_backlog_enforcer._playtime_state.held_today",
+            return_value=10 * HOUR,
+        ):
+            rules = rules_for(Config(), demo=False)
+        assert rules.budget_seconds == 10 * HOUR
+        assert rules.budget_reason.endswith("; held at 10.0h (earned earlier today)")
+
+    def test_a_higher_live_answer_wins_silently(self) -> None:
+        with patch(
+            "steam_backlog_enforcer._playtime_state.held_today", return_value=HOUR
+        ):
+            rules = rules_for(Config(), demo=False)
+        assert rules.budget_seconds > HOUR
+        assert "held at" not in rules.budget_reason

steam_backlog_enforcer/tests/test_playtime_policy.py

@@ -6,6 +6,7 @@ tests exercise the loop that feeds it.
 """
 
 from contextlib import ExitStack
+from dataclasses import replace
 from datetime import (
     datetime,
     timedelta,
@@ -175,10 +176,36 @@ class TestPolicyBudgetRoseMidDay:
 
     def test_an_unblocked_day_is_left_alone(self) -> None:
         """The release path must not fire on an ordinary under-budget tick."""
-        state = PlaytimeState(day_key=TODAY, seconds=100.0, warned_seconds=[3600])
+        spent = 8 * 3600 - 3000.0
+        state = PlaytimeState(day_key=TODAY, seconds=spent, warned_seconds=[3600])
         with (
             patch("steam_backlog_enforcer._playtime.reconcile"),
             patch("steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"),
         ):
             out = _policy(state, _rules(), now=NOW)
         assert out.warned_seconds == [3600]
+        assert out.blocked_at == 0.0
+
+    def test_a_raise_before_the_cutoff_rearms_warnings(self) -> None:
+        """All four fired at a 5h budget, then it rose to 8h (2026-10-04).
+
+        No block ever engaged, so the release path never ran; without the
+        re-arm the new end of the day would come with no warning at all.
+        """
+        state = PlaytimeState(
+            day_key=TODAY, seconds=5 * 3600.0, warned_seconds=[3600, 1800, 600, 300]
+        )
+        with (
+            patch("steam_backlog_enforcer._playtime.reconcile"),
+            patch(
+                "steam_backlog_enforcer._playtime_cutoff.notify_desktop_user"
+            ) as mock_notify,
+        ):
+            out = _policy(state, _rules(), now=NOW)
+            mock_notify.assert_not_called()
+            drained = _policy(
+                replace(out, seconds=8 * 3600 - 3500.0), _rules(), now=NOW
+            )
+        assert out.warned_seconds == []
+        assert drained.warned_seconds == [3600]
+        mock_notify.assert_called_once()

steam_backlog_enforcer/tests/test_steam_process_scope.py

@@ -0,0 +1,40 @@
+"""Tests for moving spawned processes out of the service cgroup.
+
+The unit is KillMode=control-group: a Steam left in the daemon's cgroup died
+with the running game on every ``systemctl restart``. ``spawn_detached`` wraps
+root launches in a transient scope; these pin when it does and when it must not.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from steam_backlog_enforcer._steam_process import spawn_detached
+
+MOD = "steam_backlog_enforcer._steam_process"
+SCOPE = ["systemd-run", "--scope", "--quiet", "--collect"]
+
+
+def _argv(*, euid: int, which: str | None) -> list[str]:
+    with (
+        patch(f"{MOD}.os.geteuid", return_value=euid),
+        patch(f"{MOD}.shutil.which", return_value=which),
+        patch(f"{MOD}.subprocess.Popen") as mock_popen,
+    ):
+        spawn_detached(["steam", "-silent"])
+    return mock_popen.call_args[0][0]
+
+
+class TestSpawnDetachedScope:
+    """Root launches leave the cgroup; anything else is spawned as asked."""
+
+    def test_root_with_systemd_run_gets_its_own_scope(self) -> None:
+        argv = _argv(euid=0, which="/usr/bin/systemd-run")
+        assert argv == [*SCOPE, "steam", "-silent"]
+
+    def test_root_without_systemd_run_launches_plainly(self) -> None:
+        assert _argv(euid=0, which=None) == ["steam", "-silent"]
+
+    def test_unprivileged_launch_is_never_wrapped(self) -> None:
+        argv = _argv(euid=1000, which="/usr/bin/systemd-run")
+        assert argv == ["steam", "-silent"]