earned_time 0.6.0: gate maturity classifier (python -m earned_time maturity), confirmed_on, first_credits-aware penalty start
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

How far to trust a gate is computed from its signed ledger: new / maturing /
mature (>= 14 verified credit days, first >= 21 days old), with a read-only
CLI table. Only signed, non-manual, paying credits count; a gate's own
re-evaluation grant (detail.grant_of, book-guard bonus: rows) counts at
amount 0.

Earner.confirmed_on (hand-set) is new. penalty_start = max(penalty_from, day
after first real credit, day after confirmed_on); resolve/base_for take an
optional first_credits map and can only delay a penalty, never add one.
Omitted, every day resolves exactly as in 0.5.0.

Fail closed: a gate with confirmed_on and no credit on record (unreadable
key/ledger, deleted or emptied ledger) falls back to max(penalty_from, day
after confirmed_on), so deleting a ledger cannot lift a penalty.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WdfmsGMht9ypyvxGPmNF1V
This commit is contained in:
Krzysztof Rudnicki 2026-10-09 21:21:18 +02:00
parent 6bfee3312f
commit a745d6bdf4
15 changed files with 1268 additions and 14 deletions

View File

@ -75,6 +75,35 @@ paid back (nothing done: 4 h / 19:50 on a ladder day; everything done: still
sentinel (2099-01-01) until kuhy confirms a real tutor session credits a
block; it is then set to the day after, never the deploy day.
## Gate maturity (0.6.0)
How far to trust a gate is computed from its ledger, never listed by name:
`maturity(earner, ledger, key_file, today) -> Maturity` (`_maturity.py`).
- **Real credit:** HMAC-verified `credit` row, accepted by the earner's
`match`, not manual (`detail.source` `manual`/`manual_grant`, `entry_id`
`manual:`/`manual_grant:`), and `amount > 0` when the row has one -- a
gate's own re-evaluation grant (`detail.grant_of`, book-guard `bonus:`
rows) may pay 0. A matcher cannot loosen this (`anki_match` alone accepts a
`manual_grant` row).
- **Levels:** `new` = no real credit, or no `Earner.confirmed_on` (set by
kuhy); `mature` = at least `MATURE_MIN_CREDIT_DAYS` (14) distinct credit
days and the first at least `MATURE_MIN_AGE_DAYS` (21) old; else `maturing`.
An unreadable ledger or key is `checked=False`, `new`, reason recorded.
- **Penalty start:** `penalty_start(earner, first_credit)` =
max(`penalty_from`, day after the first real credit, day after
`confirmed_on`). No first credit (never paid out, or could not check) is
not penalised -- unless `confirmed_on` is set: then fail closed,
max(`penalty_from`, day after `confirmed_on`), so deleting or locking a
ledger cannot lift a penalty. Pass `first_credits={name: Maturity.first_credit}`, computed
over `earners_for(day)` (two earners share the name `automation`), to `resolve`
/ `base_for` to apply it; omitted, every day resolves exactly as in 0.5.0.
It can only delay a penalty, never add one.
- **CLI:** `python -m earned_time maturity [--json] [--day YYYY-MM-DD]`
prints one row per ledger-backed earner of every registry. It is the only
code that resolves real paths (`~/<Earner.ledger>`,
`/etc/workout-locker/hmac.key`), and it only reads.
## API
- `resolve(answers, day=None, earners=EARNERS) -> Resolution` — the one sum.
@ -162,7 +191,7 @@ on the day itself is logged and does not count.
## Install
```console
pip install --user --break-system-packages --no-deps "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.5.0#subdirectory=earned_time"
pip install --user --break-system-packages --no-deps "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.6.0#subdirectory=earned_time"
```
The root `steam-backlog-enforcer.service` sets `HOME=/home/kuhy`, so it imports

View File

@ -22,6 +22,9 @@ A consumer's whole integration::
day = earned_time.resolve(answers)
apply(day.gaming_minutes)
How far to trust a gate is computed, not listed: :func:`maturity` (and
``python -m earned_time maturity``).
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.
"""
@ -33,6 +36,7 @@ from earned_time._credits import (
day_window,
first_credit_at,
)
from earned_time._evidence import real_credit
from earned_time._ladder import (
ANKI_WAIVED_FROM,
LADDER,
@ -49,6 +53,15 @@ from earned_time._ladder import (
shutdown_minutes_for,
)
from earned_time._ledger import done_today, entry_signature, today_window, verified
from earned_time._maturity import (
MATURE_MIN_AGE_DAYS,
MATURE_MIN_CREDIT_DAYS,
Level,
Maturity,
confirmed_start,
maturity,
penalty_start,
)
from earned_time._policy import (
ANKI,
AUTOMATION,
@ -78,6 +91,8 @@ __all__ = [
"LADDER",
"LADDER_FROM",
"LEETCODE",
"MATURE_MIN_AGE_DAYS",
"MATURE_MIN_CREDIT_DAYS",
"READING",
"SHUTDOWN_BASE_MINUTES",
"SHUTDOWN_CEILING_MINUTES",
@ -88,11 +103,14 @@ __all__ = [
"WORKOUT",
"Base",
"Earner",
"Level",
"Maturity",
"Resolution",
"Rung",
"Term",
"all_earners",
"base_for",
"confirmed_start",
"credit_units",
"day_window",
"done_today",
@ -102,7 +120,10 @@ __all__ = [
"extra_shutdown_minutes_for",
"first_credit_at",
"ladder_for",
"maturity",
"on_ladder",
"penalty_start",
"real_credit",
"registries",
"resolve",
"shutdown_ceiling_for",

View File

@ -0,0 +1,9 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""``python -m earned_time``: see :mod:`earned_time._cli`."""
from __future__ import annotations
from earned_time._cli import main
if __name__ == "__main__":
raise SystemExit(main())

View File

