earned_time 0.1.0: one penalty-then-reward registry for gaming and shutdown time
Some checks are pending
dependency freshness / dependencies on newest stable (push) Waiting to run
earned_time pre-commit / pre-commit (push) Waiting to run
earned_time tests / test (push) Waiting to run
file length / file-length (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
palette drift / palette (push) Waiting to run
repo-contract / repo contract (push) Waiting to run
shared gate tests / gate unit tests (push) Waiting to run

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krzysztof Rudnicki 2026-10-04 18:44:30 +02:00
parent 4ccaba1bb3
commit ac1aae4f7a
15 changed files with 1388 additions and 1 deletions

View File

@ -0,0 +1,73 @@
name: earned_time pre-commit
on:
push:
branches: [main]
# The workflow's own file counts as earned_time: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['earned_time/**', '.github/workflows/earned-time-precommit.yml']
pull_request:
branches: [main]
paths: ['earned_time/**', '.github/workflows/earned-time-precommit.yml']
jobs:
pre-commit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.x"
# Only pre-commit itself. Every linting hook runs in its own isolated
# env -- ruff, mypy, pylint, bandit and codespell each come from their
# own repo+rev, and the local hooks need nothing beyond bash and
# python3 -- so installing the full requirements.txt spent 9s a run on
# packages no pre-commit-stage hook imports. Verified by running all
# three configs green in a venv holding pre-commit and nothing else.
# The pin is read out of requirements.txt rather than repeated here, so
# it cannot drift from what the dependency-freshness gate governs.
- name: Install pre-commit
run: |
pip install "$(grep '^pre-commit==' earned_time/requirements.txt)"
# Keyed on THIS package's config. pre-commit/action@v3.0.1 caches on
# hashFiles('.pre-commit-config.yaml') -- the repo ROOT's -- so all three
# package jobs shared one key, raced to overwrite it with hook sets built
# from four different configs, and editing a per-package config did not
# invalidate it. The cache was therefore silently wrong rather than
# missing, and the hooks reinstalled from scratch on every run (38-49s).
- uses: actions/cache@v4
with:
path: ~/.cache/pre-commit
key: >-
pre-commit-earned_time-${{ env.pythonLocation }}-${{
hashFiles('earned_time/.pre-commit-config.yaml') }}
# Prefix fallback, or this fix would evaporate on the very
# event that motivated it: bumping a hook rev changes the
# exact key, and a total miss means reinstalling every hook
# again. pre-commit's store is keyed internally by hook
# repo+rev, so a superseded entry is ignored disk rather
# than a wrong hook, and cache still re-saves under the new
# exact key after a prefix-only hit.
restore-keys: pre-commit-earned_time-${{ env.pythonLocation }}-
# mypy re-analyses the stdlib and every third-party stub from scratch
# without this, which measured 10s of the hook step in CI against 0.25s
# warm locally -- a 19MB cache dir does the whole difference. Keyed on
# the config, not the sources: mypy's own cache metadata records the
# interpreter, the flag set and a hash per source file, so it re-checks
# what changed and cannot pass stale code. Deliberately NOT keyed on
# github.sha -- that re-saves 19MB x3 every commit and would evict the
# 68MB hook caches under the repo's 10GB LRU budget. A fixed key keeps
# the expensive part, the dependency analysis, which is what costs 10s.
- uses: actions/cache@v4
with:
path: .mypy_cache
key: >-
mypy-earned_time-${{ env.pythonLocation }}-${{
hashFiles('earned_time/.pre-commit-config.yaml') }}
restore-keys: mypy-earned_time-${{ env.pythonLocation }}-
- name: Run the hooks
run: >-
pre-commit run --config earned_time/.pre-commit-config.yaml
--all-files --show-diff-on-failure

36
.github/workflows/earned-time-tests.yml vendored Normal file
View File

@ -0,0 +1,36 @@
name: earned_time tests
on:
push:
branches: [main]
# The workflow's own file counts as earned_time: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['earned_time/**', '.github/workflows/earned-time-tests.yml']
pull_request:
branches: [main]
paths: ['earned_time/**', '.github/workflows/earned-time-tests.yml']
jobs:
test:
runs-on: ubuntu-latest
defaults:
run:
working-directory: earned_time
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.x"
cache: pip
cache-dependency-path: earned_time/requirements.txt
- name: Install dependencies
run: pip install -r requirements.txt && pip install --no-deps -e .
# No display needed: earned_time is stdlib-only policy, imported
# headless by the root gaming daemon.
- name: Run tests with coverage
run: python -m pytest -v

View File

@ -1,7 +1,7 @@
# utils
Monorepo of shared gates and libraries (dep_freshness, file_length, md_naming,
repo_contract, coverage_gaps, crdt-sync, gatelock, freedays, music_theory, ...).
repo_contract, coverage_gaps, crdt-sync, gatelock, freedays, earned_time, music_theory, ...).
Python packages with a root-level `<pkg>/tests` run with `PYTHONPATH=.`;
subprojects with their own `pyproject.toml` run via `scripts/run_subproject_tests.sh <dir>`.
`scripts/check_repo_contract.sh` is the REAL shared gate (not a shim): never

View File

