shadow: no exception can escape into the enforce loop; warn once per pricing

observe_shadow caught only OSError/ValueError, so a RecursionError from
leisure_watch escaped into the unguarded root loop and crash-looped the
daemon, stopping enforced billing. It now catches Exception, fails closed
(failclosed:watcher, unless the pause check itself answered "paused") and
logs each distinct error at most once per 10 minutes.

resolve_budget priced the same answers twice (enforced + leisure), so
earned_time's "No answer for the X earner" warning fired twice per tick.
The shadow pricing now gets the answers completed with None, which
earned_time treats identically to missing, so only the first warns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N

steam_backlog_enforcer/_budget_resolve.py

@@ -188,7 +188,10 @@ def resolve_budget(config: Config) -> BudgetResolution:
     first_paid = first_credits(registry, today)
     day = _price(answers, today, registry, first_paid, leisure=False)
     # The same answers, asked once: the shadow can never see a different day.
-    shadow = _price(answers, today, registry, first_paid, leisure=True)
+    # A missing answer was already warned about above; to earned_time it is
+    # the same as None, so spell it out rather than warn a second time.
+    complete = {e.name: None for e in registry} | answers
+    shadow = _price(complete, today, registry, first_paid, leisure=True)
     earned = {t.earner.name: t.gaming_minutes * _SECONDS_PER_MINUTE for t in day.terms}
     total = day.gaming_minutes * _SECONDS_PER_MINUTE
     parts = ", ".join(

steam_backlog_enforcer/_shadow.py

@@ -26,6 +26,7 @@ from __future__ import annotations
 from dataclasses import dataclass, replace
 import logging
 from pathlib import Path
+import time
 from typing import TYPE_CHECKING, Final
 
 from leisure_watch.facts import read_now
@@ -47,6 +48,16 @@ FAILCLOSED_KEY: Final = "failclosed:watcher"
 
 logger = logging.getLogger(__name__)
 
+# A broken input fails every 3 s tick: one warning per distinct error per
+# window, not 1200 identical journal lines an hour.
+WARN_EVERY_SECONDS: Final = 600.0
+_last_warned: Final[dict[str, float]] = {}
+
+
+def reset_warnings() -> None:
+    """Forget which errors were logged (tests; a long-lived daemon never)."""
+    _last_warned.clear()
+
 
 @dataclass(frozen=True)
 class ShadowVerdict:
@@ -95,19 +106,35 @@ def observe_shadow(owners: dict[int, str], *, stamp: float) -> ShadowVerdict:
     Returns:
         The tick's shadow verdict; the fail-closed one if an input raised.
     """
-    # The enforce loop has no per-tick guard, so a raise here would crash-loop
-    # the daemon and stop *enforced* billing. An unreadable input bills, unpaused.
+    # The enforce loop has no per-tick guard, so a raise here -- of *any* kind:
+    # leisure_watch has raised RecursionError -- would crash-loop the daemon
+    # and stop enforced billing. An unreadable input bills, unpaused, unless
+    # the pause check itself answered "paused".
     try:
         paused = paused_by(CLAIMS_DIR)
         if paused is not None:
             return ShadowVerdict(paused_by=paused, keys=())
         fact = read_now(NOW_PATH, stamp, max_age=_MAX_AGE_SECONDS)
-    except (OSError, ValueError) as exc:
-        logger.warning("Shadow inputs unreadable (%s); failing closed.", exc)
+    except Exception as exc:  # the loop must outlive any shadow input bug
+        if _should_warn(exc):
+            logger.warning(
+                "Shadow inputs unreadable (%r); failing closed.", exc, exc_info=True
+            )
         fact = None
     return ShadowVerdict(paused_by=None, keys=shadow_keys(owners, fact))
 
 
+def _should_warn(exc: BaseException) -> bool:
+    """Whether *exc* was not already logged within :data:`WARN_EVERY_SECONDS`."""
+    key = f"{type(exc).__name__}: {exc}"
+    now = time.monotonic()
+    last = _last_warned.get(key)
+    if last is not None and now - last < WARN_EVERY_SECONDS:
+        return False
+    _last_warned[key] = now
+    return True
+
+
 def shadow_accumulate(
     state: PlaytimeState,
     *,

steam_backlog_enforcer/tests/test_shadow_budget.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
+import logging
 from typing import TYPE_CHECKING
 from unittest.mock import patch
 
@@ -66,6 +67,19 @@ class TestShadowResolution:
         resolved = _resolve(done=False)
         assert resolved.shadow_seconds == resolved.seconds + 2 * _HOUR
 
+    def test_a_missing_answer_warns_once_not_once_per_pricing(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Enforced + shadow pricing is one resolution: one warning per earner."""
+        with (
+            caplog.at_level(logging.WARNING, logger="earned_time"),
+            patch("steam_backlog_enforcer._budget_resolve._answers", return_value={}),
+        ):
+            resolve_budget(Config())
+        warned = [r.getMessage() for r in caplog.records if "No answer" in r.message]
+        assert warned
+        assert len(warned) == len(set(warned))
+
 
 class TestHeldShadow:
     def test_same_day_holds_the_high_water(self) -> None:

steam_backlog_enforcer/tests/test_shadow_guard.py

@@ -0,0 +1,131 @@
+"""The shadow path can never raise into the enforce loop (no per-tick guard).
+
+Any exception from ``leisure_watch`` (a ``RecursionError`` was observed, not
+just ``OSError``/``ValueError``) must fail the shadow closed, log at most once
+per distinct error per window, and leave the enforced count untouched.
+"""
+
+from __future__ import annotations
+
+from contextlib import ExitStack
+from datetime import datetime, timedelta, timezone
+import logging
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from steam_backlog_enforcer import _shadow
+from steam_backlog_enforcer._playtime import playtime_tick
+from steam_backlog_enforcer._playtime_state import (
+    PlaytimeState,
+    load_state,
+    save_state,
+)
+from steam_backlog_enforcer._shadow import FAILCLOSED_KEY, ShadowVerdict, observe_shadow
+from steam_backlog_enforcer.config import Config
+from steam_backlog_enforcer.tests._fake_session import fake_session
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+PKG = "steam_backlog_enforcer._playtime"
+NOW = datetime(2026, 7, 27, 20, 0, tzinfo=timezone(timedelta(hours=2)))
+T0 = NOW.timestamp()
+
+
+@pytest.fixture(autouse=True)
+def _fresh_warnings() -> Iterator[None]:
+    """Each test starts with no error already logged."""
+    _shadow.reset_warnings()
+    yield
+    _shadow.reset_warnings()
+
+
+@pytest.fixture
+def quiet_tick() -> Iterator[None]:
+    """Neutralise the tick's side effects (cutoff, mounts, total block)."""
+    cutoff = "steam_backlog_enforcer._playtime_cutoff"
+    with ExitStack() as stack:
+        for name in ("reconcile", "request_steam_shutdown", "kill_gaming_processes"):
+            stack.enter_context(patch(f"{cutoff}.{name}"))
+        stack.enter_context(patch(f"{cutoff}.notify_desktop_user"))
+        stack.enter_context(patch(f"{PKG}.mounted_targets", return_value=set()))
+        stack.enter_context(patch(f"{PKG}.is_total_block_active", return_value=False))
+        yield
+
+
+@pytest.mark.parametrize("where", ["paused_by", "read_now"])
+@pytest.mark.parametrize("exc", [RecursionError("deep"), RuntimeError("boom")])
+def test_any_exception_fails_closed(where: str, exc: Exception) -> None:
+    with patch.object(_shadow, where, side_effect=exc):
+        game = observe_shadow({1: "app:440"}, stamp=T0)
+        idle = observe_shadow({}, stamp=T0)
+    assert game == ShadowVerdict(paused_by=None, keys=("app:440",))
+    assert idle == ShadowVerdict(paused_by=None, keys=(FAILCLOSED_KEY,))
+
+
+def test_a_successful_pause_check_still_pauses() -> None:
+    """Only the fact read failed: the lock answer stands, nothing bills."""
+    with (
+        patch.object(_shadow, "paused_by", return_value="i3lock"),
+        patch.object(_shadow, "read_now", side_effect=RecursionError("deep")),
+    ):
+        assert observe_shadow({}, stamp=T0) == ShadowVerdict("i3lock", ())
+
+
+def _warnings(caplog: pytest.LogCaptureFixture) -> list[str]:
+    return [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
+
+
+def test_logs_once_per_distinct_error_per_window(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    errors = [RecursionError("deep")] * 3 + [RuntimeError("other")]
+    with (
+        caplog.at_level(logging.WARNING, logger=_shadow.__name__),
+        patch.object(_shadow, "read_now", side_effect=errors),
+        patch.object(_shadow.time, "monotonic", return_value=100.0),
+    ):
+        for _ in errors:
+            observe_shadow({}, stamp=T0)
+    logged = _warnings(caplog)
+    assert len(logged) == 2
+    assert "deep" in logged[0]
+    assert "other" in logged[1]
+
+
+def test_logs_again_after_the_window(caplog: pytest.LogCaptureFixture) -> None:
+    window = _shadow.WARN_EVERY_SECONDS
+    clock = [100.0, 100.0 + window - 1, 100.0 + window + 1]
+    with (
+        caplog.at_level(logging.WARNING, logger=_shadow.__name__),
+        patch.object(_shadow, "read_now", side_effect=RecursionError("deep")),
+        patch.object(_shadow.time, "monotonic", side_effect=clock),
+    ):
+        for _ in clock:
+            observe_shadow({}, stamp=T0)
+    assert len(_warnings(caplog)) == 2
+
+
+@pytest.mark.usefixtures("quiet_tick")
+@pytest.mark.parametrize("where", ["paused_by", "read_now"])
+@pytest.mark.parametrize(
+    ("owners", "seconds", "shadow"),
+    [({7: "app:440"}, 3.0, "app:440"), ({}, 0.0, FAILCLOSED_KEY)],
+)
+def test_tick_completes_and_enforced_count_is_untouched(
+    where: str, owners: dict[int, str], seconds: float, shadow: str
+) -> None:
+    save_state(PlaytimeState(day_key="2026-07-27", last_tick_at=T0 - 3.0), demo=False)
+    with (
+        patch.object(_shadow, where, side_effect=RecursionError("deep")),
+        patch(f"{PKG}.qualifying_pids", return_value=owners),
+        patch(f"{PKG}.datetime") as mock_dt,
+    ):
+        mock_dt.now.return_value = NOW
+        playtime_tick(Config(), interval=3.0, session=fake_session())
+    state = load_state(demo=False)
+    assert state.seconds == seconds
+    assert state.last_tick_at == T0
+    assert state.shadow_per_source == {shadow: 3.0}