diff --git a/AGENTS.md b/AGENTS.md index bae85f9..60c1a56 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,7 +47,12 @@ of that derivation gets wiped by the next reset -- that is how a 00:02 workout lost its hours on 2026-09-13. A new bonus source is an `Earner` in the registry, never code here: the reset includes it as a term, and the live pass (`apply_flat_bonuses_if_new`, stamped `_bonus_date` in -`shutdown_base.json`) picks it up for flat earners. The base is the registry +`shutdown_base.json`) picks it up for flat earners. That pass runs the moment +a gate writes its ledger: `earner-bonus.path` (rendered from the registry by +`_earner_units`, re-synced by every `--sync-only` pass) starts +`earner-bonus.service` (`--apply-bonuses`). The 15-minute `workout-sync` is +only the fallback. The reset and the pass hold `_bonus_lock`, since three +processes can run them at once. The base is the registry constant, not the state file: the file once persisted `base_*_hour`, which made the code's default dead. diff --git a/earner-bonus.service b/earner-bonus.service new file mode 100644 index 0000000..8245d5d --- /dev/null +++ b/earner-bonus.service @@ -0,0 +1,11 @@ +[Unit] +Description=Apply newly earned flat shutdown bonuses (LeetCode, reading, Anki, ...) + +[Service] +Type=oneshot +# Started by earner-bonus.path the moment a gate writes its ledger. Only the +# flat-bonus pass: no phone sync, no Tk, no lock screen, so a solve moves the +# shutdown time within seconds instead of at the next 15-minute workout-sync. +# The path unit itself is generated from the earned_time registry +# (screen_locker._earner_units), so it is not a file in this repo. +ExecStart=/usr/bin/python3 -m screen_locker.screen_lock --apply-bonuses diff --git a/screen_locker/_bonus_lock.py b/screen_locker/_bonus_lock.py new file mode 100644 index 0000000..6f439c9 --- /dev/null +++ b/screen_locker/_bonus_lock.py @@ -0,0 +1,38 @@ +"""One writer at a time for the shutdown bonus state. + +The daily reset and the live bonus pass both read ``shutdown_base.json``, +push ``/etc/shutdown-schedule.conf`` with a read-add-write and stamp the file. +They run from three processes -- the locker at login, the 15-minute +``workout-sync`` pass and the ledger-triggered ``earner-bonus`` pass -- so two +of them can see the same unstamped earner and each add its time. Holding this +lock across check, write and stamp makes the second one see the stamp. + +A dedicated file, not the state file itself: ``_save_state`` truncates that. +""" + +from __future__ import annotations + +from contextlib import contextmanager +import fcntl +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +__all__ = ["bonus_lock"] + + +@contextmanager +def bonus_lock(state_file: Path) -> Iterator[None]: + """Hold an exclusive lock beside *state_file* for the ``with`` body. + + Blocks until any other holder is done; the passes it serialises take a + second or two, so waiting is cheaper than skipping a bonus. + """ + with state_file.with_suffix(".lock").open("a") as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) diff --git a/screen_locker/_earner_units.py b/screen_locker/_earner_units.py new file mode 100644 index 0000000..ba248fa --- /dev/null +++ b/screen_locker/_earner_units.py @@ -0,0 +1,99 @@ +"""The ``earner-bonus.path`` unit: apply a flat bonus the moment it is earned. + +The live bonus pass used to run only on the 15-minute ``workout-sync`` timer, +so a LeetCode solve could wait a quarter of an hour for its shutdown hour +while the gaming hour landed in seconds (2026-10-06). The path unit watches +every flat earner's ledger and starts ``earner-bonus.service`` +(``--apply-bonuses``) when a gate writes one. + +The watch list is rendered from the ``earned_time`` registry, not kept by +hand. Re-pinning the registry is a pip install that never re-runs the +installer, so :func:`sync_path_unit` -- called from every ``--sync-only`` +pass -- rewrites an installed unit that has drifted from the registry. A new +earner is watched within 15 minutes with no step to remember. + +``PathChanged=`` and not ``PathModified=``: the gates write a temp file and +rename it into place, which ``PathChanged`` sees, including the first write +into a directory that does not exist yet. ``PathModified`` would also fire +mid-write on a gate that writes in place, and a half-written ledger reads as +"could not check". +""" + +from __future__ import annotations + +import logging +from pathlib import Path +import subprocess +import sys +from typing import Final + +from screen_locker._earned import flat_earners, ledger_file + +__all__ = ["PATH_UNIT_NAME", "USER_UNIT_DIR", "render_path_unit", "sync_path_unit"] + +_logger = logging.getLogger(__name__) + +PATH_UNIT_NAME: Final = "earner-bonus.path" +USER_UNIT_DIR: Final = Path.home() / ".config" / "systemd" / "user" + +_HEADER: Final = """\ +# Generated by screen_locker._earner_units from the earned_time registry. +# Do not edit: the next --sync-only pass rewrites a unit that has drifted. +[Unit] +Description=Apply a flat earner's shutdown bonus as soon as its ledger changes + +[Path] +""" +_FOOTER: Final = """ +[Install] +WantedBy=default.target +""" + + +def render_path_unit() -> str: + """The unit text: one ``PathChanged=`` per flat earner's ledger.""" + watches = "".join(f"PathChanged={ledger_file(e)}\n" for e in flat_earners()) + return _HEADER + watches + _FOOTER + + +def sync_path_unit(unit_dir: Path | None = None) -> bool: + """Rewrite an installed path unit that no longer matches the registry. + + An absent unit is left absent: it was never installed, or it was removed + on purpose, and installing units is the installer's job. *unit_dir* + defaults to :data:`USER_UNIT_DIR`, read at call time. + + Returns True if the unit was rewritten. + """ + unit = (unit_dir or USER_UNIT_DIR) / PATH_UNIT_NAME + if not unit.is_file(): + return False + wanted = render_path_unit() + try: + if unit.read_text(encoding="utf-8") == wanted: + return False + unit.write_text(wanted, encoding="utf-8") + subprocess.run( + ["/usr/bin/systemctl", "--user", "daemon-reload"], + check=True, + capture_output=True, + ) + except (OSError, subprocess.CalledProcessError) as exc: + _logger.warning( + "%s differs from the earner registry and could not be refreshed: %s" + " -- a newly registered earner waits for the 15-minute sync", + unit, + exc, + ) + return False + _logger.warning( + "%s was out of date with the earner registry; rewritten and reloaded", + unit, + ) + return True + + +if __name__ == "__main__": + # install_systemd.sh / arm.sh: write the unit into the given directory. + target = Path(sys.argv[1]) / PATH_UNIT_NAME + target.write_text(render_path_unit(), encoding="utf-8") diff --git a/screen_locker/tests/test_earner_units.py b/screen_locker/tests/test_earner_units.py new file mode 100644 index 0000000..90fdbe7 --- /dev/null +++ b/screen_locker/tests/test_earner_units.py @@ -0,0 +1,153 @@ +"""Tests for the ledger watch (``_earner_units``) and the bonus lock.""" + +from __future__ import annotations + +import fcntl +import subprocess +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from screen_locker import _earner_units +from screen_locker._bonus_lock import bonus_lock +from screen_locker._earned import flat_earners, ledger_file +from screen_locker._shutdown_base import ( + apply_flat_bonuses_if_new, + reset_to_base_if_new_day, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def _installed(unit_dir: Path, text: str) -> Path: + """Write *text* as the installed path unit and return its path.""" + unit_dir.mkdir(parents=True, exist_ok=True) + unit = unit_dir / _earner_units.PATH_UNIT_NAME + unit.write_text(text, encoding="utf-8") + return unit + + +class TestRenderPathUnit: + """The unit watches exactly the registry's flat ledgers.""" + + def test_one_path_changed_per_flat_earner(self) -> None: + """Every flat earner's ledger is watched, in registry order.""" + lines = _earner_units.render_path_unit().splitlines() + watched = [ln.removeprefix("PathChanged=") for ln in lines if "=" in ln] + expected = [str(ledger_file(e)) for e in flat_earners()] + assert [w for w in watched if w in expected] == expected + assert sum(ln.startswith("PathChanged=") for ln in lines) == len(expected) + + def test_never_path_modified(self) -> None: + """PathModified would fire on a half-written ledger.""" + assert "PathModified" not in _earner_units.render_path_unit() + + def test_is_installable(self) -> None: + """An [Install] section, so `enable` has a target.""" + assert "[Install]\nWantedBy=default.target" in _earner_units.render_path_unit() + + +class TestSyncPathUnit: + """The sync pass repairs a drifted unit and leaves an absent one alone.""" + + def test_absent_unit_is_not_installed(self, tmp_path: Path) -> None: + """Removed on purpose stays removed.""" + with patch.object(_earner_units.subprocess, "run") as run: + assert _earner_units.sync_path_unit(tmp_path) is False + run.assert_not_called() + assert not (tmp_path / _earner_units.PATH_UNIT_NAME).exists() + + def test_defaults_to_the_patched_user_unit_dir(self) -> None: + """No argument reads USER_UNIT_DIR at call time (conftest: empty).""" + assert _earner_units.sync_path_unit() is False + + def test_current_unit_is_left_alone(self, tmp_path: Path) -> None: + """No drift, no write, no daemon-reload.""" + _installed(tmp_path, _earner_units.render_path_unit()) + with patch.object(_earner_units.subprocess, "run") as run: + assert _earner_units.sync_path_unit(tmp_path) is False + run.assert_not_called() + + def test_drifted_unit_is_rewritten_and_reloaded( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A missing watch is restored, systemd reloaded, and it says so.""" + unit = _installed(tmp_path, "[Path]\nPathChanged=/old\n") + with patch.object(_earner_units.subprocess, "run") as run: + assert _earner_units.sync_path_unit(tmp_path) is True + assert unit.read_text(encoding="utf-8") == _earner_units.render_path_unit() + run.assert_called_once_with( + ["/usr/bin/systemctl", "--user", "daemon-reload"], + check=True, + capture_output=True, + ) + assert "out of date" in caplog.text + + def test_failed_reload_is_loud( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A reload that fails is a warning, not a silent False.""" + _installed(tmp_path, "stale") + err = subprocess.CalledProcessError(1, "systemctl") + with patch.object(_earner_units.subprocess, "run", side_effect=err): + assert _earner_units.sync_path_unit(tmp_path) is False + assert "could not be refreshed" in caplog.text + + def test_unreadable_unit_is_loud( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """An OSError reading the unit is a warning too.""" + _installed(tmp_path, "stale") + with patch.object( + _earner_units.Path, "read_text", side_effect=OSError("denied") + ): + assert _earner_units.sync_path_unit(tmp_path) is False + assert "denied" in caplog.text + + +class TestBonusLock: + """Both bonus writers hold one exclusive lock beside the state file.""" + + def test_takes_and_releases_an_exclusive_lock(self, tmp_path: Path) -> None: + """LOCK_EX for the body, LOCK_UN afterwards, on the .lock file.""" + state = tmp_path / "shutdown_base.json" + with patch.object(fcntl, "flock") as flock: + with bonus_lock(state): + assert flock.call_args.args[1] == fcntl.LOCK_EX + assert flock.call_args.args[1] == fcntl.LOCK_UN + assert (tmp_path / "shutdown_base.lock").exists() + + def test_released_when_the_body_raises(self, tmp_path: Path) -> None: + """An exception inside still unlocks.""" + with patch.object(fcntl, "flock") as flock: + with pytest.raises(RuntimeError), bonus_lock(tmp_path / "s.json"): + raise RuntimeError + assert flock.call_args.args[1] == fcntl.LOCK_UN + + def test_held_by_the_live_pass(self, tmp_path: Path) -> None: + """apply_flat_bonuses_if_new runs under the lock.""" + state = tmp_path / "s.json" + seen: list[bool] = [] + with ( + patch( + "screen_locker._shutdown_base.bonus_lock", + return_value=MagicMock(__enter__=lambda _s: seen.append(True)), + ) as lock, + patch("screen_locker._shutdown_base.flat_earners", return_value=()), + ): + apply_flat_bonuses_if_new(state, object()) + lock.assert_called_once_with(state) + assert seen == [True] + + def test_held_by_the_daily_reset(self, tmp_path: Path) -> None: + """reset_to_base_if_new_day runs under the lock.""" + state = tmp_path / "s.json" + with ( + patch("screen_locker._shutdown_base.bonus_lock") as lock, + patch("screen_locker._shutdown_base._reset", return_value=True) as reset, + ): + assert reset_to_base_if_new_day(state, object()) is True + lock.assert_called_once_with(state) + reset.assert_called_once_with(state, reset.call_args.args[1], None, None)