earned_time 0.6.0: maturity-aware penalty start (first_credits), pin v0.6.0
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (mypy, mypy, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (pylint, pylint, ) (push) Waiting to run
pre-commit / pre-commit (${{ matrix.group }}) (rest, , mypy,pylint) (push) Waiting to run
Tests / test (push) Waiting to run
Tests / web-coverage (push) Waiting to run
repo-contract / repo contract (push) Waiting to run

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-09 21:42:59 +02:00
parent 51285d60b7
commit 7ec94f6e4a
9 changed files with 247 additions and 17 deletions

View File

@ -110,7 +110,7 @@ repos:
- --disable-error-code=list-item
additional_dependencies:
- types-requests
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.5.0#subdirectory=earned_time
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.6.0#subdirectory=earned_time
# ===========================================================================
# PYLINT
@ -130,7 +130,7 @@ repos:
# (steam_backlog_enforcer._mcp_server).
additional_dependencies:
- aiohttp==3.14.4
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.5.0#subdirectory=earned_time
- earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.6.0#subdirectory=earned_time
- howlongtobeatpy==1.0.23
- mcp==2.3.0
- pytest

View File

@ -16,7 +16,7 @@ dependencies = [
"websockets==17.2",
# The earner registry (gaming budget terms); stdlib-only, so the root
# daemon imports it from the desktop user's site-packages.
"earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.5.0#subdirectory=earned_time",
"earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.6.0#subdirectory=earned_time",
]
# Optional extras. The `mcp` group is ONLY needed by the MCP server

View File

@ -5,7 +5,7 @@ aiohttp==3.14.4
bandit==1.9.4
codespell==2.4.3
coverage==7.16.2
earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.5.0#subdirectory=earned_time
earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.6.0#subdirectory=earned_time
howlongtobeatpy==1.0.24
mcp==2.3.0
mypy==2.4.0

View File

@ -7,7 +7,9 @@ is its sum, capped at the gaming ceiling (8h):
base (4h) + workout (2h) + LeetCode (1h) + reading (1h) + ...
The base is 5h minus every penalised earner's cut once its day arrives
(book-guard's reading hour, from 2026-10-01). The earners are read
(book-guard's reading hour, from 2026-10-01) -- and, from earned_time 0.6, not
before the day after the gate's first real credit
(:func:`steam_backlog_enforcer._ledger_earners.first_credits`). The earners are read
**independently** -- :mod:`steam_backlog_enforcer._workout_budget`,
:mod:`steam_backlog_enforcer._leetcode_bonus`,
:mod:`steam_backlog_enforcer._reading_bonus` and, for any earner registered
@ -46,6 +48,7 @@ from typing import TYPE_CHECKING, Final
import earned_time
from steam_backlog_enforcer._ledger_earners import (
first_credits,
ledger_answer,
ledger_units,
registry_for,
@ -156,7 +159,14 @@ def resolve_budget(config: Config) -> BudgetResolution:
# the earners asked and the earners summed can never differ.
today = datetime.now().astimezone().date()
registry = registry_for(today)
day = earned_time.resolve(_answers(config, registry), today, registry)
answers = _answers(config, registry)
# Maturity only delays a penalty: a gate that never paid out costs nothing.
first_paid = first_credits(registry, today)
day = (
earned_time.resolve(answers, today, registry)
if first_paid is None
else earned_time.resolve(answers, today, registry, first_credits=first_paid)
)
earned = {t.earner.name: t.gaming_minutes * _SECONDS_PER_MINUTE for t in day.terms}
total = day.gaming_minutes * _SECONDS_PER_MINUTE
parts = ", ".join(

View File

@ -9,6 +9,10 @@ A *counted* gate (the Automation tutor, from ``earned_time.TUTOR_FROM``) pays
per unit -- each verified 15-minute block -- so :func:`ledger_units` reads its
unit count for the day instead of a yes/no.
:func:`first_credits` feeds ``earned_time.resolve`` each penalised earner's
first real credit (earned_time >= 0.6), so a gate that never paid out costs
nothing yet: maturity can only delay a penalty, never add one.
Paths resolve under :data:`LEDGER_HOME` -- the unit sets ``HOME`` to the
desktop user's, and the test suite redirects this constant.
"""
@ -19,13 +23,16 @@ from datetime import date, datetime
import logging
from pathlib import Path
import time
from typing import Final
from typing import TYPE_CHECKING, Final, cast
import earned_time
from steam_backlog_enforcer._leetcode_ledger import HMAC_KEY_FILE
from steam_backlog_enforcer._status_api import AnswerCache
if TYPE_CHECKING:
from collections.abc import Callable
logger = logging.getLogger(__name__)
LEDGER_HOME: Final = Path.home()
@ -54,6 +61,37 @@ def registry_for(day: date) -> tuple[earned_time.Earner, ...]:
return tuple(earned_time.EARNERS if fn is None else fn(day))
def first_credits(
registry: tuple[earned_time.Earner, ...], day: date
) -> dict[str, date | None] | None:
"""Each penalised earner's first real credit up to ``day``, for ``resolve``.
Keyed by name over ``registry`` -- the tuple the answers came from, so the
two ``automation`` earners (one per side of ``TUTOR_FROM``) never collide.
An earner without ``penalty_from``, a ledger or a matcher is left out:
``resolve`` then falls back to its ``confirmed_on`` (fail closed).
Returns:
The map, or ``None`` on an earned_time without ``maturity`` (< 0.6),
where the caller resolves exactly as before.
"""
maturity = cast(
"Callable[..., earned_time.Maturity] | None",
getattr(earned_time, "maturity", None),
)
if maturity is None:
return None
return {
item.name: maturity(
item, LEDGER_HOME / item.ledger, HMAC_KEY_FILE, day
).first_credit
for item in registry
if item.penalty_from is not None
and item.ledger is not None
and item.match is not None
}
def _no_ledger(earner: earned_time.Earner) -> None:
logger.warning(
"Earner %s has no ledger and no reader here; it earns nothing",

View File

@ -1,9 +1,10 @@
"""Pin the day ``resolve_budget`` believes it is, around the 5h -> 4h base cut.
``earned_time`` takes the reading hour off the floor from
``READING.penalty_from`` on. A budget test that reads the real clock would
change its answer on 2026-10-01, so every such test pins the date on one side
of the cut.
``earned_time`` takes the reading hour off the floor from its penalty start
on: ``READING.penalty_from``, delayed by maturity (0.6.0) to the day after
``confirmed_on`` when no credit is on record. A budget test that reads the
real clock would change its answer on that day, so every such test pins the
date on one side of the cut.
The module asks ``datetime.now().astimezone().date()``, so that chain is what
the pin answers.
"""
@ -20,7 +21,9 @@ if TYPE_CHECKING:
from contextlib import AbstractContextManager
from datetime import date
_CUT = earned_time.READING.penalty_from
# With no ledger on record (the suite's empty LEDGER_HOME) the cut waits for
# the day after READING.confirmed_on: fail closed, never before penalty_from.
_CUT = earned_time.penalty_start(earned_time.READING, None)
assert _CUT is not None, "earned_time.READING lost its penalty_from date"
AFTER_CUT = _CUT

View File

@ -87,13 +87,13 @@ class TestTheBase:
"""The floor carries the old extra hour until book-guard's gate starts."""
def test_the_day_before_the_cut_keeps_the_old_floor(self) -> None:
"""2026-09-30 still gets the hour that reading cannot yet earn back."""
"""The day before still gets the hour reading cannot yet earn back."""
with pin_today(BEFORE_CUT):
resolved = _resolve(earners=(False, False, False))
assert resolved.base_seconds == 5 * _HOUR
def test_the_cut_day_itself_uses_the_lower_floor(self) -> None:
"""The boundary is inclusive: 2026-10-01 is the first 4h day."""
"""The boundary is inclusive: the penalty start is the first 4h day."""
with pin_today(AFTER_CUT):
resolved = _resolve(earners=(False, False, False))
assert resolved.base_seconds == 4 * _HOUR

View File

@ -0,0 +1,173 @@
"""Tests for _ledger_earners.first_credits: maturity only delays a penalty.
A stand-in earner (``piano``) is registered with a penalty. Its signed ledger
lives under the suite's redirected ``LEDGER_HOME``; whether it has ever paid
out decides the first day its cut applies (earned_time 0.6.0), and on an
earned_time without ``maturity`` the budget resolves exactly as 0.5.0 did.
"""
from __future__ import annotations
from dataclasses import replace
from datetime import datetime, time, timedelta
from typing import TYPE_CHECKING
from unittest.mock import patch
import earned_time
import pytest
from steam_backlog_enforcer import _ledger_earners, _leetcode_bonus, _workout_budget
from steam_backlog_enforcer._budget_resolve import resolve_budget
from steam_backlog_enforcer._ledger_earners import first_credits
from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.tests._budget_dates import AFTER_CUT, pin_today
from steam_backlog_enforcer.tests._ledger_fixtures import KEY, credit, write_ledger
if TYPE_CHECKING:
from collections.abc import Iterator
from datetime import date
from pathlib import Path
_WORKOUT = "steam_backlog_enforcer._workout_budget._fetch_workout_today"
_LEETCODE = "steam_backlog_enforcer._leetcode_bonus.read_ledger_solved_today"
_ANSWER = "steam_backlog_enforcer._budget_resolve.ledger_answer"
_HOUR = 3600.0
_DAY = timedelta(days=1)
def _any_credit(_row: dict[str, object], _window: tuple[float, float]) -> bool:
"""Count every verified credit: the reader's own dating is earned_time's."""
return True
# Penalised since well before the suite's pinned day, never confirmed by kuhy:
# without a real credit it is a new gate and costs nothing.
PIANO = earned_time.Earner(
name="piano",
label="Piano",
gaming_minutes=30,
shutdown_minutes=30,
ledger=".local/share/piano_guard/ledger.json",
match=_any_credit,
penalty_from=AFTER_CUT - 5 * _DAY,
)
@pytest.fixture(autouse=True)
def _clear_caches() -> Iterator[None]:
"""Keep every earner memo from leaking answers between tests.
Yields:
None, with the caches empty before and after.
"""
for module in (_ledger_earners, _workout_budget, _leetcode_bonus):
module.reset_cache()
yield
for module in (_ledger_earners, _workout_budget, _leetcode_bonus):
module.reset_cache()
@pytest.fixture
def key(tmp_path: Path) -> Iterator[Path]:
"""Sign and verify the stand-in's ledger with a throwaway key.
Args:
tmp_path: pytest's temporary directory (also ``LEDGER_HOME``).
Yields:
The key path, patched into ``_ledger_earners``.
"""
path = tmp_path / "hmac.key"
path.write_bytes(KEY)
with patch.object(_ledger_earners, "HMAC_KEY_FILE", path):
yield path
def _credit_on(home: Path, day: date) -> None:
"""Write the stand-in's ledger with one signed credit at noon on ``day``."""
ledger = home / str(PIANO.ledger)
ledger.parent.mkdir(parents=True)
when = datetime.combine(day, time(12)).astimezone()
write_ledger(ledger.parent, [credit(when=when, entry_id="piano:1")])
def _base(day: date) -> float:
"""``resolve_budget``'s floor on ``day`` with ``piano`` registered, all "no"."""
with (
pin_today(day),
patch("earned_time.EARNERS", (*earned_time.EARNERS, PIANO)),
patch(_WORKOUT, return_value=False),
patch(_LEETCODE, return_value=False),
patch(_ANSWER, return_value=False),
):
return resolve_budget(Config()).base_seconds
class TestTheMap:
"""Only penalised, ledger-backed earners, keyed by name."""
def test_maps_each_penalised_reader_to_its_first_credit(
self, tmp_path: Path, key: Path
) -> None:
"""A bonus, a ledger-less earner and a matcher-less one are left out.
Args:
tmp_path: The suite's ``LEDGER_HOME``.
key: The patched signing key.
"""
assert key.exists()
_credit_on(tmp_path, AFTER_CUT - _DAY)
registry = (
PIANO,
replace(PIANO, name="bonus", penalty_from=None),
replace(PIANO, name="no-ledger", ledger=None),
replace(PIANO, name="no-match", match=None),
)
assert first_credits(registry, AFTER_CUT) == {"piano": AFTER_CUT - _DAY}
def test_an_unreadable_ledger_is_no_first_credit(self) -> None:
"""No ledger and no key on record: ``None``, which resolve treats closed."""
assert first_credits((PIANO,), AFTER_CUT) == {"piano": None}
def test_an_old_earned_time_gets_no_map(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Without ``maturity`` (< 0.6) there is nothing to pass on.
Args:
monkeypatch: pytest's attribute patcher.
"""
monkeypatch.delattr(earned_time, "maturity")
assert first_credits((PIANO,), AFTER_CUT) is None
class TestTheBudget:
"""The cut waits for the day after the first real credit."""
def test_a_gate_that_never_paid_out_costs_nothing(self) -> None:
assert _base(AFTER_CUT) == 4 * _HOUR
def test_the_cut_starts_the_day_after_the_first_credit(
self, tmp_path: Path, key: Path
) -> None:
"""Credited on the pinned day: free that day, penalised the next.
Args:
tmp_path: The suite's ``LEDGER_HOME``.
key: The patched signing key.
"""
assert key.exists()
_credit_on(tmp_path, AFTER_CUT)
assert _base(AFTER_CUT) == 4 * _HOUR
assert _base(AFTER_CUT + _DAY) == 3.5 * _HOUR
def test_an_old_earned_time_resolves_as_before(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""No ``maturity``, no ``first_credits``: ``penalty_from`` alone cuts.
Args:
monkeypatch: pytest's attribute patcher.
"""
monkeypatch.delattr(earned_time, "maturity")
assert _base(AFTER_CUT) == 3.5 * _HOUR

View File

@ -10,7 +10,7 @@ passes on to ``earned_time.resolve``, at call time.
from __future__ import annotations
from dataclasses import replace
from datetime import datetime
from datetime import datetime, timedelta
import logging
from typing import TYPE_CHECKING
from unittest.mock import call, patch
@ -170,8 +170,14 @@ class TestAGenericEarnerInTheBudget:
assert "Piano unknown" in resolved.reason
def test_its_penalty_lowers_the_base_by_what_it_pays_back(self) -> None:
"""Penalty, then reward: skipping it costs exactly its term."""
penalised = replace(PIANO, penalty_from=AFTER_CUT)
"""Penalty, then reward: skipping it costs exactly its term.
Confirmed the day before, so its empty ledger fails closed: penalised
from ``penalty_from`` all the same.
"""
penalised = replace(
PIANO, penalty_from=AFTER_CUT, confirmed_on=AFTER_CUT - timedelta(days=1)
)
missed = _resolve(answer=False, earner=penalised)
earned = _resolve(answer=True, earner=penalised)
assert missed.base_seconds == 3.5 * _HOUR