@ -0,0 +1,152 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""``python -m earned_time maturity``: every gate's verdict, as a table or JSON.
Claude runs this before answering any "grant me X" request and follows the
verdict: lenient to a ``new`` gate, suspicious for a ``mature`` one.
This is the one place that resolves real paths -- the library never does
(:mod:`earned_time._ledger`). Both are module names so tests redirect them.
It only ever reads a ledger.
"""
from __future__ import annotations
import argparse
from collections import Counter
from dataclasses import replace
from datetime import UTC, date, datetime
import json
from pathlib import Path
import sys
from typing import TYPE_CHECKING, Final
from earned_time._maturity import Maturity, maturity
from earned_time._registry import all_earners, earners_for
if TYPE_CHECKING:
from collections.abc import Sequence
from earned_time._policy import Earner
KEY_FILE: Final = Path("/etc/workout-locker/hmac.key")
# Display only: gates with no earn-back, so nothing here classifies them.
NOT_CLASSIFIED: Final = ("diet-guard", "home-guard", "hosts-blocker", "wake-alarm")
_COLUMNS: Final = (
"gate",
"level",
"days",
"first",
"last",
"confirmed_on",
"penalty_from",
"penalty_start",
"reasons",
)
def home() -> Path:
"""Where ``Earner.ledger`` paths are rooted (the user's home)."""
return Path.home()
def _gates(day: date) -> list[tuple[str, Earner, Maturity]]:
"""Each ledger-backed earner of every registry, with its verdict on ``day``.
Two earners can share a name across registries (``automation`` before and
after ``TUTOR_FROM``); those get their ledger's directory appended.
"""
earners = [e for e in all_earners() if e.ledger is not None]
names = Counter(e.name for e in earners)
registered = earners_for(day)
gates = []
for item in earners:
ledger = Path(str(item.ledger))
gate = item.name
if names[item.name] > 1:
gate = f"{item.name}:{ledger.parent.name}"
verdict = maturity(item, home() / ledger, KEY_FILE, day)
if item not in registered:
extra = f"not in the registry on {day}: neither paid nor penalised"
verdict = replace(verdict, reasons=(*verdict.reasons, extra))
gates.append((gate, item, verdict))
return gates
def _iso(value: date | None) -> str | None:
return None if value is None else value.isoformat()
def _record(gate: str, item: Earner, verdict: Maturity) -> dict[str, object]:
return {
"gate": gate,
"name": verdict.name,
"ledger": item.ledger,
"level": verdict.level,
"credit_days": verdict.credit_days,
"credit_rows": verdict.credit_rows,
"first_credit": _iso(verdict.first_credit),
"last_credit": _iso(verdict.last_credit),
"confirmed_on": _iso(verdict.confirmed_on),
"penalty_from": _iso(verdict.penalty_from),
"penalty_start": _iso(verdict.penalty_start),
"checked": verdict.checked,
"reasons": list(verdict.reasons),
}
def _row(gate: str, verdict: Maturity) -> list[str]:
cells = (
verdict.level,
verdict.credit_days,
verdict.first_credit,
verdict.last_credit,
verdict.confirmed_on,
verdict.penalty_from,
verdict.penalty_start,
)
shown = ["-" if c is None else str(c) for c in cells]
return [gate, *shown, "; ".join(verdict.reasons)]
def render_table(gates: Sequence[tuple[str, Maturity]]) -> str:
"""The verdicts as a fixed-width table; the reasons column runs free."""
rows = [list(_COLUMNS), *(_row(gate, verdict) for gate, verdict in gates)]
widths = [max(len(row[i]) for row in rows) for i in range(len(_COLUMNS) - 1)]
lines = [
" ".join([*(c.ljust(w) for c, w in zip(row, widths, strict=False)), row[-1]])
for row in rows
]
return "\n".join(line.rstrip() for line in lines)
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="python -m earned_time")
commands = parser.add_subparsers(dest="command", required=True)
command = commands.add_parser("maturity", help="how far to trust each gate")
command.add_argument("--json", action="store_true", help="machine-readable")
command.add_argument(
"--day", type=date.fromisoformat, help="verdict day (YYYY-MM-DD)"
)
return parser
def main(argv: Sequence[str] | None = None) -> int:
"""Run the CLI; returns the exit status."""
args = _parser().parse_args(argv)
day: date = args.day or datetime.now(tz=UTC).astimezone().date()
gates = _gates(day)
if args.json:
records = [_record(*gate) for gate in gates]
payload = {
"day": day.isoformat(),
"gates": records,
"not_classified": list(NOT_CLASSIFIED),
}
sys.stdout.write(json.dumps(payload, indent=2) + "\n")
return 0
sys.stdout.write(
f"gate maturity on {day}\n{render_table([(g, v) for g, _, v in gates])}\n"
)
sys.stdout.write(f"not classified (no earn-back): {', '.join(NOT_CLASSIFIED)}\n")
return 0

View File

@ -0,0 +1,138 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""What a gate has really paid out: the evidence :mod:`earned_time._maturity` judges.
A *real* credit is a signed ``credit`` row the earner's matcher accepts, that
no human granted by hand and that pays something -- a gate's own re-evaluation
grant (``detail.grant_of``) may pay 0. A manual grant is exactly the evidence
that the gate did *not* work, so no matcher may count it.
"""
from __future__ import annotations
from datetime import UTC, date, datetime
from typing import TYPE_CHECKING, Final, TypeGuard
from earned_time._credits import credit_time, day_window
from earned_time._ledger import read_key, read_rows, verified
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
from earned_time._policy import Earner
# Rows a human granted by hand: the gate did not earn them.
MANUAL_SOURCES: Final = frozenset({"manual", "manual_grant"})
MANUAL_PREFIXES: Final = ("manual:", "manual_grant:")
def real_credit(row: object, key: bytes) -> TypeGuard[dict[str, object]]:
"""Whether a ledger row is signed, a credit, not manual, and pays something.
A gate re-evaluation grant (``detail.grant_of``) counts at ``amount`` 0;
the manual exclusions still apply to it.
The matcher is asked separately (:func:`credit_day`): this is the part no
matcher may loosen -- ``anki_match`` alone accepts a ``manual_grant`` row.
"""
if not isinstance(row, dict) or row.get("kind") != "credit":
return False
if not verified(row, key) or _is_manual(row):
return False
return _pays(row)
def _is_manual(row: dict[str, object]) -> bool:
"""Whether a human granted ``row`` by hand (its source or its entry id)."""
detail = row.get("detail")
if isinstance(detail, dict) and detail.get("source") in MANUAL_SOURCES:
return True
return str(row.get("entry_id")).startswith(MANUAL_PREFIXES)
def _pays(row: dict[str, object]) -> bool:
"""``amount > 0`` (or no amount); ``amount`` 0 only for a gate's own grant."""
if "amount" not in row:
return True
amount = row["amount"]
if not isinstance(amount, int | float) or amount < 0:
return False
return bool(amount) or _is_grant(row.get("detail"))
def _is_grant(detail: object) -> bool:
"""Whether ``detail`` marks a gate re-evaluation grant (``grant_of``).
book-guard re-grades a past session's bonus under the current rule and
signs the result as its own ``bonus:`` row, paying ``amount`` 0 when no
pages were added: the gate worked that day, so it still counts.
"""
if not isinstance(detail, dict):
return False
grant_of = detail.get("grant_of")
return isinstance(grant_of, str) and bool(grant_of)
def _candidate_days(earner: Earner, row: dict[str, object]) -> Iterator[date]:
"""Days a row might count for: its stamp's day, its ``day``, its Anki day."""
stamp = credit_time(earner, row)
if stamp is not None:
yield datetime.fromtimestamp(stamp, tz=UTC).astimezone().date()
detail = row.get("detail")
anki_day = detail.get("anki_day") if isinstance(detail, dict) else None
for raw in (row.get("day"), anki_day):
try:
yield date.fromisoformat(str(raw))
except ValueError:
continue
def credit_day(earner: Earner, row: dict[str, object]) -> date | None:
"""The local day ``earner.match`` counts ``row`` for, or ``None``.
The matcher decides, exactly as it does for today's answer; the
candidates only say which days to ask it about.
Raises:
ValueError: ``earner`` has no ``match``.
"""
if earner.match is None:
msg = f"earner {earner.name!r} has no shared reader; supply its answer"
raise ValueError(msg)
match = earner.match
for day in dict.fromkeys(_candidate_days(earner, row)):
if match(row, day_window(day)):
return day
return None
def credit_history(
earner: Earner, ledger: Path, key_file: Path, today: date, reasons: list[str]
) -> tuple[list[date], int] | None:
"""Sorted real-credit days up to ``today``, and how many rows paid them.
``None`` -- with the reason appended to ``reasons`` -- when the key or
the ledger cannot be read. A replayed ``entry_id`` counts once.
"""
key = read_key(key_file)
if key is None:
reasons.append(f"could not check: HMAC key {key_file} unreadable")
return None
rows = read_rows(ledger, earner)
if rows is None:
reasons.append(f"could not check: ledger {ledger} unreadable")
return None
days: set[date] = set()
ids: set[object] = set()
paid = 0
for row in rows:
if not real_credit(row, key):
continue
day = credit_day(earner, row)
entry_id = row.get("entry_id")
if day is None or day > today or (entry_id is not None and entry_id in ids):
continue
ids.add(entry_id)
days.add(day)
paid += 1
return sorted(days), paid

