shutdown bonus: apply the moment a gate writes its ledger (earner-bonus.path)

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

.gitignore

@@ -27,6 +27,7 @@ screen_locker/sick_history.json
 screen_locker/extra_benefits_state.json
 screen_locker/scheduled_skips.json
 screen_locker/shutdown_base.json
+screen_locker/shutdown_base.lock
 screen_locker/sick_day_state.json
 screen_locker/early_bird_pending.json
 

arm.sh

@@ -33,13 +33,16 @@ readonly UNITS=(
 	"early-bird-workout-check.timer"
 	"workout-sync.service"
 	"workout-sync.timer"
+	"earner-bonus.service"
 )
 
-# Only the timers get enabled; the service is triggered by them and by login.
+# Only the timers (and the ledger watch) get enabled; the services are
+# triggered by them and by login.
 readonly TIMERS=(
 	"early-bird-workout-check.timer"
 	"workout-locker.timer"
 	"workout-sync.timer"
+	"earner-bonus.path"
 )
 
 SKIP_TODAY=0
@@ -87,6 +90,8 @@ main() {
 	for unit in "${UNITS[@]}"; do
 		cp "$SCRIPT_DIR/$unit" "$USER_UNIT_DIR/$unit"
 	done
+	# Rendered from the earned_time registry rather than copied.
+	PYTHONPATH="$SCRIPT_DIR" python3 -m screen_locker._earner_units "$USER_UNIT_DIR"
 
 	systemctl --user daemon-reload
 

install_systemd.sh

@@ -13,6 +13,8 @@ EARLY_BIRD_TIMER_NAME="early-bird-workout-check.timer"
 LOCKER_TIMER_NAME="workout-locker.timer"
 SYNC_SERVICE_NAME="workout-sync.service"
 SYNC_TIMER_NAME="workout-sync.timer"
+BONUS_SERVICE_NAME="earner-bonus.service"
+BONUS_PATH_NAME="earner-bonus.path"
 
 # Runtime dependencies. screen_lock.py imports tkinter at module scope. On Arch
 # the tkinter MODULE ships inside the `python` package, but the shared library
@@ -113,6 +115,12 @@ cp "$LOCKER_TIMER_FILE" "$USER_SERVICE_DIR/$LOCKER_TIMER_NAME"
 cp "$SYNC_SERVICE_FILE" "$USER_SERVICE_DIR/$SYNC_SERVICE_NAME"
 cp "$SYNC_TIMER_FILE" "$USER_SERVICE_DIR/$SYNC_TIMER_NAME"
 
+# The ledger watch: a gate writing its ledger applies the earned shutdown
+# bonus at once instead of at the next 15-minute sync. The .path unit is
+# rendered from the earned_time registry, so it is generated, not copied.
+cp "$SCRIPT_DIR/$BONUS_SERVICE_NAME" "$USER_SERVICE_DIR/$BONUS_SERVICE_NAME"
+PYTHONPATH="$SCRIPT_DIR" python3 -m screen_locker._earner_units "$USER_SERVICE_DIR"
+
 # Update paths in the service file to use absolute paths
 REPO_ROOT="$SCRIPT_DIR"
 sed -i "s|WorkingDirectory=.*|WorkingDirectory=$REPO_ROOT|" "$USER_SERVICE_DIR/$SERVICE_NAME"
@@ -134,6 +142,9 @@ systemctl --user enable --now "$LOCKER_TIMER_NAME"
 # Enable the periodic workout sync
 systemctl --user enable --now "$SYNC_TIMER_NAME"
 
+# Enable the instant bonus pass on ledger writes
+systemctl --user enable --now "$BONUS_PATH_NAME"
+
 # Verify enforcement is actually armed. enable can silently no-op when systemd
 # breaks an ordering cycle by deleting the timer's job -- exactly what happened
 # on 2026-08-04 -- so a successful `enable` is NOT proof the locker will run.

remove_systemd.sh

@@ -13,6 +13,8 @@ UNITS=(
 	"early-bird-workout-check.timer"
 	"workout-sync.timer"
 	"workout-sync.service"
+	"earner-bonus.path"
+	"earner-bonus.service"
 )
 
 for unit in "${UNITS[@]}"; do

screen_locker/_cli.py

@@ -13,8 +13,10 @@ from pathlib import Path
 import sys
 from typing import TYPE_CHECKING
 
+from screen_locker._constants import SHUTDOWN_BASE_FILE
 from screen_locker._decision_log import record_no_decision
 from screen_locker._manual_cli import run_manual_log
+from screen_locker._shutdown_base import apply_flat_bonuses_if_new
 from screen_locker._status import run_status
 
 if TYPE_CHECKING:
@@ -22,6 +24,8 @@ if TYPE_CHECKING:
 
 __all__ = ["main"]
 
+_logger = logging.getLogger(__name__)
+
 _LOG_FILE_NAME = "log.json"
 _LOG_MANUAL_FLAG = "--log-manual-workout"
 
@@ -73,6 +77,15 @@ def main(locker_cls: type[ScreenLocker], argv: list[str]) -> None:
         record_no_decision("--sync-only")
         sys.exit(0)
 
+    if "--apply-bonuses" in argv:
+        # earner-bonus.service, started by earner-bonus.path when a gate
+        # writes its ledger: only the flat-bonus pass, no phone sync, so a
+        # solve moves shutdown within seconds rather than at the next sync.
+        _logger.info("Ledger changed: applying any newly earned flat bonus.")
+        apply_flat_bonuses_if_new(SHUTDOWN_BASE_FILE, _headless_locker(locker_cls))
+        record_no_decision("--apply-bonuses")
+        sys.exit(0)
+
     locker = locker_cls(
         demo_mode="--production" not in argv,
         verify_only="--verify-workout" in argv,

screen_locker/_shutdown_base.py

@@ -33,6 +33,7 @@ from typing import TYPE_CHECKING, Any
 
 import earned_time
 
+from screen_locker._bonus_lock import bonus_lock
 from screen_locker._day import today_str
 from screen_locker._earned import (
     earned_today,
@@ -137,6 +138,17 @@ def reset_to_base_if_new_day(
 
     Returns True if a reset was performed, False if today was already reset.
     """
+    with bonus_lock(state_file):
+        return _reset(state_file, mixin, sick_day_state_file, log_file)
+
+
+def _reset(
+    state_file: Path,
+    mixin: object,
+    sick_day_state_file: Path | None,
+    log_file: Path | None,
+) -> bool:
+    """:func:`reset_to_base_if_new_day`'s body; the caller holds the lock."""
     today = today_str()
     if _load_state(state_file).get("last_reset_date") == today:
         return False
@@ -227,5 +239,6 @@ def apply_flat_bonuses_if_new(state_file: Path, mixin: object) -> None:
     The workout is not among them: it is counted, and its live credit is
     applied by :class:`~screen_locker._workout_credit.WorkoutCreditMixin`.
     """
-    for earner in flat_earners():
-        _apply_flat_bonus(state_file, mixin, earner)
+    with bonus_lock(state_file):
+        for earner in flat_earners():
+            _apply_flat_bonus(state_file, mixin, earner)

screen_locker/_sync_mixin.py

@@ -12,6 +12,7 @@ import logging
 
 from screen_locker._constants import EXTRA_BENEFITS_FILE, SHUTDOWN_BASE_FILE
 from screen_locker._earned import span
+from screen_locker._earner_units import sync_path_unit
 from screen_locker._extra_benefits import weekly_shutdown_bonus_hours
 from screen_locker._manual_push import push_pc_workouts
 from screen_locker._manual_sync import ingest_manual_records
@@ -53,6 +54,8 @@ class SyncMixin:
         # Same reason: a LeetCode solve or a reading session made after login
         # must not wait for the next login to earn its hour.
         apply_flat_bonuses_if_new(SHUTDOWN_BASE_FILE, self)
+        # The ledger watch that makes that pass instant follows the registry.
+        sync_path_unit()
 
     def _ingest_synced_manual_workouts(self) -> None:
         """Sync manual workouts: publish this PC's, ingest everyone else's.

screen_locker/tests/conftest.py

@@ -192,10 +192,10 @@ def _isolate_shutdown_base(tmp_path: Path) -> Iterator[None]:
 
     Pre-seeded with today's date so reset_to_base_if_new_day() is a no-op by
     default (matching the real file's steady state) -- tests that want to
-    exercise the actual reset path patch reset_to_base_if_new_day directly,
-    same as the rest of the suite already does. Every flat earner's stamp is
-    seeded too, so apply_flat_bonuses_if_new() never reads a ledger (or the
-    host's real HMAC key) by default.
+    exercise the actual reset path patch reset_to_base_if_new_day directly.
+    Every flat earner's stamp is seeded too, so apply_flat_bonuses_if_new()
+    never reads a ledger (or the real HMAC key), and sync_now's ledger-watch
+    refresh sees an empty unit dir, never the real one.
     """
     target = tmp_path / "shutdown_base.json"
     today = datetime.now(tz=UTC).strftime("%Y-%m-%d")
@@ -204,6 +204,8 @@ def _isolate_shutdown_base(tmp_path: Path) -> Iterator[None]:
         patch("screen_locker._constants.SHUTDOWN_BASE_FILE", target),
         patch("screen_locker._startup_checks.SHUTDOWN_BASE_FILE", target),
         patch("screen_locker._sync_mixin.SHUTDOWN_BASE_FILE", target),
+        patch("screen_locker._cli.SHUTDOWN_BASE_FILE", target),
+        patch("screen_locker._earner_units.USER_UNIT_DIR", tmp_path / "units"),
     ):
         yield
 

screen_locker/tests/test_cli.py

@@ -129,6 +129,28 @@ class TestSyncOnlyMode:
         assert synced == [True]
 
 
+class TestApplyBonusesMode:
+    """--apply-bonuses runs only the flat-bonus pass, headless, and exits 0."""
+
+    def test_applies_flat_bonuses_then_exits_zero(self) -> None:
+        """The pass gets the state file and a headless locker; nothing syncs."""
+        with (
+            patch.object(_cli.logging, "basicConfig"),
+            patch.object(_cli.sys, "exit") as sys_exit,
+            patch.object(_cli, "apply_flat_bonuses_if_new") as apply_bonuses,
+            patch.object(_cli, "record_no_decision") as no_decision,
+        ):
+            _cli.main(_FakeLocker, ["screen_lock.py", "--apply-bonuses"])
+
+        state_file, locker = apply_bonuses.call_args.args
+        assert state_file == _cli.SHUTDOWN_BASE_FILE
+        assert isinstance(locker, _FakeLocker)
+        # Headless: __init__ never ran, so no UI and no lock screen.
+        assert not hasattr(locker, "demo_mode")
+        no_decision.assert_called_once_with("--apply-bonuses")
+        sys_exit.assert_any_call(0)
+
+
 class TestLockMode:
     """With no subcommand flag, the real locker is built and run."""
 

screen_locker/tests/test_screen_lock_coverage_part3.py

@@ -189,7 +189,10 @@ class TestSyncNow:
                 "screen_locker._sync_mixin.ingest_manual_records",
                 return_value=["manual:x"],
             ) as ingest,
+            patch("screen_locker._sync_mixin.sync_path_unit") as refresh,
         ):
             locker.sync_now()
         push.assert_called_once()
         ingest.assert_called_once()
+        # Every sync pass keeps the ledger watch in step with the registry.
+        refresh.assert_called_once_with()

screen_locker/tests/test_shutdown_base.py

@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+from contextlib import nullcontext
 from datetime import UTC, datetime
 import json
 from typing import TYPE_CHECKING
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 from screen_locker._shutdown_base import (
     _load_state,
@@ -155,7 +156,11 @@ class TestResetToBaseIfNewDay:
         state_mock.open.side_effect = OSError("disk full")
         mixin = self._make_mixin()
         # _write_shutdown_config succeeds, so True is returned even if state write fails.
-        assert reset_to_base_if_new_day(state_mock, mixin) is True
+        # The lock file would come from the same failing mock; lock elsewhere.
+        with patch(
+            "screen_locker._shutdown_base.bonus_lock", return_value=nullcontext()
+        ):
+            assert reset_to_base_if_new_day(state_mock, mixin) is True
 
     def test_sick_day_state_file_not_deleted_when_absent(self, tmp_path: Path) -> None:
         """No sick-day file passed → branch skipped, no AttributeError (line 79 branch False)."""

scripts/disarm_guard.sh

@@ -56,6 +56,8 @@ clear_disarm_marker() {
 		"early-bird-workout-check.timer"
 		"workout-sync.service"
 		"workout-sync.timer"
+		"earner-bonus.path"
+		"earner-bonus.service"
 	)
 
 	if [[ -f $DISARM_MARKER ]]; then