@ -0,0 +1,143 @@
# Pre-commit Configuration for earned_time
# Install: pre-commit install && pre-commit install --hook-type pre-push
# Run: pre-commit run --all-files
# Update: pre-commit autoupdate
# NOTE: this subproject lives inside the ~/src/utils monorepo, whose git root is
# one level up. pre-commit always runs hooks from the git root, so every path
# below is written relative to THAT, not to this directory, and the top-level
# `files` key keeps these hooks off the sibling subprojects. CI invokes this
# config exactly the same way:
# pre-commit run --config earned_time/.pre-commit-config.yaml --all-files
files: ^earned_time/
default_language_version:
python: python3
default_stages: [pre-commit]
fail_fast: false
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v4.6.0
hooks:
- id: trailing-whitespace
args: [--markdown-linebreak-ext=md]
- id: end-of-file-fixer
- id: check-yaml
- id: check-json
- id: check-toml
- id: check-added-large-files
args: [--maxkb=2000]
- id: check-merge-conflict
- id: detect-private-key
- id: debug-statements
- id: name-tests-test
args: [--pytest-test-first]
- id: check-ast
- id: mixed-line-ending
args: [--fix=lf]
- id: requirements-txt-fixer
- repo: local
hooks:
- id: file-length-cap
name: file length <= 250 lines
entry: bash scripts/check_file_length.sh
language: system
pass_filenames: true
- id: subproject-scope
name: subproject scope regex still matches
entry: scripts/check_subproject_scope.sh earned_time
language: script
# always_run so the top-level `files` filter cannot skip the one hook
# whose job is to notice that filter has stopped matching anything.
always_run: true
pass_filenames: false
- id: no-noqa
name: Block noqa comments
entry: '(?i)#\s*(noqa|type:\s*ignore)'
language: pygrep
types: [python]
- id: no-ruff-noqa
name: Block ruff noqa file-level comments
entry: '(?i)#\s*ruff:\s*noqa'
language: pygrep
types: [python]
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.16.7
hooks:
- id: ruff
args: [--fix, --unsafe-fixes, --exit-non-zero-on-fix, --show-fixes]
types_or: [python, pyi]
- id: ruff-format
types_or: [python, pyi]
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.13.0
hooks:
- id: mypy
args:
- --ignore-missing-imports
- --no-error-summary
- --disable-error-code=no-untyped-def
- --disable-error-code=no-untyped-call
- --disable-error-code=var-annotated
- --disable-error-code=no-any-unimported
- --disable-error-code=type-arg
- --disable-error-code=no-any-return
- --disable-error-code=misc
- --disable-error-code=unused-ignore
- --disable-error-code=unreachable
- --disable-error-code=assignment
- --disable-error-code=no-redef
- --disable-error-code=attr-defined
- --disable-error-code=arg-type
- --disable-error-code=union-attr
- --disable-error-code=call-overload
- --disable-error-code=return-value
- --disable-error-code=redundant-cast
- --disable-error-code=empty-body
- --disable-error-code=list-item
- repo: https://github.com/pylint-dev/pylint
rev: v4.0.8
hooks:
- id: pylint
args:
- --rcfile=earned_time/pyproject.toml
- --fail-under=10
- --jobs=4
additional_dependencies:
- pytest
- repo: https://github.com/PyCQA/bandit
rev: 1.9.4
hooks:
- id: bandit
args:
- -c
- earned_time/pyproject.toml
- --severity-level=high
- --confidence-level=medium
- --skip=B113
additional_dependencies: ["bandit[toml]"]
exclude: ^(earned_time/earned_time/tests/|.*test.*\.py$)
- repo: local
hooks:
- id: pytest-coverage
name: pytest with coverage enforcement
entry: scripts/run_subproject_tests.sh earned_time
language: script
types: [python]
pass_filenames: false
require_serial: true
stages: [pre-push]
- repo: https://github.com/codespell-project/codespell
rev: v2.3.0
hooks:
- id: codespell
args:
- --skip=*.json,*.lock,.git,__pycache__,.venv

68
earned_time/README.md Normal file
View File