View File

@ -0,0 +1,194 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""How far to trust a gate: computed from its signed ledger, never listed.
A gate is **new** until it has paid out for real and kuhy has confirmed a
working session (``Earner.confirmed_on``), **mature** once it has paid on
:data:`MATURE_MIN_CREDIT_DAYS` distinct days and its first payout is at least
:data:`MATURE_MIN_AGE_DAYS` old, and **maturing** in between. A new gate is
lenient (no penalty before it has worked); a mature one is suspicious of
"just grant me the points".
Only a *real* credit counts (:func:`~earned_time._evidence.real_credit`):
HMAC-verified, accepted by the earner's matcher, not a hand-granted row, and
paying something. A manual grant is exactly the evidence that the gate did
*not* work.
**The penalty never starts before the gate has paid out**
(:func:`penalty_start`): a gate that never earned anything cannot cost
anything. The adjustment can only delay a penalty. A ledger or key that
cannot be read is "could not check": no evidence of maturity.
**Fail closed once confirmed.** A gate with a hand-set ``confirmed_on`` has
been seen to work, so "no credit on record" -- unreadable key or ledger, a
deleted ledger, an emptied one -- cannot lift its penalty: it falls back to
:func:`confirmed_start`. Only a gate without ``confirmed_on`` is lenient.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, date, datetime, timedelta
from typing import TYPE_CHECKING, Final, Literal
from earned_time._evidence import credit_history
if TYPE_CHECKING:
from pathlib import Path
from earned_time._policy import Earner
Level = Literal["new", "maturing", "mature"]
# Distinct days with a real credit, and how old the first one must be.
MATURE_MIN_CREDIT_DAYS: Final = 14
MATURE_MIN_AGE_DAYS: Final = 21
_DAY: Final = timedelta(days=1)
@dataclass(frozen=True)
class Maturity:
"""One gate's verdict, with the evidence behind it.
Attributes:
name: The earner's registry name.
level: ``"new"``, ``"maturing"`` or ``"mature"``.
reasons: Human-readable, in the order they were decided.
credit_days: Distinct local days, up to the verdict's day, with a
real credit.
credit_rows: Real credit rows behind them (a replayed ``entry_id`` once).
first_credit: The first such day, or ``None``.
last_credit: The last such day, or ``None``.
confirmed_on: The earner's ``confirmed_on``.
penalty_from: The earner's ``penalty_from``.
penalty_start: The effective first penalised day
(:func:`penalty_start`); ``None`` means not penalised.
checked: ``False`` when the ledger or the key could not be read.
"""
name: str
level: Level
reasons: tuple[str, ...]
credit_days: int
credit_rows: int
first_credit: date | None
last_credit: date | None
confirmed_on: date | None
penalty_from: date | None
penalty_start: date | None
checked: bool
def confirmed_start(earner: Earner) -> date | None:
"""``max(penalty_from, day after confirmed_on)``; ``None`` for a pure bonus.
What is left of :func:`penalty_start` when the ledger was not consulted:
a ``confirmed_on`` of ``None`` drops out of the max.
"""
if earner.penalty_from is None:
return None
if earner.confirmed_on is None:
return earner.penalty_from
return max(earner.penalty_from, earner.confirmed_on + _DAY)
def penalty_start(earner: Earner, first_credit: date | None) -> date | None:
"""The first day ``earner``'s penalty may apply, given its first real credit.
``max(penalty_from, day after first_credit, day after confirmed_on)``;
a ``confirmed_on`` of ``None`` drops out of the max. No first credit
(never paid out, or could not check): :func:`confirmed_start` when kuhy
set ``confirmed_on`` -- fail closed, deleting a ledger lifts nothing --
else ``None``, not penalised. ``None`` for a pure bonus. Never earlier
than ``penalty_from``.
"""
start = confirmed_start(earner)
if start is None:
return None
if first_credit is None:
return start if earner.confirmed_on is not None else None
return max(start, first_credit + _DAY)
def _level(days: list[date], earner: Earner, today: date, reasons: list[str]) -> Level:
if not days:
reasons.append("no verified credit yet")
return "new"
if earner.confirmed_on is None:
reasons.append("no confirmed_on: kuhy has not confirmed a real session")
return "new"
age = (today - days[0]).days
if len(days) >= MATURE_MIN_CREDIT_DAYS and age >= MATURE_MIN_AGE_DAYS:
reasons.append(
f"{len(days)} credit days >= {MATURE_MIN_CREDIT_DAYS} and first "
f"credit {age} days ago >= {MATURE_MIN_AGE_DAYS}"
)
return "mature"
reasons.append(
f"{len(days)}/{MATURE_MIN_CREDIT_DAYS} credit days, first credit "
f"{age}/{MATURE_MIN_AGE_DAYS} days ago"
)
return "maturing"
def _penalty_reason(
earner: Earner, start: date | None, first: date | None, *, checked: bool
) -> str:
if earner.penalty_from is None:
return "no penalty_from: a pure bonus"
if start is not None and first is None:
why = "could not check" if not checked else "no verified credit"
return (
f"{why} but confirmed_on {earner.confirmed_on}: "
f"fail closed, penalty from {start}"
)
if start is None:
return f"never paid out: no penalty (penalty_from {earner.penalty_from})"
if start > earner.penalty_from:
return f"penalty delayed from {earner.penalty_from} to {start}"
return f"penalty from {start}, as registered"
def maturity(
earner: Earner, ledger: Path, key_file: Path, today: date | None = None
) -> Maturity:
"""Classify ``earner`` from its signed ledger, as of ``today`` (default now).
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.
today: The verdict's day; credits after it are ignored.
Returns:
The verdict. An unreadable ledger or key is ``checked=False``: level
``"new"``, with the reason recorded; no penalty unless ``confirmed_on``
is set (then :func:`confirmed_start`, fail closed).
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)
target = today or datetime.now(tz=UTC).astimezone().date()
reasons: list[str] = []
found = credit_history(earner, ledger, key_file, target, reasons)
days, paid = found if found is not None else ([], 0)
level = _level(days, earner, target, reasons)
first = days[0] if days else None
start = penalty_start(earner, first)
reasons.append(_penalty_reason(earner, start, first, checked=found is not None))
return Maturity(
name=earner.name,
level=level,
reasons=tuple(reasons),
credit_days=len(days),
credit_rows=paid,
first_credit=first,
last_credit=days[-1] if days else None,
confirmed_on=earner.confirmed_on,
penalty_from=earner.penalty_from,
penalty_start=start,
checked=found is not None,
)

View File

@ -80,6 +80,9 @@ class Earner:
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.
confirmed_on: The day kuhy confirmed a real working session; set by
hand, never derived. ``None`` keeps the gate "new"
(:func:`earned_time.maturity`).
``shutdown_minutes`` and ``extra_shutdown_minutes`` are the pre-ladder
values; from ``LADDER_FROM`` the earner's :class:`Rung` in ``LADDER``
@ -98,6 +101,7 @@ class Earner:
missing_ledger_is_no: bool = False
extra_gaming_minutes: int = 0
max_units: int | None = None
confirmed_on: date | None = None
def capped(self, units: int) -> int:
"""``units`` clamped to ``[0, max_units]``."""
@ -138,6 +142,9 @@ WORKOUT: Final = Earner(
ledger=".local/share/workout_locker/ledger.json",
match=workout_match,
missing_ledger_is_no=True,
# The day of its first verified credit, read off the real ledger on
# 2026-10-09 (earned_time 0.6.0); kuhy had been using the gate for real.
confirmed_on=date(2026, 10, 1),
)
LEETCODE: Final = Earner(
name="leetcode",
@ -146,6 +153,9 @@ LEETCODE: Final = Earner(
shutdown_minutes=60,
ledger=".local/share/leetcode_guard/ledger.json",
match=leetcode_match,
# The day of its first verified credit, read off the real ledger on
# 2026-10-09 (earned_time 0.6.0); kuhy had been using the gate for real.
confirmed_on=date(2026, 8, 12),
)
READING: Final = Earner(
name="reading",
@ -156,6 +166,9 @@ READING: Final = Earner(
ledger=".local/share/book_guard/ledger.json",
match=reading_match,
missing_ledger_is_no=True,
# The day of its first verified credit, read off the real ledger on
# 2026-10-09 (earned_time 0.6.0); kuhy had been using the gate for real.
confirmed_on=date(2026, 10, 2),
)
ANKI: Final = Earner(
name="anki",

View File

@ -10,6 +10,12 @@ 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.
**Maturity can only delay a penalty.** Given ``first_credits`` (each earner's
first real credit, from :func:`earned_time.maturity`), a penalty starts no
earlier than the day after the gate first paid out
(:func:`~earned_time._maturity.penalty_start`). Without it every day resolves
exactly as in 0.5.0.
"""
from __future__ import annotations
@ -20,6 +26,7 @@ import logging
from typing import TYPE_CHECKING, Final
from earned_time._ladder import on_ladder, shutdown_ceiling_for
from earned_time._maturity import confirmed_start, penalty_start
from earned_time._policy import (
GAMING_BASE_MINUTES,
GAMING_CEILING_MINUTES,
@ -101,7 +108,32 @@ def _today() -> date:
return datetime.now(tz=UTC).astimezone().date()
def _shutdown_floor(day: date, earners: tuple[Earner, ...]) -> int:
def _penalised(
item: Earner, day: date, first_credits: Mapping[str, date | None] | None
) -> bool:
"""Whether ``item``'s cut is in force on ``day``.
Without ``first_credits``: ``penalty_from`` alone, as in 0.5.0. With it:
:func:`penalty_start` of the earner's first real credit (``None`` -- never
paid out, or could not check -- is not penalised unless ``confirmed_on``
is set: fail closed); an earner missing from it still waits for the day
after its ``confirmed_on``.
"""
if first_credits is None:
return item.penalised_on(day)
start = (
penalty_start(item, first_credits[item.name])
if item.name in first_credits
else confirmed_start(item)
)
return start is not None and day >= start
def _shutdown_floor(
day: date,
earners: tuple[Earner, ...],
first_credits: Mapping[str, date | None] | None = None,
) -> int:
"""Shutdown of a day that earned nothing.
On the ladder: the ceiling minus the most every registered earner can
@ -112,26 +144,31 @@ def _shutdown_floor(day: date, earners: tuple[Earner, ...]) -> int:
return shutdown_ceiling_for(day) - sum(
e.shutdown_for(e.max_units or 1, day) for e in earners
)
penalised = [e for e in earners if e.penalised_on(day)]
penalised = [e for e in earners if _penalised(e, day, first_credits)]
return SHUTDOWN_BASE_MINUTES - sum(e.shutdown_minutes for e in penalised)
def base_for(
day: date | None = None, earners: tuple[Earner, ...] | None = None
day: date | None = None,
earners: tuple[Earner, ...] | None = None,
*,
first_credits: Mapping[str, date | None] | None = None,
) -> Base:
"""The floor for ``day`` (default today) under ``earners`` (default its own).
Gaming: every penalty in force, the most each earner can pay, taken off
the gaming base. Shutdown: see :func:`_shutdown_floor` -- the two are
derived separately so the ladder can never move a gaming minute.
``first_credits`` (per earner name) delays each penalty to
:func:`penalty_start`; omitted, 0.5.0's ``penalty_from`` applies.
"""
target = day or _today()
registry = earners_for(target) if earners is None else earners
penalised = [e for e in registry if e.penalised_on(target)]
penalised = [e for e in registry if _penalised(e, target, first_credits)]
return Base(
gaming_minutes=GAMING_BASE_MINUTES
- sum(e.max_gaming_minutes for e in penalised),
shutdown_minutes=_shutdown_floor(target, registry),
shutdown_minutes=_shutdown_floor(target, registry, first_credits),
)
@ -141,10 +178,23 @@ def _units(answer: Answer) -> int | None:
return max(0, int(answer))
def _check_registered(
answers: Mapping[str, Answer], registry: tuple[Earner, ...]
) -> None:
"""Raise ``KeyError`` for an answer no registry knows: a typo, not a "no"."""
known = {e.name for e in (*registry, *all_earners())}
unknown = sorted(set(answers) - known)
if unknown:
msg = f"not registered earners: {', '.join(unknown)}"
raise KeyError(msg)
def resolve(
answers: Mapping[str, Answer],
day: date | None = None,
earners: tuple[Earner, ...] | None = None,
*,
first_credits: Mapping[str, date | None] | None = None,
) -> Resolution:
"""Sum the base and every earned term, each capped at its ceiling.
@ -157,6 +207,9 @@ def resolve(
earners: The registry (default :func:`earners_for` the day); tests
pass their own. An answer for an earner registered on other days
only (Anki after ``TUTOR_FROM``) is ignored.
first_credits: Per earner name, its first real credit day (``None``
for never, or could not check), e.g. ``Maturity.first_credit``.
Only ever delays a penalty; omitted, 0.5.0's ``penalty_from``.
Returns:
The resolution, with a term for every registered earner.
@ -167,12 +220,8 @@ def resolve(
"""
target = day or _today()
registry = earners_for(target) if earners is None else earners
known = {e.name for e in (*registry, *all_earners())}
unknown = sorted(set(answers) - known)
if unknown:
msg = f"not registered earners: {', '.join(unknown)}"
raise KeyError(msg)
base = base_for(target, registry)
_check_registered(answers, registry)
base = base_for(target, registry, first_credits=first_credits)
terms: list[Term] = []
for item in registry:
if item.name not in answers:

View File

@ -0,0 +1,67 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Signed ledger rows and files for the maturity tests."""
from __future__ import annotations
from datetime import date, datetime, time, timedelta
import json
from typing import TYPE_CHECKING
from earned_time import entry_signature
if TYPE_CHECKING:
from pathlib import Path
KEY = b"test-key"
TODAY = date(2026, 10, 9)
def at(day: date, hour: int = 12) -> str:
"""``day`` at ``hour`` local, as a unix-stamp string."""
return str(datetime.combine(day, time(hour)).astimezone().timestamp())
def signed(row: dict[str, object]) -> dict[str, object]:
"""``row`` with its HMAC under :data:`KEY`."""
return {**row, "hmac": entry_signature(row, KEY)}
def solve(day: date, entry: str | None = None, **extra: object) -> dict[str, object]:
"""A signed LeetCode solve credited on ``day``."""
row: dict[str, object] = {
"kind": "credit",
"entry_id": entry or f"ac:{day}",
"day": day.isoformat(),
"amount": 1,
"detail": {"source": "leetcode", "submitted_at": at(day)},
**extra,
}
return signed(row)
def ledger_file(tmp_path: Path, rows: list[object]) -> Path:
"""A ledger file holding ``rows``."""
path = tmp_path / "ledger.json"
path.write_text(json.dumps({"entries": rows}), encoding="utf-8")
return path
def grant(
*,
entry: str = "bonus:session:a",
amount: object = 0,
grant_of: object = "session:a",
source: str | None = None,
) -> dict[str, object]:
"""A book-guard bonus re-evaluation row: signed, ``amount`` 0, ``grant_of``."""
detail: dict[str, object] = {"bonus": "1", "ended_at": at(TODAY)}
detail["grant_of"] = grant_of
if source is not None:
detail["source"] = source
row = {"kind": "credit", "entry_id": entry, "amount": amount, "detail": detail}
return signed(row)
def days(first: date, count: int) -> list[object]:
"""One solve on each of ``count`` days from ``first``."""
return [solve(first + timedelta(days=i)) for i in range(count)]

View File

@ -0,0 +1,21 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Shared fixtures: the HMAC key file the maturity rows are signed with."""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from earned_time.tests._maturity_rows import KEY
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture
def key_file(tmp_path: Path) -> Path:
"""The test HMAC key, written where a test can point ``maturity`` at it."""
path = tmp_path / "hmac.key"
path.write_bytes(KEY)
return path

View File

@ -0,0 +1,131 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""``python -m earned_time maturity``: table, JSON, and the paths it reads."""
from __future__ import annotations
from datetime import date, datetime, time
import json
import runpy
import sys
from typing import TYPE_CHECKING
import pytest
import earned_time
from earned_time import LEETCODE, READING, Earner, entry_signature
from earned_time import _cli as cli
if TYPE_CHECKING:
from pathlib import Path
KEY = b"test-key"
DAY = date(2026, 10, 9)
@pytest.fixture
def home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""A fake home with a key and one LeetCode solve on DAY."""
key = tmp_path / "hmac.key"
key.write_bytes(KEY)
monkeypatch.setattr(cli, "KEY_FILE", key)
monkeypatch.setattr(cli, "home", lambda: tmp_path)
stamp = datetime.combine(DAY, time(12)).astimezone().timestamp()
row: dict[str, object] = {
"kind": "credit",
"entry_id": "ac:1",
"day": DAY.isoformat(),
"amount": 1,
"detail": {"submitted_at": str(stamp)},
}
ledger = tmp_path / str(LEETCODE.ledger)
ledger.parent.mkdir(parents=True)
entries = [{**row, "hmac": entry_signature(row, KEY)}]
ledger.write_text(json.dumps({"entries": entries}), encoding="utf-8")
return tmp_path
def test_table(home: Path, capsys: pytest.CaptureFixture[str]) -> None:
assert cli.main(["maturity", "--day", DAY.isoformat()]) == 0
out = capsys.readouterr().out.splitlines()
assert out[0] == f"gate maturity on {DAY}"
assert out[1].split()[:3] == ["gate", "level", "days"]
rows = {line.split()[0]: line for line in out[2:-1]}
assert set(rows) == {
"workout",
"leetcode",
"reading",
"anki",
"automation:anki_guard",
"automation:automation_tutor",
}
assert rows["leetcode"].split()[1:5] == ["maturing", "1", str(DAY), str(DAY)]
assert "not in the registry on 2026-10-09" in rows["anki"]
assert "could not check: ledger" not in rows["reading"]
assert rows["reading"].endswith(
"no verified credit but confirmed_on 2026-10-02: "
"fail closed, penalty from 2026-10-03"
)
assert out[-1].startswith("not classified (no earn-back): diet-guard")
def test_json(home: Path, capsys: pytest.CaptureFixture[str]) -> None:
assert cli.main(["maturity", "--json", "--day", DAY.isoformat()]) == 0
payload = json.loads(capsys.readouterr().out)
assert payload["day"] == DAY.isoformat()
assert payload["not_classified"] == list(cli.NOT_CLASSIFIED)
gates = {g["gate"]: g for g in payload["gates"]}
assert gates["leetcode"]["first_credit"] == DAY.isoformat()
assert gates["leetcode"]["checked"] is True
assert gates["reading"]["penalty_from"] == READING.penalty_from.isoformat()
# No reading ledger, but confirmed_on is set: fail closed, not lifted.
assert gates["reading"]["penalty_start"] == "2026-10-03"
assert gates["anki"]["penalty_start"] is None
assert gates["anki"]["last_credit"] is None
def test_default_day_and_unregistered_earners(
home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
consumer = Earner(name="piano", label="piano", gaming_minutes=1, shutdown_minutes=1)
monkeypatch.setattr(earned_time, "EARNERS", (consumer, LEETCODE))
assert cli.main(["maturity"]) == 0
lines = capsys.readouterr().out.splitlines()
assert [line.split()[0] for line in lines[2:-1]] == [
"leetcode",
"workout",
"reading",
"automation",
]
def test_render_table_pads_all_but_the_reasons() -> None:
verdict = earned_time.Maturity(
name="x",
level="new",
reasons=("a", "b"),
credit_days=0,
credit_rows=0,
first_credit=None,
last_credit=None,
confirmed_on=None,
penalty_from=None,
penalty_start=None,
checked=True,
)
table = cli.render_table([("x", verdict)]).splitlines()
assert table[1].split() == ["x", "new", "0", *["-"] * 5, "a;", "b"]
assert len(table) == 2
def test_home_is_the_users_home() -> None:
assert cli.home().is_absolute()
def test_python_dash_m(
home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr(sys, "argv", ["earned_time", "maturity", "--json"])
with pytest.raises(SystemExit) as exit_info:
runpy.run_module("earned_time", run_name="__main__")
assert exit_info.value.code == 0
assert json.loads(capsys.readouterr().out)["gates"]

View File

@ -0,0 +1,120 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The maturity levels, the counts behind them and the penalty reasons."""
from __future__ import annotations
from dataclasses import replace
from datetime import date, timedelta
from typing import TYPE_CHECKING
import pytest
from earned_time import (
LEETCODE,
MATURE_MIN_AGE_DAYS,
MATURE_MIN_CREDIT_DAYS,
READING,
Earner,
maturity,
)
from earned_time._evidence import credit_day
from earned_time.tests._maturity_rows import (
TODAY,
at,
days,
ledger_file,
signed,
solve,
)
if TYPE_CHECKING:
from pathlib import Path
def test_an_earner_without_a_matcher_raises(tmp_path: Path, key_file: Path) -> None:
bare = Earner(name="x", label="x", gaming_minutes=1, shutdown_minutes=1)
with pytest.raises(ValueError, match="no shared reader"):
maturity(bare, tmp_path / "none.json", key_file, TODAY)
with pytest.raises(ValueError, match="no shared reader"):
credit_day(bare, {})
def test_mature_at_the_thresholds(tmp_path: Path, key_file: Path) -> None:
first = TODAY - timedelta(days=MATURE_MIN_AGE_DAYS)
ledger = ledger_file(tmp_path, days(first, MATURE_MIN_CREDIT_DAYS))
verdict = maturity(LEETCODE, ledger, key_file, TODAY)
assert verdict.level == "mature"
assert verdict.credit_days == MATURE_MIN_CREDIT_DAYS
assert verdict.first_credit == first
assert verdict.checked
assert verdict.reasons[-1] == "no penalty_from: a pure bonus"
@pytest.mark.parametrize(
("age", "count"),
[
(MATURE_MIN_AGE_DAYS - 1, MATURE_MIN_CREDIT_DAYS),
(MATURE_MIN_AGE_DAYS + 10, MATURE_MIN_CREDIT_DAYS - 1),
],
)
def test_maturing_below_either_threshold(
tmp_path: Path, key_file: Path, age: int, count: int
) -> None:
ledger = ledger_file(tmp_path, days(TODAY - timedelta(days=age), count))
verdict = maturity(LEETCODE, ledger, key_file, TODAY)
assert verdict.level == "maturing"
assert verdict.reasons[0].startswith(f"{count}/{MATURE_MIN_CREDIT_DAYS} credit")
def test_unconfirmed_gate_stays_new(tmp_path: Path, key_file: Path) -> None:
ledger = ledger_file(tmp_path, days(TODAY - timedelta(days=60), 40))
verdict = maturity(replace(LEETCODE, confirmed_on=None), ledger, key_file, TODAY)
assert verdict.level == "new"
assert "no confirmed_on" in verdict.reasons[0]
assert verdict.credit_days == 40
def test_counts_skip_future_replayed_and_unmatched_rows(
tmp_path: Path, key_file: Path
) -> None:
rows: list[object] = [
solve(TODAY),
solve(TODAY), # replayed: same entry_id
solve(TODAY + timedelta(days=1)), # after the verdict's day
solve(TODAY, entry="ac:other"),
signed({"kind": "credit", "detail": {"submitted_at": at(TODAY)}}),
signed({"kind": "credit", "detail": {"submitted_at": at(TODAY)}}),
signed({"kind": "credit", "entry_id": "no-day", "detail": {}}),
]
verdict = maturity(LEETCODE, ledger_file(tmp_path, rows), key_file, TODAY)
assert (verdict.credit_rows, verdict.credit_days) == (4, 1)
assert verdict.first_credit == verdict.last_credit == TODAY
def test_penalty_reasons(tmp_path: Path, key_file: Path) -> None:
early = replace(
READING,
penalty_from=date(2026, 10, 5),
confirmed_on=date(2026, 10, 1),
)
first = date(2026, 10, 2)
row = signed(
{
"kind": "credit",
"entry_id": "session:a",
"amount": 44,
"detail": {"bonus": "1", "ended_at": at(first)},
}
)
ledger = ledger_file(tmp_path, [row])
on_time = maturity(early, ledger, key_file, TODAY)
assert on_time.penalty_start == early.penalty_from
assert on_time.reasons[-1] == "penalty from 2026-10-05, as registered"
late = maturity(READING, ledger, key_file, TODAY)
assert late.penalty_start == date(2026, 10, 3)
assert late.reasons[-1] == "penalty delayed from 2026-10-01 to 2026-10-03"
def test_today_defaults_to_now(tmp_path: Path, key_file: Path) -> None:
verdict = maturity(READING, tmp_path / "missing.json", key_file)
assert verdict.level == "new"

View File

@ -0,0 +1,169 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""What counts as a real credit, and failing closed when none is on record."""
from __future__ import annotations
from dataclasses import replace
from datetime import date, timedelta
from typing import TYPE_CHECKING
import pytest
from earned_time import (
ANKI,
LEETCODE,
READING,
WORKOUT,
maturity,
real_credit,
)
from earned_time._evidence import credit_day
from earned_time.tests._maturity_rows import (
KEY,
TODAY,
at,
grant,
ledger_file,
signed,
solve,
)
if TYPE_CHECKING:
from pathlib import Path
@pytest.mark.parametrize(
("row", "real"),
[
(solve(TODAY), True),
({**solve(TODAY), "hmac": "forged"}, False),
("not a row", False),
(signed({"kind": "charge", "entry_id": "x", "amount": 1}), False),
(signed({"kind": "credit", "entry_id": "x", "detail": "flat"}), True),
(signed({"kind": "credit", "entry_id": "x"}), True),
(solve(TODAY, detail={"source": "manual"}), False),
(solve(TODAY, detail={"source": "manual_grant"}), False),
(solve(TODAY, entry="manual:2026-10-09"), False),
(solve(TODAY, entry="manual_grant:anki:2026-10-09"), False),
(solve(TODAY, amount=0), False),
(solve(TODAY, amount=-3), False),
(solve(TODAY, amount="5"), False),
(solve(TODAY, amount=2.5), True),
(grant(), True),
(grant(amount=-1), False),
(grant(amount="0"), False),
(grant(source="manual_grant"), False),
(grant(entry="manual_grant:session:a"), False),
({**grant(), "hmac": "forged"}, False),
({k: v for k, v in grant().items() if k != "hmac"}, False),
(grant(grant_of=""), False),
(grant(grant_of=7), False),
(solve(TODAY, amount=0, detail="flat"), False),
],
)
def test_real_credit(row: object, real: object) -> None:
assert real_credit(row, KEY) is real
def test_a_matcher_cannot_launder_a_manual_grant(
tmp_path: Path, key_file: Path
) -> None:
grant = signed(
{
"kind": "credit",
"entry_id": "manual_grant:anki:2026-10-09",
"amount": 0,
"detail": {"anki_day": "2026-10-09", "source": "manual_grant"},
}
)
assert ANKI.match is not None
assert credit_day(ANKI, grant) == TODAY # anki_match alone would count it
verdict = maturity(ANKI, ledger_file(tmp_path, [grant]), key_file, TODAY)
assert (verdict.level, verdict.credit_days, verdict.penalty_start) == (
"new",
0,
None,
)
def test_credit_day_asks_the_matcher_on_each_candidate_day() -> None:
rest: dict[str, object] = {
"kind": "credit",
"entry_id": "rest_day:2026-10-09",
"day": "2026-10-09",
"created_at": f"{TODAY - timedelta(days=1)}T20:00:00+00:00",
"detail": {"source": "rest_day", "declared_at": "1"},
}
assert credit_day(WORKOUT, rest) == TODAY # by its ``day``, not created_at
bare = {"kind": "credit", "entry_id": "x", "day": "soon", "detail": {}}
assert credit_day(LEETCODE, bare) is None
flat = {"kind": "credit", "entry_id": "x", "day": "soon", "detail": "flat"}
assert credit_day(ANKI, flat) is None
def test_a_re_evaluation_grant_counts_its_day(tmp_path: Path, key_file: Path) -> None:
session = signed(
{
"kind": "credit",
"entry_id": "session:a",
"amount": 17,
"detail": {"bonus": "0", "ended_at": at(TODAY)},
}
)
ledger = ledger_file(tmp_path, [session, grant()])
verdict = maturity(READING, ledger, key_file, TODAY)
assert (verdict.credit_days, verdict.credit_rows) == (1, 1)
assert verdict.first_credit == TODAY
def test_unreadable_key_or_ledger_is_new(tmp_path: Path, key_file: Path) -> None:
unconfirmed = replace(READING, confirmed_on=None)
no_key = maturity(unconfirmed, ledger_file(tmp_path, []), tmp_path / "nokey", TODAY)
assert not no_key.checked
assert no_key.reasons[0].startswith("could not check: HMAC key")
no_ledger = maturity(LEETCODE, tmp_path / "missing.json", key_file, TODAY)
assert not no_ledger.checked
assert no_ledger.reasons[0].startswith("could not check: ledger")
for verdict in (no_key, no_ledger):
assert (verdict.level, verdict.penalty_start, verdict.credit_rows) == (
"new",
None,
0,
)
assert no_key.reasons[-1].startswith("never paid out: no penalty")
def test_a_confirmed_gate_fails_closed_when_it_cannot_be_checked(
tmp_path: Path, key_file: Path
) -> None:
assert READING.confirmed_on == date(2026, 10, 2)
expected = date(2026, 10, 3) # max(penalty_from 10-01, confirmed_on + 1)
no_key = maturity(READING, ledger_file(tmp_path, []), tmp_path / "nokey", TODAY)
locked = tmp_path / "locked.json"
locked.write_text("{not json", encoding="utf-8")
no_ledger = maturity(READING, locked, key_file, TODAY)
for verdict in (no_key, no_ledger):
assert not verdict.checked
assert (verdict.level, verdict.penalty_start) == ("new", expected)
assert verdict.reasons[-1] == (
"could not check but confirmed_on 2026-10-02: "
"fail closed, penalty from 2026-10-03"
)
def test_deleting_a_confirmed_gates_ledger_lifts_nothing(
tmp_path: Path, key_file: Path
) -> None:
# READING's missing ledger is an honest "no" (missing_ledger_is_no), so
# this is *checked* with no credit -- and still fails closed.
verdict = maturity(READING, tmp_path / "missing.json", key_file, TODAY)
assert verdict.checked
assert verdict.reasons[0] == "no verified credit yet"
assert verdict.penalty_start == date(2026, 10, 3)
assert verdict.reasons[-1] == (
"no verified credit but confirmed_on 2026-10-02: "
"fail closed, penalty from 2026-10-03"
)
unconfirmed = replace(READING, confirmed_on=None)
lenient = maturity(unconfirmed, tmp_path / "missing.json", key_file, TODAY)
assert lenient.penalty_start is None

View File

@ -0,0 +1,141 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Maturity only ever delays a penalty: ``first_credits`` vs 0.5.0's resolve."""
from __future__ import annotations
from dataclasses import replace
from datetime import date, timedelta
import pytest
from earned_time import (
ANKI_WAIVED_FROM,
LADDER_FROM,
READING,
Earner,
base_for,
confirmed_start,
earners_for,
penalty_start,
resolve,
)
DAY = timedelta(days=1)
# Pre-waiver (reading cut only), pre-waiver with Anki/Automation cut too,
# the first waiver day, and the first ladder day.
DAYS = (date(2026, 10, 3), date(2026, 10, 7), ANKI_WAIVED_FROM, LADDER_FROM)
# A first credit relative to each earner's penalty_from; "missing" leaves the
# earner out of the map, None is "never paid out / could not check".
FIRSTS = ("missing", None, -5, 0, 3)
CONFIRMED = (None, -5, 3)
def _anchor(item: Earner) -> date:
return item.penalty_from or date(2026, 9, 1)
def _registry(day: date, confirmed: int | None) -> tuple[Earner, ...]:
return tuple(
replace(
e,
confirmed_on=None if confirmed is None else _anchor(e) + confirmed * DAY,
)
for e in earners_for(day)
)
def _firsts(
registry: tuple[Earner, ...], first: str | int | None
) -> dict[str, date | None]:
if first == "missing":
return {}
if first is None:
return dict.fromkeys((e.name for e in registry), None)
assert isinstance(first, int)
return {e.name: _anchor(e) + first * DAY for e in registry}
@pytest.mark.parametrize("day", DAYS)
@pytest.mark.parametrize("first", FIRSTS)
@pytest.mark.parametrize("confirmed", CONFIRMED)
@pytest.mark.parametrize("done", [0, 1])
def test_raise_only_on_every_combination(
day: date, first: str | int | None, confirmed: int | None, done: int
) -> None:
registry = _registry(day, confirmed)
answers = {e.name: done for e in registry}
old = resolve(answers, day, registry) # 0.5.0: no first_credits
new = resolve(answers, day, registry, first_credits=_firsts(registry, first))
assert new.gaming_minutes >= old.gaming_minutes
assert new.shutdown_minutes >= old.shutdown_minutes
assert new.base.gaming_minutes >= old.base.gaming_minutes
assert new.base.shutdown_minutes >= old.base.shutdown_minutes
if first == -5 and confirmed in {None, -5}:
# Paid out and confirmed well before the cut: exactly 0.5.0.
assert new == old
def test_omitting_first_credits_is_0_5_0() -> None:
for day in DAYS:
registry = earners_for(day)
old = [e for e in registry if e.penalised_on(day)]
expected = 300 - sum(e.max_gaming_minutes for e in old)
assert base_for(day).gaming_minutes == expected
def test_an_unconfirmed_gate_that_never_paid_out_costs_nothing() -> None:
day = date(2026, 10, 3)
unconfirmed = replace(READING, confirmed_on=None)
registry = (unconfirmed,)
cut = base_for(day, registry)
lifted = base_for(day, registry, first_credits={"reading": None})
assert lifted.gaming_minutes - cut.gaming_minutes == READING.max_gaming_minutes
assert lifted.shutdown_minutes - cut.shutdown_minutes == READING.shutdown_minutes
def test_a_confirmed_gate_with_no_credit_on_record_fails_closed() -> None:
# Could not check, or the ledger was deleted: Maturity.first_credit None.
assert READING.confirmed_on == date(2026, 10, 2)
assert penalty_start(READING, None) == date(2026, 10, 3)
for day, penalised in ((date(2026, 10, 2), False), (date(2026, 10, 3), True)):
base = base_for(day, (READING,), first_credits={"reading": None})
assert base.gaming_minutes == 300 - 60 * penalised
def test_the_penalty_waits_for_the_day_after_the_first_credit() -> None:
first = {"reading": date(2026, 10, 2)}
assert READING.penalty_from == date(2026, 10, 1)
for day, penalised in (
(date(2026, 10, 1), False),
(date(2026, 10, 2), False),
(date(2026, 10, 3), True),
):
base = base_for(day, (READING,), first_credits=first)
assert base.gaming_minutes == 300 - 60 * penalised
def test_a_missing_earner_still_waits_for_its_confirmation() -> None:
late = replace(READING, confirmed_on=date(2026, 10, 4))
assert confirmed_start(late) == date(2026, 10, 5)
for day, penalised in ((date(2026, 10, 4), False), (date(2026, 10, 5), True)):
base = base_for(day, (late,), first_credits={})
assert base.gaming_minutes == 300 - 60 * penalised
def test_penalty_start_never_precedes_penalty_from() -> None:
bonus = replace(READING, penalty_from=None)
assert confirmed_start(bonus) is None
assert penalty_start(bonus, date(2026, 1, 1)) is None
unconfirmed = replace(READING, confirmed_on=None)
assert confirmed_start(unconfirmed) == READING.penalty_from
assert penalty_start(unconfirmed, date(2026, 1, 1)) == READING.penalty_from
assert penalty_start(unconfirmed, None) is None
early = replace(READING, confirmed_on=date(2026, 9, 1))
assert confirmed_start(early) == READING.penalty_from
assert penalty_start(early, None) == READING.penalty_from
def test_a_pure_bonus_is_never_penalised_with_a_map() -> None:
bonus = replace(READING, penalty_from=None)
base = base_for(date(2026, 10, 3), (bonus,), first_credits={"reading": None})
assert base.gaming_minutes == 300

View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "earned-time"
version = "0.5.0"
version = "0.6.0"
description = "Penalty-then-reward registry: what every gate app earns in gaming time and shutdown time"
requires-python = ">=3.14"
dependencies = []
@ -107,7 +107,7 @@ disable = []
min-public-methods = 0
max-module-lines = 1000
# Earner is a frozen registry record: its fields are the policy (cf. music_theory).
max-attributes = 12
max-attributes = 13
[tool.pylint.typecheck]
generated-members = [