@ -0,0 +1,68 @@
# earned_time
Penalty-then-reward for gate apps: **one** table of what every gate earns in
gaming time and shutdown time, read by both consumers.
| consumer | applies | via |
|---|---|---|
| steam-backlog-enforcer | `Resolution.gaming_minutes` | `_budget_resolve.resolve_budget` |
| screen-locker | `Resolution.shutdown_minutes` | `_shutdown_base.reset_to_base_if_new_day` + live pass |
A gate (leetcode-guard, book-guard, ...) never touches either. It publishes a
fact — HMAC-signed `credit` rows in its ledger — and is registered here once.
## Adding a gate
1. The gate writes `{"entries": [...]}` to its ledger. Each row has
`kind: "credit"`, a `detail` dict, and an `hmac` signed with
`/etc/workout-locker/hmac.key` (`earned_time.entry_signature`, the same as
gatelock's `log_integrity`).
2. Add an `Earner` to `earned_time/_policy.py` and to `EARNERS`:
```python
ANKI = Earner(
name="anki", # also screen-locker's "anki_bonus_date" stamp
label="Anki",
gaming_minutes=30,
shutdown_minutes=30,
penalty_from=date(2026, 10, 10), # base drops by the same 30/30 from then on
ledger=".local/share/anki_guard/ledger.json",
match=_anki_match, # which verified credit rows count today
missing_ledger_is_no=True,
)
```
3. Bump the version, tag `earned-time-vX.Y.Z`, and bump the pin in both
consumers. Neither needs code for a ledger-backed flat earner.
`penalty_from` is the gate's start date, so the base is never cut before the
reward that pays it back exists. The best case stays the same, because the
base drops by exactly what the new earner adds.
## Rules the consumers rely on
- **Minutes everywhere.** Shutdown is minutes after local midnight.
screen-locker still stores whole hours and refuses (raises) a remainder
rather than flooring it.
- **Paths are passed in.** `done_today(earner, ledger, key_file)` never
resolves a real path itself; consumers' test suites redirect their own
constants.
- **`None` is "could not check", never "no".** It earns nothing (fail closed),
but consumers log it. A missing answer in `resolve()` is logged as well; an
unknown earner name raises.
- **Stdlib only.** The gaming daemon imports this as root, headless.
## Install
```console
pip install --user "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.1.0#subdirectory=earned_time"
```
The root `steam-backlog-enforcer.service` sets `HOME=/home/kuhy`, so it imports
from the same user site-packages.
## Test
```console
scripts/run_subproject_tests.sh earned_time
```

View File

@ -0,0 +1,66 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Penalty-then-reward for gate apps: one registry, one sum, two consumers.
A gate (leetcode-guard, book-guard, ...) never touches gaming time or the
shutdown schedule. It publishes a fact -- HMAC-signed ``credit`` rows in its
ledger -- and registers one :class:`Earner` here. The two consumers read the
registry and apply the result:
* steam-backlog-enforcer enforces ``Resolution.gaming_minutes``;
* screen-locker writes ``Resolution.shutdown_minutes`` to the schedule.
A consumer's whole integration::
import earned_time
answers = {
e.name: earned_time.done_today(e, home / e.ledger, KEY_FILE)
for e in earned_time.EARNERS
if e.ledger is not None
}
answers["workout"] = my_workout_count()
day = earned_time.resolve(answers)
apply(day.gaming_minutes)
Adding a gate is one :class:`Earner` in :mod:`earned_time._policy`; with
``penalty_from`` set, the base drops by what it pays back from that day on.
"""
from __future__ import annotations
from earned_time._ledger import done_today, entry_signature, today_window, verified
from earned_time._policy import (
EARNERS,
GAMING_BASE_MINUTES,
GAMING_CEILING_MINUTES,
LEETCODE,
READING,
SHUTDOWN_BASE_MINUTES,
SHUTDOWN_CEILING_MINUTES,
WORKOUT,
Earner,
earner,
)
from earned_time._resolve import Base, Resolution, Term, base_for, resolve
__all__ = [
"EARNERS",
"GAMING_BASE_MINUTES",
"GAMING_CEILING_MINUTES",
"LEETCODE",
"READING",
"SHUTDOWN_BASE_MINUTES",
"SHUTDOWN_CEILING_MINUTES",
"WORKOUT",
"Base",
"Earner",
"Resolution",
"Term",
"base_for",
"done_today",
"earner",
"entry_signature",
"resolve",
"today_window",
"verified",
]

View File

@ -0,0 +1,157 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The shared reader: did a gate's signed ledger record today's credit?
Every ledger-backed earner publishes the same way -- HMAC-signed ``credit``
rows in ``{"entries": [...]}`` -- so one reader serves all of them, with the
earner's ``match`` deciding which rows count. A new gate that writes that
shape needs no reader of its own in any consumer.
**Paths are always passed in.** The ledger and the key are resolved by the
consumer, never here: both consumers' test suites redirect their own module
constants, and a default living in this package would let a test read the
real ledger and the real key.
**"Cannot check" is not "no".** Every unreadable state returns ``None``. The
consumer fails closed to no bonus, but logs it -- failing open would let the
coupling be defeated by deleting a file.
The HMAC canonicalisation is gatelock's (``log_integrity.compute_entry_hmac``),
reproduced in stdlib instead of imported: gatelock pulls in tkinter, and the
root gaming daemon must be able to import this package headless.
"""
from __future__ import annotations
from datetime import UTC, datetime, time
import hashlib
import hmac
import json
import logging
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from pathlib import Path
from earned_time._policy import Earner
_logger: Final = logging.getLogger(__name__)
_CREDIT: Final = "credit"
def entry_signature(entry: dict[str, object], key: bytes) -> str:
"""The HMAC-SHA256 every locker signs a ledger row with.
Args:
entry: The row; any existing ``hmac`` field is left out.
key: The shared signing key.
Returns:
The hex digest over the canonical JSON of the rest of the row.
"""
body = {k: v for k, v in entry.items() if k != "hmac"}
payload = json.dumps(body, sort_keys=True, separators=(",", ":"))
return hmac.new(key, payload.encode(), hashlib.sha256).hexdigest()
def verified(entry: dict[str, object], key: bytes) -> bool:
"""Whether a row's ``hmac`` is genuine. A forged credit earns time."""
stored = entry.get("hmac")
if not isinstance(stored, str):
return False
return hmac.compare_digest(stored, entry_signature(entry, key))
def today_window(now: datetime | None = None) -> tuple[float, float]:
"""Local midnight and now, as unix seconds: where today's credit must fall."""
moment = (now or datetime.now(tz=UTC)).astimezone()
midnight = datetime.combine(moment.date(), time.min, moment.tzinfo)
return midnight.timestamp(), moment.timestamp()
def read_key(key_file: Path) -> bytes | None:
"""The signing key, or ``None`` (logged) when it is unreadable or empty."""
try:
key = key_file.read_bytes().strip()
except OSError as exc:
_logger.warning("Cannot read the integrity key at %s (%s)", key_file, exc)
return None
if not key:
_logger.warning("Integrity key at %s is empty", key_file)
return None
return key
def read_rows(ledger: Path, earner: Earner) -> list[object] | None:
"""The ledger's ``entries`` array, or ``None`` when it cannot be read.
A ledger that does not exist is ``[]`` -- silently -- for an earner whose
gate may simply never have run (``missing_ledger_is_no``); for every other
earner it is a fault like any unreadable file.
"""
try:
raw = json.loads(ledger.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
if earner.missing_ledger_is_no:
return []
_logger.warning(
"Cannot read the %s ledger at %s (%s)", earner.label, ledger, exc
)
return None
except OSError as exc:
_logger.warning(
"Cannot read the %s ledger at %s (%s)", earner.label, ledger, exc
)
return None
except ValueError as exc:
_logger.warning(
"%s ledger at %s is not valid JSON (%s)", earner.label, ledger, exc
)
return None
rows = raw.get("entries") if isinstance(raw, dict) else None
if not isinstance(rows, list):
_logger.warning("%s ledger at %s has no entries array", earner.label, ledger)
return None
return rows
def done_today(
earner: Earner,
ledger: Path,
key_file: Path,
*,
now: datetime | None = None,
) -> bool | None:
"""Whether ``earner``'s ledger holds a verified credit that counts today.
Args:
earner: A ledger-backed earner (it must have ``match``).
ledger: The gate's ledger file, resolved by the caller.
key_file: The shared HMAC key, resolved by the caller.
now: Stand-in for the current time, for tests.
Returns:
True or False when the ledger could be read and verified, ``None``
when it could not -- which is never "not done".
Raises:
ValueError: ``earner`` has no ``match``; its answer is the consumer's.
"""
if earner.match is None:
msg = f"earner {earner.name!r} has no shared reader; supply its answer"
raise ValueError(msg)
key = read_key(key_file)
if key is None:
return None
rows = read_rows(ledger, earner)
if rows is None:
return None
window = today_window(now)
match = earner.match
return any(
isinstance(row, dict)
and row.get("kind") == _CREDIT
and verified(row, key)
and match(row, window)
for row in rows
)

View File

@ -0,0 +1,179 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The one table of what every gate earns: the earner registry and the bases.
Every number that used to be duplicated between screen-locker (shutdown) and
steam-backlog-enforcer (gaming) lives here, once. A consumer never hard-codes
an hour; it asks :func:`earned_time.resolve` and applies the answer.
**Penalty, then reward.** A gate with ``penalty_from`` lowers the base by
exactly what it pays back, from that day on. Doing the thing restores the old
day; skipping it costs the full amount. The date is the gate's own start, so
the cut never lands before the reward that pays for it -- taking an hour a
reader could not yet earn back was a same-day loss on 2026-09-26.
**Ceilings stay put.** Adding a penalised earner leaves the best case
unchanged: the base drops by what the new earner adds.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, date, datetime
import logging
from typing import TYPE_CHECKING, Final, Literal
if TYPE_CHECKING:
from collections.abc import Callable
Window = tuple[float, float]
# Whether one HMAC-verified ``credit`` row counts for the window.
CreditMatch = Callable[[dict[str, object], Window], bool]
_logger: Final = logging.getLogger(__name__)
# The day before any earner's penalty: 5h of gaming, shutdown at 20:00.
# Minutes, both of them -- shutdown is minutes after local midnight.
GAMING_BASE_MINUTES: Final = 5 * 60
SHUTDOWN_BASE_MINUTES: Final = 20 * 60
# Hard ceilings, whatever the terms add up to.
GAMING_CEILING_MINUTES: Final = 8 * 60
SHUTDOWN_CEILING_MINUTES: Final = 23 * 60
@dataclass(frozen=True)
class Earner:
"""One gate's reward, and (optionally) the base cut that comes with it.
Attributes:
name: Registry key. screen-locker stamps its live pass as
``f"{name}_bonus_date"``, so renaming an earner re-applies its
bonus once.
label: Human name, for logs and reason strings.
gaming_minutes: Gaming time the first unit earns.
shutdown_minutes: How much later shutdown moves for the first unit.
kind: ``"flat"`` earns once per day from a yes/no answer;
``"counted"`` earns per unit (the workout), and its live credit is
applied by the consumer that counts the units.
extra_shutdown_minutes: Each further unit, for ``"counted"`` earners.
penalty_from: From this day the base drops by ``gaming_minutes`` and
``shutdown_minutes``. ``None`` means a pure bonus.
ledger: The gate's HMAC-signed ledger, relative to the home directory.
``None`` means no shared reader: the consumer supplies the answer.
match: Which verified ``credit`` rows count for today.
missing_ledger_is_no: A ledger that does not exist yet is an honest
"no" (the gate never ran), not a fault.
"""
name: str
label: str
gaming_minutes: int
shutdown_minutes: int
kind: Literal["flat", "counted"] = "flat"
extra_shutdown_minutes: int = 0
penalty_from: date | None = None
ledger: str | None = None
match: CreditMatch | None = None
missing_ledger_is_no: bool = False
def penalised_on(self, day: date) -> bool:
"""Whether this earner's base cut is in force on ``day``."""
return self.penalty_from is not None and day >= self.penalty_from
def gaming_for(self, units: int) -> int:
"""Gaming minutes ``units`` earn: the first unit only, never more."""
return self.gaming_minutes if units > 0 else 0
def shutdown_for(self, units: int) -> int:
"""Shutdown minutes ``units`` earn: first unit, then each extra one."""
if units <= 0:
return 0
return self.shutdown_minutes + (units - 1) * self.extra_shutdown_minutes
def _within(raw: object, window: Window) -> bool | None:
"""Whether a unix-seconds stamp falls in ``window``; ``None`` if unusable."""
try:
stamp = float(str(raw))
except ValueError:
return None
start, end = window
return start <= stamp <= end
def _leetcode_match(row: dict[str, object], window: Window) -> bool:
"""A solve counts on the day LeetCode accepted it (``submitted_at``).
``day`` is the *harvesting* run's date and runs late, so it is only the
fallback for a row without a usable stamp: it can miss a late solve but
never invent one.
"""
detail = row.get("detail")
raw = detail.get("submitted_at") if isinstance(detail, dict) else None
if raw is not None:
landed = _within(raw, window)
if landed is not None:
return landed
_logger.warning(
"LeetCode credit %r has an unparsable submitted_at (%r); "
"falling back to its day key",
row.get("entry_id"),
raw,
)
local = datetime.fromtimestamp(window[0], tz=UTC).astimezone()
return row.get("day") == local.date().isoformat()
def _reading_match(row: dict[str, object], window: Window) -> bool:
"""A reading credit counts if it earned the bonus and ended today."""
detail = row.get("detail")
if not isinstance(detail, dict) or detail.get("bonus") != "1":
return False
landed = _within(detail.get("ended_at"), window)
if landed is None:
_logger.warning("reading credit %r has no usable ended_at", row.get("entry_id"))
return False
return landed
WORKOUT: Final = Earner(
name="workout",
label="workout",
gaming_minutes=2 * 60,
shutdown_minutes=2 * 60,
kind="counted",
extra_shutdown_minutes=60,
)
LEETCODE: Final = Earner(
name="leetcode",
label="LeetCode",
gaming_minutes=60,
shutdown_minutes=60,
ledger=".local/share/leetcode_guard/ledger.json",
match=_leetcode_match,
)
READING: Final = Earner(
name="reading",
label="reading",
gaming_minutes=60,
shutdown_minutes=60,
penalty_from=date(2026, 10, 1),
ledger=".local/share/book_guard/ledger.json",
match=_reading_match,
missing_ledger_is_no=True,
)
# Order is the order of reason strings and of screen-locker's live pass.
EARNERS: Final[tuple[Earner, ...]] = (WORKOUT, LEETCODE, READING)
def earner(name: str, earners: tuple[Earner, ...] = EARNERS) -> Earner:
"""Look an earner up by name.
Raises:
KeyError: No earner is registered under ``name``.
"""
for candidate in earners:
if candidate.name == name:
return candidate
raise KeyError(name)

View File

@ -0,0 +1,173 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Today's gaming budget and shutdown time: the base plus every earned term.
Both consumers call :func:`resolve` with their own answers and apply the
result: steam-backlog-enforcer the gaming minutes, screen-locker the shutdown
minutes. The sum is computed in exactly one place, so the two can no longer
disagree about what a day earned.
An answer is a unit count (``True``/``False`` work as 1/0) or ``None`` for
"could not check". ``None`` -- and an earner the consumer did not answer for at
all -- earns nothing: fail closed. The difference shows up only in
:attr:`Term.answer`, so the consumer can log or report it.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, date, datetime
import logging
from typing import TYPE_CHECKING, Final
from earned_time._policy import (
EARNERS,
GAMING_BASE_MINUTES,
GAMING_CEILING_MINUTES,
SHUTDOWN_BASE_MINUTES,
SHUTDOWN_CEILING_MINUTES,
Earner,
)
if TYPE_CHECKING:
from collections.abc import Mapping
_logger: Final = logging.getLogger(__name__)
Answer = int | bool | None
@dataclass(frozen=True)
class Base:
"""The floor for one day, after every penalty in force on it.
Attributes:
gaming_minutes: Gaming time a day that earned nothing gets.
shutdown_minutes: Shutdown time of such a day, minutes after midnight.
"""
gaming_minutes: int
shutdown_minutes: int
@dataclass(frozen=True)
class Term:
"""What one earner contributed.
Attributes:
earner: The earner.
answer: The consumer's answer: a unit count, or ``None`` for
"could not check" (and for no answer at all).
gaming_minutes: Gaming minutes earned.
shutdown_minutes: Shutdown minutes earned.
"""
earner: Earner
answer: int | None
gaming_minutes: int
shutdown_minutes: int
@dataclass(frozen=True)
class Resolution:
"""One day's budget and shutdown, and what earned them.
Attributes:
day: The day resolved.
base: Its floor.
terms: One entry per registered earner, in registry order.
gaming_minutes: Base plus terms, capped at the gaming ceiling.
shutdown_minutes: Base plus terms, capped at the shutdown ceiling.
"""
day: date
base: Base
terms: tuple[Term, ...]
gaming_minutes: int
shutdown_minutes: int
def term(self, name: str) -> Term:
"""The term of the earner called ``name``.
Raises:
KeyError: No such earner was resolved.
"""
for candidate in self.terms:
if candidate.earner.name == name:
return candidate
raise KeyError(name)
def _today() -> date:
return datetime.now(tz=UTC).astimezone().date()
def base_for(day: date | None = None, earners: tuple[Earner, ...] = EARNERS) -> Base:
"""The floor for ``day`` (default today): every penalty in force, taken off."""
target = day or _today()
penalised = [e for e in earners if e.penalised_on(target)]
return Base(
gaming_minutes=GAMING_BASE_MINUTES - sum(e.gaming_minutes for e in penalised),
shutdown_minutes=SHUTDOWN_BASE_MINUTES
- sum(e.shutdown_minutes for e in penalised),
)
def _units(answer: Answer) -> int | None:
if answer is None:
return None
return max(0, int(answer))
def resolve(
answers: Mapping[str, Answer],
day: date | None = None,
earners: tuple[Earner, ...] = EARNERS,
) -> Resolution:
"""Sum the base and every earned term, each capped at its ceiling.
Args:
answers: Per earner name, a unit count or ``None`` ("could not check").
A registered earner missing from it earns nothing and is logged:
the consumer forgot to ask, which must not pass as a "no".
day: The day to resolve (default today); only the base depends on it.
earners: The registry; tests pass their own.
Returns:
The resolution, with a term for every registered earner.
Raises:
KeyError: ``answers`` names an earner that is not registered -- a
typo that would otherwise silently earn nothing.
"""
known = {e.name for e in earners}
unknown = sorted(set(answers) - known)
if unknown:
msg = f"not registered earners: {', '.join(unknown)}"
raise KeyError(msg)
target = day or _today()
base = base_for(target, earners)
terms: list[Term] = []
for item in earners:
if item.name not in answers:
_logger.warning(
"No answer for the %s earner; it earns nothing today", item.label
)
units = _units(answers.get(item.name))
count = units or 0
terms.append(
Term(
earner=item,
answer=units,
gaming_minutes=item.gaming_for(count),
shutdown_minutes=item.shutdown_for(count),
)
)
gaming = base.gaming_minutes + sum(t.gaming_minutes for t in terms)
shutdown = base.shutdown_minutes + sum(t.shutdown_minutes for t in terms)
return Resolution(
day=target,
base=base,
terms=tuple(terms),
gaming_minutes=min(GAMING_CEILING_MINUTES, gaming),
shutdown_minutes=min(SHUTDOWN_CEILING_MINUTES, shutdown),
)

View File

@ -0,0 +1,2 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""earned_time tests."""

View File

@ -0,0 +1,118 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The shared ledger reader, and every way it refuses to answer."""
from __future__ import annotations
from datetime import UTC, datetime
import json
import logging
from typing import TYPE_CHECKING
import pytest
from earned_time import LEETCODE, READING, WORKOUT, done_today, entry_signature
from earned_time._ledger import read_key, read_rows, today_window, verified
if TYPE_CHECKING:
from pathlib import Path
KEY = b"test-key"
NOON = datetime(2026, 10, 4, 12, tzinfo=UTC)
INSIDE = today_window(NOON)[1] - 60
def _signed(row: dict[str, object]) -> dict[str, object]:
return {**row, "hmac": entry_signature(row, KEY)}
@pytest.fixture
def key_file(tmp_path: Path) -> Path:
path = tmp_path / "hmac.key"
path.write_bytes(KEY + b"\n")
return path
def _ledger(tmp_path: Path, rows: list[object]) -> Path:
path = tmp_path / "ledger.json"
path.write_text(json.dumps({"entries": rows}), encoding="utf-8")
return path
def test_signature_is_gatelocks_canonical_form() -> None:
# Pinned vector: sorted keys, compact separators, hmac field excluded --
# gatelock's log_integrity.compute_entry_hmac gives the same digest.
row = {"b": 1, "a": "x", "hmac": "ignored"}
assert entry_signature(row, b"k") == (
"5fd4480841273ec8336e76e19bbeb1dd7bd9ded2c442c84cdb2786f7e030987b"
)
def test_verified() -> None:
row = _signed({"kind": "credit"})
assert verified(row, KEY)
assert not verified({**row, "kind": "charge"}, KEY)
assert not verified({"kind": "credit"}, KEY)
def test_today_window_defaults_to_now() -> None:
start, end = today_window()
assert start <= end
def test_read_key(tmp_path: Path, key_file: Path) -> None:
assert read_key(key_file) == KEY
assert read_key(tmp_path / "missing") is None
empty = tmp_path / "empty"
empty.write_bytes(b" \n")
assert read_key(empty) is None
def test_missing_ledger_is_no_only_where_declared(tmp_path: Path) -> None:
missing = tmp_path / "missing.json"
assert read_rows(missing, READING) == []
assert read_rows(missing, LEETCODE) is None
@pytest.mark.parametrize(
("content", "message"),
[("{not json", "not valid JSON"), ('{"entries": 3}', "no entries array")],
)
def test_unusable_ledger(
tmp_path: Path, caplog: pytest.LogCaptureFixture, content: str, message: str
) -> None:
path = tmp_path / "ledger.json"
path.write_text(content, encoding="utf-8")
with caplog.at_level(logging.WARNING):
assert read_rows(path, READING) is None
assert message in caplog.text
def test_unreadable_ledger(tmp_path: Path) -> None:
assert read_rows(tmp_path, READING) is None # a directory
def test_done_today(tmp_path: Path, key_file: Path) -> None:
good = _signed({"kind": "credit", "detail": {"submitted_at": INSIDE}})
path = _ledger(tmp_path, [good])
assert done_today(LEETCODE, path, key_file, now=NOON) is True
def test_not_done_today(tmp_path: Path, key_file: Path) -> None:
rows: list[object] = [
"junk",
_signed({"kind": "charge", "detail": {"submitted_at": INSIDE}}),
{"kind": "credit", "detail": {"submitted_at": INSIDE}, "hmac": "forged"},
]
path = _ledger(tmp_path, rows)
assert done_today(LEETCODE, path, key_file, now=NOON) is False
def test_cannot_check(tmp_path: Path, key_file: Path) -> None:
path = _ledger(tmp_path, [])
assert done_today(LEETCODE, path, tmp_path / "nokey") is None
assert done_today(LEETCODE, tmp_path / "missing.json", key_file) is None
def test_earner_without_reader_is_refused(tmp_path: Path, key_file: Path) -> None:
with pytest.raises(ValueError, match="no shared reader"):
done_today(WORKOUT, tmp_path / "x.json", key_file)

View File

@ -0,0 +1,99 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The registry: earner arithmetic and each gate's credit rule."""
from __future__ import annotations
from datetime import UTC, date, datetime
import logging
from typing import TYPE_CHECKING
import pytest
from earned_time import EARNERS, LEETCODE, READING, WORKOUT, earner
from earned_time._ledger import today_window
if TYPE_CHECKING:
from collections.abc import Mapping
NOON = datetime(2026, 10, 4, 12, tzinfo=UTC)
WINDOW = today_window(NOON)
INSIDE = WINDOW[1] - 60
BEFORE = WINDOW[0] - 60
TODAY = datetime.fromtimestamp(WINDOW[0]).astimezone().date().isoformat()
def test_registry_order_and_names() -> None:
assert [e.name for e in EARNERS] == ["workout", "leetcode", "reading"]
def test_lookup_by_name() -> None:
assert earner("reading") is READING
def test_lookup_unknown_raises() -> None:
with pytest.raises(KeyError):
earner("nope")
def test_penalty_starts_on_its_day() -> None:
assert not READING.penalised_on(date(2026, 9, 30))
assert READING.penalised_on(date(2026, 10, 1))
assert not LEETCODE.penalised_on(date(2030, 1, 1))
def test_flat_earner_pays_once() -> None:
assert [LEETCODE.gaming_for(n) for n in range(3)] == [0, 60, 60]
assert [LEETCODE.shutdown_for(n) for n in range(3)] == [0, 60, 60]
def test_counted_earner_pays_each_extra_unit_in_shutdown_only() -> None:
assert [WORKOUT.shutdown_for(n) for n in range(4)] == [0, 120, 180, 240]
assert [WORKOUT.gaming_for(n) for n in range(3)] == [0, 120, 120]
def _match(e: object, row: Mapping[str, object]) -> bool:
fn = getattr(e, "match", None)
assert fn is not None
return bool(fn(dict(row), WINDOW))
@pytest.mark.parametrize(
("detail", "day", "expected"),
[
({"submitted_at": str(INSIDE)}, "1999-01-01", True),
({"submitted_at": BEFORE}, TODAY, False),
({}, TODAY, True),
(None, TODAY, True),
(None, "1999-01-01", False),
],
)
def test_leetcode_match(detail: object, day: str, *, expected: bool) -> None:
assert _match(LEETCODE, {"detail": detail, "day": day}) is expected
def test_leetcode_unparsable_stamp_falls_back_to_day(
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level(logging.WARNING):
row = {"detail": {"submitted_at": "soon"}, "day": TODAY, "entry_id": "x"}
assert _match(LEETCODE, row)
assert "unparsable submitted_at" in caplog.text
@pytest.mark.parametrize(
("detail", "expected"),
[
({"bonus": "1", "ended_at": INSIDE}, True),
({"bonus": "1", "ended_at": BEFORE}, False),
({"bonus": "0", "ended_at": INSIDE}, False),
("not a dict", False),
],
)
def test_reading_match(detail: object, *, expected: bool) -> None:
assert _match(READING, {"detail": detail}) is expected
def test_reading_unusable_end_is_logged(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level(logging.WARNING):
assert not _match(READING, {"detail": {"bonus": "1"}, "entry_id": "r"})
assert "no usable ended_at" in caplog.text

View File

@ -0,0 +1,96 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The one sum: base, terms, ceilings, and failing closed."""
from __future__ import annotations
from datetime import date
import logging
import pytest
from earned_time import Earner, base_for, resolve
BEFORE_CUT = date(2026, 9, 30)
AFTER_CUT = date(2026, 10, 4)
def test_base_before_and_after_the_reading_cut() -> None:
assert base_for(BEFORE_CUT).gaming_minutes == 300
assert base_for(BEFORE_CUT).shutdown_minutes == 20 * 60
assert base_for(AFTER_CUT).gaming_minutes == 240
assert base_for(AFTER_CUT).shutdown_minutes == 19 * 60
def test_base_defaults_to_today() -> None:
assert base_for().gaming_minutes in {240, 300}
def test_nothing_earned_is_the_base() -> None:
day = resolve({"workout": 0, "leetcode": False, "reading": False}, AFTER_CUT)
assert (day.gaming_minutes, day.shutdown_minutes) == (240, 19 * 60)
def test_terms_add_up() -> None:
day = resolve({"workout": 1, "leetcode": True, "reading": False}, AFTER_CUT)
assert day.gaming_minutes == 240 + 120 + 60
assert day.shutdown_minutes == 19 * 60 + 120 + 60
assert day.term("reading").answer == 0
def test_ceilings_hold() -> None:
day = resolve({"workout": 5, "leetcode": True, "reading": True}, AFTER_CUT)
assert (day.gaming_minutes, day.shutdown_minutes) == (8 * 60, 23 * 60)
def test_unknown_answer_earns_nothing_but_is_kept() -> None:
day = resolve({"workout": None, "leetcode": True, "reading": True}, AFTER_CUT)
assert day.term("workout").answer is None
assert day.term("workout").gaming_minutes == 0
def test_negative_count_is_zero() -> None:
day = resolve({"workout": -3, "leetcode": False, "reading": False}, AFTER_CUT)
assert day.term("workout").answer == 0
def test_missing_answer_is_logged_and_earns_nothing(
caplog: pytest.LogCaptureFixture,
) -> None:
with caplog.at_level(logging.WARNING):
day = resolve({"leetcode": True}, AFTER_CUT)
assert "No answer for the workout earner" in caplog.text
assert day.term("workout").answer is None
def test_unregistered_name_raises() -> None:
with pytest.raises(KeyError, match="anky"):
resolve({"anky": True})
def test_term_lookup_unknown_raises() -> None:
with pytest.raises(KeyError):
resolve({}, AFTER_CUT).term("nope")
def test_new_penalised_earner_keeps_the_best_case() -> None:
anki = Earner(
name="anki",
label="Anki",
gaming_minutes=30,
shutdown_minutes=30,
penalty_from=date(2026, 10, 10),
)
from earned_time import EARNERS
earners = (*EARNERS, anki)
later = date(2026, 10, 11)
assert base_for(later, earners).gaming_minutes == 210
assert base_for(later, earners).shutdown_minutes == 18 * 60 + 30
skipped = resolve(
{"workout": 0, "leetcode": 0, "reading": 0, "anki": 0}, later, earners
)
done = resolve(
{"workout": 0, "leetcode": 0, "reading": 0, "anki": 1}, later, earners
)
assert done.gaming_minutes - skipped.gaming_minutes == 30
assert done.shutdown_minutes == 19 * 60

167
earned_time/pyproject.toml Normal file
View File

@ -0,0 +1,167 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "earned-time"
version = "0.1.0"
description = "Penalty-then-reward registry: what every gate app earns in gaming time and shutdown time"
requires-python = ">=3.14"
dependencies = []
[tool.setuptools.packages.find]
include = ["earned_time*"]
[tool.ruff]
target-version = "py314"
include = ["*.py", "**/*.py"]
exclude = [".git", ".venv", "__pycache__", "build", "dist", ".eggs"]
[tool.ruff.lint]
select = ["ALL"]
ignore = [
"D203",
"D213",
"COM812",
"ISC001",
"S603",
]
fixable = ["ALL"]
unfixable = []
[tool.ruff.lint.per-file-ignores]
"**/tests/**/*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
"**/test_*.py" = ["ARG", "D", "PLC0415", "PLR2004", "S101", "SLF001"]
[tool.ruff.lint.pydocstyle]
convention = "google"
[tool.ruff.lint.isort]
force-single-line = false
force-sort-within-sections = true
known-first-party = ["earned_time"]
[tool.ruff.lint.flake8-quotes]
docstring-quotes = "double"
inline-quotes = "double"
[tool.ruff.lint.flake8-tidy-imports]
ban-relative-imports = "all"
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
skip-magic-trailing-comma = false
line-ending = "auto"
docstring-code-format = true
[tool.mypy]
python_version = "3.14"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
disallow_any_unimported = true
disallow_any_explicit = false
disallow_any_generics = true
disallow_subclassing_any = true
strict_equality = true
extra_checks = true
ignore_missing_imports = true
show_error_codes = true
color_output = true
exclude = [".venv/"]
[tool.pylint.main]
# The repo root on the path, so pylint resolves the package's own imports the
# same way pytest does instead of reporting import-error for them.
init-hook = "import sys; sys.path.insert(0, 'earned_time')"
analyse-fallback-blocks = true
persistent = true
jobs = 0
py-version = "3.14"
ignore = [".venv", "__pycache__"]
# Test suites intentionally use patterns (protected-access, fixture-arg
# shadowing) that don't apply to source code; ruff's per-file-ignores carry
# the equivalent exemptions, so tests stay linted -- by ruff. This is the same
# source-only pylint scope wake-alarm / diet-guard / testsAndMisc use. It has
# to be `ignore-paths`, not `ignore = ["tests"]`: `ignore` only matches the
# basename of what pylint is *given*, and pre-commit hands it explicit file
# paths, so a directory name in `ignore` never fires under the hook.
ignore-paths = ["^(.*/)?tests/.*$"]
ignore-patterns = [".*\\.pyi$"]
[tool.pylint.messages_control]
enable = "all"
disable = []
[tool.pylint.design]
min-public-methods = 0
max-module-lines = 1000
max-attributes = 10
[tool.pylint.typecheck]
generated-members = [
".*\\.assert_called_once_with",
".*\\.assert_called_once",
".*\\.assert_called",
".*\\.assert_not_called",
".*\\.assert_any_call",
".*\\.call_args",
".*\\.call_args_list",
".*\\.call_count",
]
[tool.bandit]
exclude_dirs = ["tests", ".venv"]
[tool.pytest.ini_options]
testpaths = ["earned_time/tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"-v",
"--strict-markers",
"--strict-config",
"-ra",
"--cov=earned_time",
"--cov-branch",
"--cov-report=term-missing",
"--cov-report=lcov",
]
filterwarnings = [
"error",
"ignore::DeprecationWarning",
"default::pytest.PytestUnraisableExceptionWarning",
]
[tool.coverage.run]
source = ["earned_time"]
branch = true
omit = [
"*/__pycache__/*",
"*/tests/*",
"*/.venv/*",
]
[tool.coverage.report]
fail_under = 100
show_missing = true
skip_covered = false
exclude_lines = [
"pragma: no cover",
"raise NotImplementedError",
"raise AssertionError",
"if TYPE_CHECKING:",
'if __name__ == "__main__":',
]
partial_branches = ["pragma: no branch"]

View File

@ -0,0 +1,10 @@
# Exact pins, enforced by the shared dependency-freshness gate: a `>=`
# floor is satisfied by anything newer, so it tells a checker nothing
# about whether this repo is actually current.
bandit[toml]==1.9.4
mypy==2.4.0
pre-commit==4.6.2
pylint==4.1.2
pytest==9.1.1
pytest-cov==7.1.0
ruff==0.16.10