kuhyx/utils
1
0
Fork 0
mirror of https://github.com/kuhyx/utils.git synced 2026-11-10 19:07:17 +01:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity

sickdays 0.1.0 + earned_time 0.11.0: uncapped sick-day pool, 20:00 sick shutdown floor

Browse Source 
sickdays: per-date pool, no cap or reason, add/remove today or later only,
import_legacy for the migration. LWW state flip per date (no tombstones),
Firebase RTDB sickdays-sync/devices/<uuid>/sick_days.json, HMAC audit trail.
Readers fail closed (unreadable pool = not sick, logged); writers refuse a
corrupt pool. Free-day cross-check via a lazily imported freedays predicate.

earned_time: resolve(..., sick=True) floors the final shutdown at 20:00
(capped at the ceiling); gaming, terms and base unchanged. Result dataclasses
moved to _resolution.py for the line cap.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Xfj1sxKH9VBPqpcAni8Jry
This commit is contained in:
Krzysztof Rudnicki 2026-10-16 08:23:00 +02:00
parent ba34b603a1
commit 894d76f811
44 changed files with 3310 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
.github/workflows/sickdays-precommit.yml vendored Normal file
View File

@ -0,0 +1,73 @@
name: sickdays pre-commit
on:
push:
branches: [main]
# The workflow's own file counts as sickdays: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['sickdays/**', '.github/workflows/sickdays-precommit.yml']
pull_request:
branches: [main]
paths: ['sickdays/**', '.github/workflows/sickdays-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==' sickdays/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-sickdays-${{ env.pythonLocation }}-${{
hashFiles('sickdays/.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-sickdays-${{ 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-sickdays-${{ env.pythonLocation }}-${{
hashFiles('sickdays/.pre-commit-config.yaml') }}
restore-keys: mypy-sickdays-${{ env.pythonLocation }}-
- name: Run the hooks
run: >-
pre-commit run --config sickdays/.pre-commit-config.yaml
--all-files --show-diff-on-failure

44
.github/workflows/sickdays-tests.yml vendored Normal file
View File

@ -0,0 +1,44 @@
name: sickdays tests
on:
push:
branches: [main]
# The workflow's own file counts as sickdays: editing only this file would
# otherwise trigger nothing, so a fix to the job itself could never prove
# itself.
paths: ['sickdays/**', '.github/workflows/sickdays-tests.yml']
pull_request:
branches: [main]
paths: ['sickdays/**', '.github/workflows/sickdays-tests.yml']
jobs:
test:
runs-on: ubuntu-latest
defaults:
run:
working-directory: sickdays
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.x"
cache: pip
cache-dependency-path: sickdays/requirements.txt
# crdt-sync and freedays are siblings in this repo, so they install from
# the checkout rather than from a tag -- CI must test what is on this
# branch, not what was last released. Deliberately NOT gatelock:
# importing it pulls in tkinter, and this job has no display precisely
# because sickdays must work without one.
- name: Install the sibling packages
run: pip install ../crdt-sync && pip install --no-deps ../freedays
- name: Install dependencies
run: pip install -r requirements.txt && pip install --no-deps -e .
# No display needed: sickdays draws nothing. It is policy, a JSON file
# and a CLI.
- name: Run tests with coverage
run: python -m pytest -v

2
CLAUDE.md
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, earned_time, music_theory, ...).
repo_contract, coverage_gaps, crdt-sync, gatelock, freedays, sickdays, 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

32
earned_time/README.md
View File

@ -48,6 +48,13 @@ never a same-day cut):
The two floors are derived separately (`_resolve._shutdown_floor`), so the
ladder cannot move a gaming minute.
**Sick days (0.11.0, `_sick.py`).** `resolve(..., sick=True)` floors the
*final* shutdown at 20:00 (`SICK_SHUTDOWN_FLOOR_MINUTES`), capped at the
ceiling: 18:50 becomes 20:00, 21:30 stays 21:30 (nothing stacks on the floor).
Base, terms, gaming, waivers and penalties are unchanged; earners still earn.
`sick` is the consumer's answer (`sickdays.is_sick_day`, never imported here).
Live credits are paid out of `Resolution.sick_lift_minutes` first (a grace lift).
**The tutor cutover (`TUTOR_FROM`, one constant in `_ladder.py`).** The
registry is per day: `earners_for(day)` is `EARNERS` before it and
`TUTOR_EARNERS` from it. From `TUTOR_FROM` Anki is retired (neither penalised
@ -100,10 +107,9 @@ per-matcher reader `_match.ROW_UNITS`):
- Rollout order: no minute row may be written until every consumer process
runs 0.8.0. A 0.7.0 reader pays each row as a 15-minute block.
Days before `TUTOR_FROM` resolve on
`EARNERS` exactly as before. Consumers must iterate `earners_for(day)`, never
`EARNERS`; `resolve`/`base_for` default to it, and ignore answers for an
earner of another day's registry.
Days before `TUTOR_FROM` resolve on `EARNERS` exactly as before. Consumers must
iterate `earners_for(day)`, never `EARNERS`; `resolve`/`base_for` default to
it, and ignore answers for an earner of another day's registry.
**The gap waiver (`ANKI_WAIVED_FROM`, 2026-10-09).** anki-guard never wrote a
credit, and the tutor is not confirmed yet, so from `ANKI_WAIVED_FROM` until
@ -152,7 +158,8 @@ How far to trust a gate is computed from its ledger, never listed by name:
## API
- `resolve(answers, day=None, earners=EARNERS) -> Resolution` — the one sum.
- `resolve(answers, day=None, earners=None, *, first_credits=None, sick=False)`
`-> Resolution` — the one sum. `sick_shutdown_minutes(minutes, day) -> int`.
- `base_for(day=None, earners=EARNERS) -> Base` — the day's floors.
- `done_today(earner, ledger, key_file, *, now=None) -> bool | None`
- `first_credit_at(earner, ledger, key_file, day) -> float | None` — unix time
@ -168,9 +175,8 @@ How far to trust a gate is computed from its ledger, never listed by name:
on a day, and every earner of every registry (what a ledger watcher must
watch). They read `earned_time.EARNERS` through the package, so a consumer
test that patches it still registers its stand-in.
- `shutdown_minutes_for(earner, day) -> int`,
`extra_shutdown_minutes_for(earner, day) -> int`,
`shutdown_ceiling_for(day) -> int`, `on_ladder(day) -> bool` — what a status
- `shutdown_minutes_for(earner, day) -> int`, `extra_shutdown_minutes_for(earner,
day) -> int`, `shutdown_ceiling_for(day) -> int`, `on_ladder(day) -> bool` — what a status
view needs to list what is still left to earn. `Earner.shutdown_for(units,
day=None)` is the day-aware sum (default today). Never read
`earner.shutdown_minutes` directly: it is the pre-ladder value.
@ -237,14 +243,8 @@ 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.10.0#subdirectory=earned_time"
pip install --user --break-system-packages --no-deps "earned-time @ git+https://github.com/kuhyx/utils@earned-time-v0.11.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
```
from the same user site-packages. Test: `scripts/run_subproject_tests.sh earned_time`.

7
earned_time/earned_time/__init__.py
View File

@ -19,9 +19,11 @@ A consumer's whole integration::
if e.ledger is not None
}
answers["workout"] = my_workout_count()
day = earned_time.resolve(answers)
day = earned_time.resolve(answers, sick=sickdays.is_sick_day())
apply(day.gaming_minutes)
``sick`` (0.11.0) only raises shutdown to at least 20:00; gaming is unchanged.
How far to trust a gate is computed, not listed: :func:`maturity` (and
``python -m earned_time maturity``).
@ -80,6 +82,7 @@ from earned_time._policy import (
from earned_time._quota import TUTOR_QUOTA_FROM, TUTOR_QUOTA_MINUTES, tutor_quota_for
from earned_time._registry import all_earners, earners_for, registries
from earned_time._resolve import Base, Resolution, Term, base_for, resolve
from earned_time._sick import SICK_SHUTDOWN_FLOOR_MINUTES, sick_shutdown_minutes
__all__ = [
"ANKI",
@ -96,6 +99,7 @@ __all__ = [
"READING",
"SHUTDOWN_BASE_MINUTES",
"SHUTDOWN_CEILING_MINUTES",
"SICK_SHUTDOWN_FLOOR_MINUTES",
"TUTOR",
"TUTOR_EARNERS",
"TUTOR_FROM",
@ -131,6 +135,7 @@ __all__ = [
"resolve",
"shutdown_ceiling_for",
"shutdown_minutes_for",
"sick_shutdown_minutes",
"today_window",
"tutor_quota_for",
"verified",

84
earned_time/earned_time/_resolution.py Normal file
View File

@ -0,0 +1,84 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""What :func:`earned_time.resolve` returns: the base, each term, the sum.
Split from :mod:`earned_time._resolve` (0.11.0) to keep both under the line
cap; ``earned_time._resolve`` still re-exports all three.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from datetime import date
from earned_time._policy import Earner
@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;
on a sick day raised to the sick floor (:mod:`earned_time._sick`).
sick: Whether the day was resolved as a sick day (0.11.0).
sick_lift_minutes: How far the sick floor raised ``shutdown_minutes``
above base plus terms (``0`` unless sick): a consumer adding live
credits on top pays them out of this first, as with a grace lift.
"""
day: date
base: Base
terms: tuple[Term, ...]
gaming_minutes: int
shutdown_minutes: int
sick: bool = False
sick_lift_minutes: int = 0
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)

78
earned_time/earned_time/_resolve.py
View File

@ -21,7 +21,6 @@ whose penalty has not started does not lower it (0.6.1).
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, date, datetime
import logging
from typing import TYPE_CHECKING, Final
@ -35,6 +34,8 @@ from earned_time._policy import (
Earner,
)
from earned_time._registry import all_earners, earners_for
from earned_time._resolution import Base, Resolution, Term
from earned_time._sick import sick_shutdown_minutes
if TYPE_CHECKING:
from collections.abc import Mapping
@ -44,67 +45,6 @@ _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()
@ -194,6 +134,7 @@ def resolve(
earners: tuple[Earner, ...] | None = None,
*,
first_credits: Mapping[str, date | None] | None = None,
sick: bool = False,
) -> Resolution:
"""Sum the base and every earned term, each capped at its ceiling.
@ -209,6 +150,9 @@ def resolve(
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``.
sick: The day is a sick day (the consumer asks ``sickdays``): shutdown
is at least 20:00 (:func:`~earned_time._sick.sick_shutdown_minutes`).
Gaming and every term are unchanged.
Returns:
The resolution, with a term for every registered earner.
@ -238,11 +182,17 @@ def resolve(
)
)
gaming = base.gaming_minutes + sum(t.gaming_minutes for t in terms)
shutdown = base.shutdown_minutes + sum(t.shutdown_minutes for t in terms)
earned = min(
shutdown_ceiling_for(target),
base.shutdown_minutes + sum(t.shutdown_minutes for t in terms),
)
shutdown = sick_shutdown_minutes(earned, target) if sick else earned
return Resolution(
day=target,
base=base,
terms=tuple(terms),
gaming_minutes=min(GAMING_CEILING_MINUTES, gaming),
shutdown_minutes=min(shutdown_ceiling_for(target), shutdown),
shutdown_minutes=shutdown,
sick=sick,
sick_lift_minutes=shutdown - earned,
)

39
earned_time/earned_time/_sick.py Normal file
View File

@ -0,0 +1,39 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The sick-day shutdown floor (0.11.0).
On a sick day shutdown is never earlier than 20:00: the resolved shutdown is
``max(SICK_SHUTDOWN_FLOOR_MINUTES, the normal ladder result)``, still capped
at the day's ceiling. It is applied to the *final* sum, never to the base --
raising the base would stack every earner's minutes on top of 20:00.
Only shutdown moves. Gaming time, the registry, waivers and penalties are
untouched, and every earner still earns on a sick day.
Whether a day is sick is the consumer's answer (the ``sickdays`` pool):
this package is stdlib-only and never imports it.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Final
from earned_time._ladder import shutdown_ceiling_for
if TYPE_CHECKING:
from datetime import date
# 20:00, minutes after local midnight.
SICK_SHUTDOWN_FLOOR_MINUTES: Final = 20 * 60
def sick_shutdown_minutes(minutes: int, day: date) -> int:
"""``minutes`` (a resolved shutdown) with the sick-day floor applied.
Args:
minutes: The normal resolved shutdown, minutes after midnight.
day: The day, which picks the ceiling.
Returns:
``min(ceiling, max(SICK_SHUTDOWN_FLOOR_MINUTES, minutes))``.
"""
return min(shutdown_ceiling_for(day), max(SICK_SHUTDOWN_FLOOR_MINUTES, minutes))

89
earned_time/earned_time/tests/test_sick.py Normal file
View File

@ -0,0 +1,89 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The sick-day shutdown floor (0.11.0): 20:00 at least, gaming untouched."""
from __future__ import annotations
from datetime import date
import pytest
from earned_time import (
SICK_SHUTDOWN_FLOOR_MINUTES,
resolve,
shutdown_ceiling_for,
sick_shutdown_minutes,
)
LADDER_DAY = date(2026, 10, 16)
PRE_LADDER_DAY = date(2026, 10, 4)
NOTHING: dict[str, int | bool | None] = {
"workout": 0,
"leetcode": False,
"reading": False,
"tutor": 0,
}
SOME: dict[str, int | bool | None] = {**NOTHING, "workout": 1, "leetcode": True}
EVERYTHING: dict[str, int | bool | None] = {
"workout": 1,
"leetcode": True,
"reading": True,
"tutor": 60,
}
def test_the_floor_is_eight_pm() -> None:
assert SICK_SHUTDOWN_FLOOR_MINUTES == 20 * 60
def test_a_sick_day_that_earned_nothing_shuts_down_at_eight() -> None:
normal = resolve(NOTHING, LADDER_DAY)
sick = resolve(NOTHING, LADDER_DAY, sick=True)
assert normal.shutdown_minutes == 18 * 60 + 50
assert sick.shutdown_minutes == 20 * 60
assert sick.sick
assert sick.sick_lift_minutes == 70
assert not normal.sick
assert normal.sick_lift_minutes == 0
def test_earning_past_the_floor_is_not_stacked_on_it() -> None:
"""max(20:00, earned) -- never 20:00 plus what was earned."""
normal = resolve(SOME, LADDER_DAY)
sick = resolve(SOME, LADDER_DAY, sick=True)
assert normal.shutdown_minutes > 20 * 60
assert sick.shutdown_minutes == normal.shutdown_minutes
assert sick.sick_lift_minutes == 0
def test_a_fully_done_sick_day_still_stops_at_the_ceiling() -> None:
sick = resolve(EVERYTHING, LADDER_DAY, sick=True)
assert sick.shutdown_minutes == shutdown_ceiling_for(LADDER_DAY) == 23 * 60
def test_the_floor_applies_before_the_ladder_too() -> None:
answers = {"workout": 0, "leetcode": False, "reading": False}
normal = resolve(answers, PRE_LADDER_DAY)
sick = resolve(answers, PRE_LADDER_DAY, sick=True)
assert normal.shutdown_minutes == 19 * 60
assert sick.shutdown_minutes == 20 * 60
@pytest.mark.parametrize("answers", [NOTHING, SOME, EVERYTHING])
def test_gaming_and_every_term_are_unchanged(
answers: dict[str, int | bool | None],
) -> None:
normal = resolve(answers, LADDER_DAY)
sick = resolve(answers, LADDER_DAY, sick=True)
assert sick.gaming_minutes == normal.gaming_minutes
assert sick.terms == normal.terms
assert sick.base == normal.base
def test_the_helper_is_capped_at_the_ceiling(monkeypatch: pytest.MonkeyPatch) -> None:
assert sick_shutdown_minutes(18 * 60, LADDER_DAY) == 20 * 60
assert sick_shutdown_minutes(21 * 60, LADDER_DAY) == 21 * 60
monkeypatch.setattr(
"earned_time._sick.shutdown_ceiling_for", lambda _day: 19 * 60 + 30
)
assert sick_shutdown_minutes(18 * 60, LADDER_DAY) == 19 * 60 + 30

2
earned_time/pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "earned-time"
version = "0.10.0"
version = "0.11.0"
description = "Penalty-then-reward registry: what every gate app earns in gaming time and shutdown time"
requires-python = ">=3.14"
dependencies = []

146
sickdays/.pre-commit-config.yaml Normal file
View File

@ -0,0 +1,146 @@
# Pre-commit Configuration for sickdays
# 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 sickdays/.pre-commit-config.yaml --all-files
files: ^sickdays/
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 sickdays
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.17.0
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=sickdays/pyproject.toml
- --fail-under=10
- --jobs=4
additional_dependencies:
- pytest
# Kept in step with pyproject.toml, so pylint resolves the crdt_sync
# imports instead of scoring every one of them as import-error.
- crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync
- repo: https://github.com/PyCQA/bandit
rev: 1.9.4
hooks:
- id: bandit
args:
- -c
- sickdays/pyproject.toml
- --severity-level=high
- --confidence-level=medium
- --skip=B113
additional_dependencies: ["bandit[toml]"]
exclude: ^(sickdays/sickdays/tests/|.*test.*\.py$)
- repo: local
hooks:
- id: pytest-coverage
name: pytest with coverage enforcement
entry: scripts/run_subproject_tests.sh sickdays
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

157
sickdays/README.md Normal file
View File

@ -0,0 +1,157 @@
# sickdays
One shared pool of **sick days** that every gate respects.
Every day is a normal day until it is marked sick. There is no cap, no
budget, no reason to give and no countdown. The one rule is the calendar:
only today or a later day can be marked or un-marked — never a past one.
This replaces screen-locker's capped sick day (rolling windows, 120-character
justifications). It is **not** the free-day pool: `freedays` is a separate
pool with its own file and remote node, and a date is never in both.
## For a gate app
```python
import sickdays
if sickdays.is_sick_day(today): # default: today
... # apply this gate's sick-day rules
```
`is_sick_day()` reads one local JSON file, never the network, and **never
raises**. A missing file means "no sick days". An unreadable or corrupt file
is logged as a warning and also reads as *not sick*: it fails closed, so the
alarm rings and the locks apply.
## For a person
```console
$ sickdays add # today
$ sickdays add 2026-10-20 # or 'tomorrow'
$ sickdays remove 2026-10-20 # today or later only
$ sickdays list # every sick day, oldest first
$ sickdays check tomorrow # exit 0 if sick, 1 if not -- for scripts
$ sickdays sync # pull and push now
```
`python -m sickdays ...` is the same command. Exit codes: `0` done, `1`
refused (past day, already sick, not sick, a free day, sync unavailable),
`2` input not understood.
## API
| call | does |
|---|---|
| `is_sick_day(day=None, *, log_path=None, now=None) -> bool` | the gate question; never raises |
| `sick_days(*, log_path=None) -> frozenset[date]` | every date currently sick |
| `add_sick_day(day, *, is_free=None, paths=None, now=None) -> SickDay` | today or later; refuses a free day |
| `remove_sick_day(day, *, paths=None, now=None) -> SickDay` | today or later |
| `import_legacy(days, *, paths=None) -> list[date]` | migration only: may write past dates; idempotent |
| `ensure_not_sick(day, *, paths=None)` | for freedays' callers, before `freedays.mark` |
| `sync()`, `sync_quietly()` | one pull/merge/push tick |
Refusals subclass `SickDayError`: `PastDateError`, `AlreadySickError`,
`NotSickError`, `FreeDayConflictError`, `SickDayConflictError`,
`ConflictCheckUnavailableError`, `UnreadablePoolError`. Writers refuse an
existing-but-corrupt pool (`UnreadablePoolError`) instead of overwriting it
with an empty one; `sync` reads it as empty, so a peer's copy restores it.
`import_legacy` is for screen-locker's `sick_history.json` (2026-06-21 and
2026-07-24, both before every gate start). It skips the free-day check.
## A date is never in both pools
Neither package imports the other at module level, so they can never form an
import cycle:
- **Adding a sick day** checks a predicate `is_free(day) -> bool`. By default
`add_sick_day` imports `freedays` *lazily*, by name, at call time
(`_conflict.free_day_predicate`). If freedays is not importable the add is
**refused** (`ConflictCheckUnavailableError`). The invariant is hard, so
"could not check" never counts as "not free". A phone port or a test passes
its own `is_free`.
- **Marking a free day** is freedays' write. Its callers (CLIs, phone, lock
screen) call `sickdays.ensure_not_sick(day)` first. That needs no freedays
import at all. freedays itself does not check yet.
## Storage
| what | where |
|---|---|
| the pool | `~/.local/share/sickdays/sick_days.json` |
| device uuid (per install) | `~/.local/share/sickdays/device_id` |
| audit trail (HMAC-signed, append-only) | `~/.local/share/sickdays/audit.jsonl` + `hmac.key` |
| sync bookkeeping (local only) | `~/.local/share/sickdays/sync_state.json` |
Days are local calendar dates (`_day.today()`), the same as freedays.
## Sync and the wire format (for the phone)
Firebase RTDB only, shared credential `~/.config/crdt-sync/` (crdt-sync's
`firebase_client_for("sickdays")`). One directory per device; each device
writes only its own node:
| RTDB node | value |
|---|---|
| `sickdays-sync/devices/<uuid>/sick_days~2Ejson` | a JSON **string** holding the log (below) |
| `sickdays-sync/revs/<uuid>` | hex SHA-256 of that exact string |
The logical path is `sickdays-sync/devices/<uuid>/sick_days.json`. crdt-sync
escapes the `.` in keys as `~2E`. Write the log first, then the rev; a peer
skips any device whose rev it has already merged. The `<uuid>` is a
persisted per-install uuid4, never `pc`/`phone`.
The log is a JSON object keyed by ISO date. Each value is a crdt-sync record:
```json
{
"2026-10-20": {
"id": "2026-10-20",
"fields": {
"state": ["sick", "2026-10-16T08:12:03.517Z-0000-<uuid>"],
"actor": ["<uuid>", "2026-10-16T08:12:03.517Z-0000-<uuid>"]
},
"deleted": false,
"deleted_hlc": null
}
}
```
- `id` equals the key: `YYYY-MM-DD`, local date.
- `state` is `"sick"` or `"cleared"`. Anything else, or a missing record,
means a normal day. `actor` is the uuid of the device that last wrote it.
- Each field is `[value, hlc]`, and every field of one write carries the
**same** HLC. The HLC string is
`YYYY-MM-DDTHH:MM:SS.mmmZ-<counter, 4 hex>-<node uuid>` (UTC wall time). It
orders as the tuple `(wall_ms, counter, node_id)`. A new tick is
`max(now_ms, last_own.wall_ms)`, with the counter bumped when the wall time
did not move.
- **Merge = per-field last-writer-wins**: for each date, keep the field with
the greater HLC. So the latest add *or remove* of a date wins on every
device, and a date can be added and removed any number of times.
- **Remove = write `state: "cleared"`. Never write `deleted: true`.**
crdt-sync OR-merges tombstones, so a deleted date could never be marked
sick again. A tombstoned record reads as "not sick".
Don't hand-roll this on Dart: use `crdt_sync_dart` (`syncLog`, `logToJson`
/ `logFromJson`), exactly as wake-alarm's `free_day_sync.dart` does for
`freedays-sync`. Only the prefix (`sickdays-sync/devices`), the filename
(`sick_days.json`) and the `"sick"` state value differ. There is no
`consumed` or `reason` field.
## Install
```console
$ ./install.sh # own venv (crdt-sync + freedays + sickdays), CLI on PATH, 15-min sync timer
$ ./install.sh --no-timer
```
## Development
```console
$ pip install ../crdt-sync && pip install --no-deps ../freedays
$ pip install -r requirements.txt && pip install --no-deps -e .
$ python -m pytest # 100% branch coverage is enforced
$ pre-commit run --config sickdays/.pre-commit-config.yaml --all-files # from the repo root
```

119
sickdays/install.sh Executable file
View File

@ -0,0 +1,119 @@
#!/bin/bash
# ============================================================================
# Install the sickdays CLI and its sync timer for the current user.
#
# sickdays gets its OWN venv, like freedays: the pool is shared by every gate,
# so the command that manages it must not depend on whichever app venv
# happened to be active. freedays goes into the same venv because adding a
# sick day checks the free-day pool first (a date is never in both).
# Idempotent: rerun it after any change.
# ============================================================================
set -euo pipefail
SCRIPT_NAME="$(basename "$0")"
REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
UTILS_DIR="$(dirname "$REPO_DIR")"
readonly SCRIPT_NAME REPO_DIR UTILS_DIR
readonly STATE_DIR="${HOME}/.local/share/sickdays"
readonly VENV_DIR="${STATE_DIR}/venv"
readonly BIN_DIR="${HOME}/.local/bin"
readonly UNIT_DIR="${HOME}/.config/systemd/user"
readonly SYSTEM_PYTHON="/usr/bin/python3"
log() { printf 'install: %s\n' "$1" >&2; }
fail() { printf 'install: FAILED -- %s\n' "$1" >&2; exit 1; }
usage() {
echo "Usage: $SCRIPT_NAME [--no-timer]"
echo " --no-timer install the CLI only; do not enable the sync timer"
exit 0
}
INSTALL_TIMER=1
build_venv() {
log "building the sickdays venv at $VENV_DIR"
mkdir -p "$STATE_DIR"
if [[ ! -x "$VENV_DIR/bin/python" ]]; then
"$SYSTEM_PYTHON" -m venv "$VENV_DIR" || fail "could not create the venv"
fi
# Siblings install from the working tree, not from a tag: this script runs
# from a checkout, and installing the released tag here would silently
# test something other than what is on disk.
"$VENV_DIR/bin/pip" install --quiet --upgrade pip || fail "pip self-upgrade"
"$VENV_DIR/bin/pip" install --quiet "$UTILS_DIR/crdt-sync" \
|| fail "the crdt-sync sibling package"
"$VENV_DIR/bin/pip" install --quiet --no-deps "$UTILS_DIR/freedays" \
|| fail "the freedays sibling package"
"$VENV_DIR/bin/pip" install --quiet --no-deps -e "$REPO_DIR" \
|| fail "sickdays itself"
}
link_cli() {
log "putting the sickdays command on PATH at $BIN_DIR/sickdays"
mkdir -p "$BIN_DIR"
ln -sf "$VENV_DIR/bin/sickdays" "$BIN_DIR/sickdays"
}
install_timer() {
log "installing the sync timer"
mkdir -p "$UNIT_DIR"
install -m 644 "$REPO_DIR/systemd/sickdays-sync.service" "$UNIT_DIR/"
install -m 644 "$REPO_DIR/systemd/sickdays-sync.timer" "$UNIT_DIR/"
systemctl --user daemon-reload || fail "systemctl daemon-reload"
systemctl --user enable --now sickdays-sync.timer || fail "enabling the timer"
}
verify() {
# Two environments, either of which can be broken on its own: the CLI's
# venv, and the system python the gate apps import the library from.
log "verifying"
"$BIN_DIR/sickdays" list >/dev/null || fail "the sickdays CLI does not run"
"$VENV_DIR/bin/python" -c "import freedays" \
|| fail "freedays is missing from the venv; adding a sick day would refuse"
"$SYSTEM_PYTHON" -c "import sickdays" 2>/dev/null || {
log "NOTE: the system python cannot import sickdays yet."
log " The gate apps need it there; their own install.sh does that."
}
if [[ "$INSTALL_TIMER" -eq 1 ]]; then
systemctl --user is-enabled sickdays-sync.timer >/dev/null \
|| fail "the sync timer is not enabled"
fi
}
main() {
build_venv
link_cli
if [[ "$INSTALL_TIMER" -eq 1 ]]; then
install_timer
fi
verify
echo "============================================================"
"$BIN_DIR/sickdays" list
echo "============================================================"
log "done. 'sickdays add' marks today sick, everywhere."
if [[ ":$PATH:" != *":$BIN_DIR:"* ]]; then
log "NOTE: $BIN_DIR is not on your PATH; add it to use 'sickdays' bare."
fi
}
while [[ $# -gt 0 ]]; do
case $1 in
--no-timer)
INSTALL_TIMER=0
shift
;;
-h|--help)
usage
;;
*)
echo "Unknown option: $1" >&2
exit 1
;;
esac
done
main "$@"

177
sickdays/pyproject.toml Normal file
View File

@ -0,0 +1,177 @@
[build-system]
requires = ["setuptools>=61"]
build-backend = "setuptools.build_meta"
[project]
name = "sickdays"
version = "0.1.0"
description = "Shared sick-day pool: per-date, uncapped, synced; every gate respects it"
requires-python = ">=3.14"
# Deliberately NOT gatelock (its __init__ imports tkinter); the HMAC lives in
# sickdays/_hmac.py, as in freedays. freedays is a runtime dependency of the
# cross-pool check only, and is imported lazily by name (sickdays/_conflict.py)
# so the two packages can never form an import cycle.
dependencies = [
"crdt-sync @ git+https://github.com/kuhyx/utils@crdt-sync-v0.9.2#subdirectory=crdt-sync",
"freedays @ git+https://github.com/kuhyx/utils@freedays-v0.3.1#subdirectory=freedays",
]
[project.scripts]
sickdays = "sickdays._cli:main"
[tool.setuptools.packages.find]
include = ["sickdays*"]
[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 = ["sickdays"]
[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, 'sickdays')"
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 = ["sickdays/tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"-v",
"--strict-markers",
"--strict-config",
"-ra",
"--cov=sickdays",
"--cov-branch",
"--cov-report=term-missing",
"--cov-report=lcov",
]
filterwarnings = [
"error",
"ignore::DeprecationWarning",
"default::pytest.PytestUnraisableExceptionWarning",
]
[tool.coverage.run]
source = ["sickdays"]
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"]

10
sickdays/requirements.txt Normal file
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.7.0
pylint==4.1.3
pytest==9.1.1
pytest-cov==7.1.0
ruff==0.17.1

82
sickdays/sickdays/__init__.py Normal file
View File

@ -0,0 +1,82 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""One shared pool of sick days that every gate respects.
A gate's whole integration::
import sickdays
if sickdays.is_sick_day(today):
... # the sick-day rules for this gate apply
It reads one local file, never the network, and never raises: an unreadable
pool is logged and reads as "not sick", so the failure mode is "the normal
rules apply", never "everything is switched off".
Sick days are not free days. They are a separate pool, with a separate file
and a separate remote node, and a date is never in both
(:mod:`sickdays._conflict`). There is no cap and no justification: every day
is normal by default, and any day from today on can be marked or un-marked.
"""
from __future__ import annotations
from sickdays._api import (
actor_id,
add_sick_day,
import_legacy,
is_sick_day,
load,
remove_sick_day,
sick_days,
)
from sickdays._conflict import (
FreePredicate,
ensure_not_free,
ensure_not_sick,
free_day_predicate,
)
from sickdays._day import parse_iso, resolve, to_iso, today
from sickdays._errors import (
AlreadySickError,
ConflictCheckUnavailableError,
FreeDayConflictError,
NotSickError,
PastDateError,
SickDayConflictError,
SickDayError,
UnreadablePoolError,
)
from sickdays._model import SickDay
from sickdays._paths import Paths
from sickdays._sync import SyncUnavailableError, sync, sync_quietly
__all__ = [
"AlreadySickError",
"ConflictCheckUnavailableError",
"FreeDayConflictError",
"FreePredicate",
"NotSickError",
"PastDateError",
"Paths",
"SickDay",
"SickDayConflictError",
"SickDayError",
"SyncUnavailableError",
"UnreadablePoolError",
"actor_id",
"add_sick_day",
"ensure_not_free",
"ensure_not_sick",
"free_day_predicate",
"import_legacy",
"is_sick_day",
"load",
"parse_iso",
"remove_sick_day",
"resolve",
"sick_days",
"sync",
"sync_quietly",
"to_iso",
"today",
]

10
sickdays/sickdays/__main__.py Normal file
View File

@ -0,0 +1,10 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""``python -m sickdays``: the same command as the ``sickdays`` script."""
from __future__ import annotations
import sys
from sickdays._cli import main
sys.exit(main())

178
sickdays/sickdays/_api.py Normal file
View File

@ -0,0 +1,178 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Ask, add, remove, import -- the whole public surface.
:func:`is_sick_day` is the one every gate calls, and it is deliberately the
cheapest thing here: one local file read, no network, no lock. It never
raises: an unreadable pool is logged and reads as "not sick", so the failure
mode is "the gates apply", never "everything is switched off".
There is no cap, no budget and no justification. Every day is normal until
someone marks it sick, and only today or a later day can be marked or
un-marked -- never a past one.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from crdt_sync import Hlc, load_device_identity
from sickdays._audit import record_event
from sickdays._conflict import ensure_not_free, free_day_predicate
from sickdays._day import to_iso, today
from sickdays._errors import AlreadySickError, NotSickError, PastDateError
from sickdays._model import (
SickDay,
build_record,
is_sick_record,
latest_hlc,
sick_dates,
)
from sickdays._paths import Paths, resolve_paths
from sickdays._store import read_pool, read_pool_strict, write_pool
if TYPE_CHECKING:
from collections.abc import Iterable
from datetime import date
from pathlib import Path
from crdt_sync import Log
from sickdays._conflict import FreePredicate
def load(log_path: Path | None = None) -> Log:
"""Read the pool. Missing reads as empty; unreadable is logged, then empty."""
return read_pool(log_path if log_path is not None else resolve_paths(None).log)
def actor_id(paths: Paths | None = None) -> str:
"""This install's persisted device uuid, created on first call."""
return load_device_identity(resolve_paths(paths).device_id).device_id
def is_sick_day(
day: date | None = None,
*,
log_path: Path | None = None,
now: date | None = None,
) -> bool:
"""Whether ``day`` (default: today) is a sick day. Never raises."""
target = day if day is not None else today(now)
record = load(log_path).get(to_iso(target))
return record is not None and is_sick_record(record)
def sick_days(*, log_path: Path | None = None) -> frozenset[date]:
"""Every date currently marked sick, past and future."""
return sick_dates(load(log_path))
def _refuse_past(day: date, reference: date, verb: str) -> None:
if day < reference:
msg = (
f"{to_iso(day)} has already passed -- sick days can only be "
f"{verb} for today or later"
)
raise PastDateError(msg)
def _write(log: Log, entry: SickDay, resolved: Paths) -> None:
"""Stamp ``entry`` with this node's next tick and persist the pool."""
tick = Hlc.new_tick(entry.actor, latest_hlc(log, entry.actor))
log[to_iso(entry.day)] = build_record(entry, tick)
write_pool(resolved.log, log)
def add_sick_day(
day: date,
*,
is_free: FreePredicate | None = None,
paths: Paths | None = None,
now: date | None = None,
) -> SickDay:
"""Mark ``day`` sick, everywhere, for every gate.
Args:
day: The day. Today or later.
is_free: "Is this a free day?"; ``None`` asks freedays, imported
lazily (:func:`sickdays._conflict.free_day_predicate`).
paths: Where the pool, uuid and audit trail live.
now: Reference date, for tests.
Returns:
The stored entry.
Raises:
PastDateError: ``day`` has already passed.
AlreadySickError: ``day`` is already a sick day.
FreeDayConflictError: ``day`` is a free day.
ConflictCheckUnavailableError: The free-day pool cannot be checked.
UnreadablePoolError: The local pool exists but cannot be parsed.
"""
resolved = resolve_paths(paths)
_refuse_past(day, today(now), "added")
log = read_pool_strict(resolved.log)
existing = log.get(to_iso(day))
if existing is not None and is_sick_record(existing):
msg = f"{to_iso(day)} is already a sick day"
raise AlreadySickError(msg)
ensure_not_free(day, is_free if is_free is not None else free_day_predicate())
entry = SickDay(day=day, is_sick=True, actor=actor_id(resolved))
_write(log, entry, resolved)
record_event("add", day, actor=entry.actor, paths=resolved)
return entry
def remove_sick_day(
day: date,
*,
paths: Paths | None = None,
now: date | None = None,
) -> SickDay:
"""Make ``day`` a normal day again. Today or later only.
Raises:
PastDateError: ``day`` has already passed.
NotSickError: ``day`` is not a sick day.
UnreadablePoolError: The local pool exists but cannot be parsed.
"""
resolved = resolve_paths(paths)
_refuse_past(day, today(now), "removed")
log = read_pool_strict(resolved.log)
existing = log.get(to_iso(day))
if existing is None or not is_sick_record(existing):
msg = f"{to_iso(day)} is not a sick day"
raise NotSickError(msg)
entry = SickDay(day=day, is_sick=False, actor=actor_id(resolved))
_write(log, entry, resolved)
record_event("remove", day, actor=entry.actor, paths=resolved)
return entry
def import_legacy(days: Iterable[date], *, paths: Paths | None = None) -> list[date]:
"""Migration only: record ``days`` as sick, past dates included.
The one writer that may write the past -- used once, for screen-locker's
``sick_history.json`` (2026-06-21, 2026-07-24), both of which predate
every gate start. No free-day check: those dates predate freedays.
Idempotent: a day already sick is skipped and gets no audit line.
Returns:
The days actually written, ascending.
Raises:
UnreadablePoolError: The local pool exists but cannot be parsed.
"""
resolved = resolve_paths(paths)
log = read_pool_strict(resolved.log)
node = actor_id(resolved)
written: list[date] = []
for day in sorted(set(days)):
existing = log.get(to_iso(day))
if existing is not None and is_sick_record(existing):
continue
_write(log, SickDay(day=day, is_sick=True, actor=node), resolved)
record_event("import", day, actor=node, paths=resolved)
written.append(day)
return written

66
sickdays/sickdays/_audit.py Normal file
View File

@ -0,0 +1,66 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Append-only, HMAC-signed trail of every add, remove and legacy import.
Nothing reads this to make a decision -- the CRDT log is the truth. It
exists so "when did I mark that day, and from which device" is answerable
later, and so a hand-edited line is detectable rather than invisible.
Signing is best-effort, as in freedays: with no key the line is written
unsigned, because an unsigned audit line is a much smaller problem than a
sick day that could not be taken because a key file was missing.
"""
from __future__ import annotations
from datetime import UTC, datetime
import json
import logging
from typing import TYPE_CHECKING, Any
from sickdays._day import to_iso
from sickdays._hmac import generate_key, sign
from sickdays._paths import Paths, resolve_paths
if TYPE_CHECKING:
from datetime import date
_logger = logging.getLogger(__name__)
def record_event(
action: str,
day: date,
*,
actor: str,
paths: Paths | None = None,
) -> None:
"""Append one signed line describing a change to the pool.
Args:
action: ``"add"``, ``"remove"`` or ``"import"``.
day: The day that changed.
actor: The device id that made the change.
paths: Where the trail and its signing key live.
"""
resolved = resolve_paths(paths)
entry: dict[str, Any] = {
"action": action,
"day": to_iso(day),
"actor": actor,
"at": datetime.now(tz=UTC).isoformat(),
}
key_target = resolved.hmac_key
if not key_target.exists():
# First run on this machine: generating the key here means the trail
# is signed from its first line, with no setup step to skip.
generate_key(key_target)
entry["hmac"] = sign(entry, key_file=key_target)
target = resolved.audit
try:
target.parent.mkdir(parents=True, exist_ok=True)
with target.open("a", encoding="utf-8") as handle:
handle.write(json.dumps(entry, sort_keys=True) + "\n")
except OSError as exc:
# The change is already committed to the log; an unwritable trail
# must never be the reason it fails.
_logger.warning("could not append to the sick-day audit trail: %s", exc)

134
sickdays/sickdays/_cli.py Normal file
View File

@ -0,0 +1,134 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The ``sickdays`` command: mark, un-mark and list sick days.
Every day is a normal day until marked here (or on the phone). There is no
budget and no reason to give; the one rule is the calendar -- today or later
only -- plus the cross-pool check: a free day cannot also be a sick day.
``sync-quiet`` is what the timer runs: it reports success even with no
network, because a unit that goes red whenever the laptop is offline is a
unit you learn to ignore.
"""
from __future__ import annotations
import argparse
import sys
from typing import TYPE_CHECKING
from sickdays._api import add_sick_day, is_sick_day, remove_sick_day, sick_days
from sickdays._cli_output import emit, emit_error
from sickdays._day import resolve, to_iso, today
from sickdays._errors import SickDayError
from sickdays._sync import SyncUnavailableError, sync, sync_quietly
if TYPE_CHECKING:
from collections.abc import Sequence
_EXIT_OK = 0
_EXIT_REFUSED = 1
_EXIT_BAD_INPUT = 2
_DAY_HELP = "YYYY-MM-DD, or 'today'/'tomorrow' (default: today)"
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="sickdays",
description=(
"One shared pool of sick days every gate respects. No cap, no "
"reason; today or later only."
),
)
sub = parser.add_subparsers(dest="command", required=True)
add = sub.add_parser("add", help="mark a day sick (today or later)")
add.add_argument("day", nargs="?", default="today", help=_DAY_HELP)
remove = sub.add_parser("remove", help="make a day normal again")
remove.add_argument("day", nargs="?", default="today", help=_DAY_HELP)
sub.add_parser("list", help="every sick day, oldest first")
check = sub.add_parser(
"check", help="exit 0 if the day is sick, 1 if not (for scripts)"
)
check.add_argument("day", nargs="?", default="today", help=_DAY_HELP)
sub.add_parser("sync", help="pull and push the pool now")
sub.add_parser(
"sync-quiet",
help="sync, but exit 0 when the network or credential is unavailable",
)
return parser
def _do_add(raw_day: str) -> int:
entry = add_sick_day(resolve(raw_day))
emit(f"{to_iso(entry.day)} is now a sick day")
return _EXIT_OK
def _do_remove(raw_day: str) -> int:
entry = remove_sick_day(resolve(raw_day))
emit(f"{to_iso(entry.day)} is a normal day again")
return _EXIT_OK
def _do_list() -> int:
days = sorted(sick_days())
if not days:
emit("no sick days")
return _EXIT_OK
reference = today()
for day in days:
emit(f"{to_iso(day)}{' (today)' if day == reference else ''}")
return _EXIT_OK
def _do_check(raw_day: str) -> int:
day = resolve(raw_day)
sick = is_sick_day(day)
emit(f"{to_iso(day)}: {'sick day' if sick else 'normal day'}")
return _EXIT_OK if sick else _EXIT_REFUSED
def _do_sync() -> int:
sync()
emit("pool synced")
return _EXIT_OK
def _do_sync_quiet() -> int:
emit("pool synced" if sync_quietly() else "pool not synced (no remote available)")
return _EXIT_OK
def _dispatch(args: argparse.Namespace) -> int:
if args.command == "add":
return _do_add(args.day)
if args.command == "remove":
return _do_remove(args.day)
if args.command == "list":
return _do_list()
if args.command == "check":
return _do_check(args.day)
if args.command == "sync":
return _do_sync()
return _do_sync_quiet()
def main(argv: Sequence[str] | None = None) -> int:
"""Parse ``argv`` and run one subcommand.
Returns:
``0`` on success, ``1`` for a refused or negative answer, ``2`` for
input that could not be understood.
"""
args = _build_parser().parse_args(argv)
try:
return _dispatch(args)
except ValueError as exc:
emit_error(f"error: {exc}")
return _EXIT_BAD_INPUT
except (SickDayError, SyncUnavailableError) as exc:
emit_error(f"error: {exc}")
return _EXIT_REFUSED
if __name__ == "__main__":
sys.exit(main())

22
sickdays/sickdays/_cli_output.py Normal file
View File

@ -0,0 +1,22 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The one stdout/stderr sink every ``sickdays`` subcommand writes through.
A thin wrapper over ``sys.stdout.write`` (the shape diet-guard and home-guard
already use) so genuine CLI output does not trip ruff's ``T201`` without a
suppression. Kept deliberately trivial: ``T201`` is *auto-fixable*, and its
unsafe fix deletes the whole ``print`` call -- output and all.
"""
from __future__ import annotations
import sys
def emit(text: str = "") -> None:
"""Write one line to stdout."""
sys.stdout.write(f"{text}\n")
def emit_error(text: str) -> None:
"""Write one line to stderr."""
sys.stderr.write(f"{text}\n")

93
sickdays/sickdays/_conflict.py Normal file
View File

@ -0,0 +1,93 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The one rule shared with freedays: a date is never in both pools.
The design is one-way on purpose. freedays never imports sickdays, and
sickdays never imports freedays at module level -- so neither package can
form an import cycle, and ``import sickdays`` costs nothing extra in a gate.
Instead:
* **Adding a sick day** asks a predicate ``is_free(day) -> bool``.
:func:`sickdays.add_sick_day` uses :func:`free_day_predicate` by default,
which imports ``freedays`` *lazily*, at call time, by name. A phone port or
a test passes its own predicate.
* **Marking a free day** is freedays' write, so its callers (CLIs, the phone,
the lock screen) call :func:`ensure_not_sick` first. That needs no import
of freedays at all.
If freedays cannot be imported, the add is *refused*
(:class:`ConflictCheckUnavailableError`): the invariant is hard, and "could
not check" must not pass as "not free".
"""
from __future__ import annotations
from collections.abc import Callable
import importlib
from typing import TYPE_CHECKING, cast
from sickdays._day import to_iso
from sickdays._errors import (
ConflictCheckUnavailableError,
FreeDayConflictError,
SickDayConflictError,
)
from sickdays._model import is_sick_record
from sickdays._paths import resolve_paths
from sickdays._store import read_pool
if TYPE_CHECKING:
from datetime import date
from sickdays._paths import Paths
#: "Is this date a free day?" -- freedays' ``is_free_day`` or a stand-in.
FreePredicate = Callable[["date"], bool]
def free_day_predicate() -> FreePredicate:
"""Return ``freedays.is_free_day``, imported now rather than at load time.
Raises:
ConflictCheckUnavailableError: freedays is not importable here.
"""
try:
module = importlib.import_module("freedays")
except ImportError as exc:
msg = (
"cannot check the free-day pool (freedays is not installed), "
"so the sick day was not added"
)
raise ConflictCheckUnavailableError(msg) from exc
return cast("FreePredicate", module.is_free_day)
def ensure_not_free(day: date, is_free: FreePredicate) -> None:
"""Refuse a sick day on a date the free-day pool already holds.
Raises:
FreeDayConflictError: ``day`` is a free day.
"""
if is_free(day):
msg = (
f"{to_iso(day)} is a free day -- release it with "
"'freedays release' before marking it sick"
)
raise FreeDayConflictError(msg)
def ensure_not_sick(day: date, *, paths: Paths | None = None) -> None:
"""Refuse a free day on a date the sick-day pool already holds.
For freedays' callers, before ``freedays.mark``. Reads the local pool
only; an unreadable pool reads as "not sick", like every other reader.
Raises:
SickDayConflictError: ``day`` is a sick day.
"""
record = read_pool(resolve_paths(paths).log).get(to_iso(day))
if record is not None and is_sick_record(record):
msg = (
f"{to_iso(day)} is a sick day -- remove it with "
"'sickdays remove' before marking it free"
)
raise SickDayConflictError(msg)

56
sickdays/sickdays/_constants.py Normal file
View File

@ -0,0 +1,56 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""On-disk locations, remote paths and record field names.
Every path here is a *default*. Public functions take the corresponding
argument and only fall back to these when it is ``None``, so tests can point
the whole package at a tmp_path and a bug in this package can never write
into real user state. See ``sickdays/tests/conftest.py``, which redirects
these as a second line of defence.
Nothing here is shared with freedays. The two pools are different things
with different names, so they never share a directory, a file or a remote
node -- a sync bug in one can never touch the other.
"""
from __future__ import annotations
from pathlib import Path
from typing import Final
APP_NAME: Final = "sickdays"
STATE_DIR: Final = Path.home() / ".local" / "share" / "sickdays"
#: The CRDT log itself: ISO date -> Record(state=sick|cleared).
LOG_PATH: Final = STATE_DIR / "sick_days.json"
#: crdt-sync's per-device push/pull bookkeeping. Never merged, purely local.
SYNC_STATE_PATH: Final = STATE_DIR / "sync_state.json"
#: This install's persisted uuid. Never a fixed "pc"/"phone" constant, so two
#: machines can never collide on one device directory.
DEVICE_ID_PATH: Final = STATE_DIR / "device_id"
#: Append-only HMAC-signed record of every add/remove/import, for
#: after-the-fact review. Not consulted by any decision -- the log is the truth.
AUDIT_PATH: Final = STATE_DIR / "audit.jsonl"
#: Signing key for the audit trail, generated on first use. User-owned, as in
#: freedays: the signature answers "was this line edited after the fact" for
#: the person's own review, it does not defend a budget (there is none).
HMAC_KEY_FILE: Final = STATE_DIR / "hmac.key"
#: Remote layout, mirroring every other app: one directory per device under
#: a prefix, each device writing only its own file.
SYNC_PATH_PREFIX: Final = "sickdays-sync/devices"
SYNC_FILENAME: Final = "sick_days.json"
#: Field names inside a Record.
FIELD_STATE: Final = "state"
FIELD_ACTOR: Final = "actor"
#: Values of FIELD_STATE. Removing flips the field rather than tombstoning
#: the record: crdt-sync deletes are monotonic (OR-merged), so a tombstoned
#: date could never be marked sick again.
STATE_SICK: Final = "sick"
STATE_CLEARED: Final = "cleared"

78
sickdays/sickdays/_day.py Normal file
View File

@ -0,0 +1,78 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""What "today" means, in one place.
The apps freedays serves disagreed about this before it existed:
screen-locker's lock chain computed the day in UTC while the tool that wrote
its skip file computed it in *local* time, so a skip added near midnight
could name a date the chain never matched. Sick days are a human-facing
calendar concept -- if it is Tuesday where the person is standing, it is
Tuesday -- so this package is local-time throughout and nothing else is
allowed a second opinion.
"""
from __future__ import annotations
from datetime import UTC, date, datetime, timedelta
ISO_FORMAT = "%Y-%m-%d"
def today(now: date | None = None) -> date:
"""Return the current local calendar date.
Args:
now: Override for tests and for callers that already resolved a
reference date. ``None`` reads the system clock in local time.
Returns:
The local calendar date.
"""
return now if now is not None else datetime.now(tz=UTC).astimezone().date()
def parse_iso(text: str) -> date:
"""Parse ``YYYY-MM-DD`` into a :class:`~datetime.date`.
Args:
text: The date string.
Returns:
The parsed date.
Raises:
ValueError: If ``text`` is not a valid ``YYYY-MM-DD`` date. The
message names the offending input, because the usual way to see
this is a typo on the command line.
"""
try:
return date.fromisoformat(text.strip())
except ValueError as exc:
msg = f"not a YYYY-MM-DD date: {text!r}"
raise ValueError(msg) from exc
def to_iso(day: date) -> str:
"""Render ``day`` as ``YYYY-MM-DD``."""
return day.strftime(ISO_FORMAT)
def resolve(text: str, *, now: date | None = None) -> date:
"""Parse a user-supplied day, accepting ``today`` and ``tomorrow``.
Args:
text: A ``YYYY-MM-DD`` date, or the words ``today``/``tomorrow``.
now: Reference date for the relative words.
Returns:
The resolved date.
Raises:
ValueError: If ``text`` is neither a keyword nor a valid date.
"""
cleaned = text.strip().lower()
reference = today(now)
if cleaned == "today":
return reference
if cleaned == "tomorrow":
return reference + timedelta(days=1)
return parse_iso(text)

57
sickdays/sickdays/_errors.py Normal file
View File

@ -0,0 +1,57 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Why an add, a remove or a free-day mark was refused.
Each of these is a *policy* refusal with a message meant to be shown to the
person who typed the command. None of them indicates a bug, so callers are
expected to catch :class:`SickDayError` and print ``str(exc)`` rather than
let a traceback out.
"""
from __future__ import annotations
class SickDayError(Exception):
"""Base class for every policy refusal in this package."""
class PastDateError(SickDayError):
"""A past day can be neither added nor removed.
Without this a gate that already fired and was already failed could be
erased after the fact -- or a day that was rightly excused could be
un-excused retroactively. Only :func:`sickdays.import_legacy` writes past
dates, once, from screen-locker's old history.
"""
class AlreadySickError(SickDayError):
"""The day is already a sick day, so adding it again would be a no-op."""
class NotSickError(SickDayError):
"""The day is not a sick day, so there is nothing to remove."""
class FreeDayConflictError(SickDayError):
"""The day is a free day; a date cannot be in both pools."""
class SickDayConflictError(SickDayError):
"""The day is a sick day, so it cannot also be marked free."""
class ConflictCheckUnavailableError(SickDayError):
"""The free-day pool could not be consulted, so the add is refused.
"A date is never in both pools" is a hard invariant, so not being able
to check it is a refusal, not a pass.
"""
class UnreadablePoolError(SickDayError):
"""The local pool file exists but cannot be read or parsed.
Readers treat that as "no sick days" (fail closed: the gates apply).
Writers refuse instead: rewriting the file from an empty pool would
silently destroy every entry the unreadable file still holds.
"""

84
sickdays/sickdays/_hmac.py Normal file
View File

@ -0,0 +1,84 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Signing the audit trail, using nothing but the standard library.
The same scheme as freedays' ``_hmac`` (HMAC-SHA256 over compact, sorted-key
JSON), copied rather than imported: sickdays and freedays must never import
each other's internals, and ``gatelock.log_integrity`` -- where the scheme
came from -- imports ``tkinter`` through ``gatelock/__init__``, which would
make ``import sickdays`` fail on any machine without Tk.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import secrets
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pathlib import Path
_logger = logging.getLogger(__name__)
_KEY_BYTES = 32
def load_key(key_file: Path) -> bytes | None:
"""Return the key bytes at ``key_file``, or None if unreadable."""
try:
return key_file.read_bytes().strip()
except OSError:
_logger.warning("cannot read the sick-day signing key from %s", key_file)
return None
def generate_key(key_file: Path) -> bytes | None:
"""Create a key at ``key_file`` with owner-only permissions.
Returns:
The new key bytes, or None if it could not be written.
"""
key = secrets.token_bytes(_KEY_BYTES)
try:
key_file.parent.mkdir(parents=True, exist_ok=True)
key_file.write_bytes(key)
key_file.chmod(0o600)
except OSError:
_logger.warning("cannot write the sick-day signing key to %s", key_file)
return None
return key
def _payload(entry: dict[str, Any]) -> bytes:
return json.dumps(entry, sort_keys=True, separators=(",", ":")).encode()
def sign(entry: dict[str, Any], *, key_file: Path) -> str | None:
"""Return the hex HMAC of ``entry``, or None when the key is unavailable.
``entry`` must not already contain a ``hmac`` field.
"""
key = load_key(key_file)
if key is None:
return None
return hmac.new(key, _payload(entry), hashlib.sha256).hexdigest()
def verify(entry: dict[str, Any], *, key_file: Path) -> bool:
"""Whether ``entry``'s stored ``hmac`` matches its contents.
False when the field is missing or not a string, and when the key cannot
be read -- "cannot check" and "does not match" are the same answer to the
only question a caller asks, which is whether to trust the line.
"""
stored = entry.get("hmac")
if not isinstance(stored, str):
return False
key = load_key(key_file)
if key is None:
return False
without = {name: value for name, value in entry.items() if name != "hmac"}
expected = hmac.new(key, _payload(without), hashlib.sha256).hexdigest()
return hmac.compare_digest(stored, expected)

120
sickdays/sickdays/_model.py Normal file
View File

@ -0,0 +1,120 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Translating between a sick day and a crdt-sync ``Record``.
The record id *is* the ISO date, so two devices marking the same day produce
the same record and converge instead of accumulating duplicates.
Adding and removing are a last-writer-wins flip of the ``state`` field, every
field stamped with one hybrid logical clock tick. Merging two copies keeps,
per field, the one with the greater clock -- so the most recent add or
remove of a date wins on every device, and a removed day can be added again
(and removed again) any number of times. A tombstone (``Record.deleted``)
would not allow that: crdt-sync OR-merges it, so a deleted date could never
come back. Nothing here ever sets it, and a tombstoned record reads as "not
sick".
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from crdt_sync import Hlc, Record
from sickdays._constants import FIELD_ACTOR, FIELD_STATE, STATE_CLEARED, STATE_SICK
from sickdays._day import parse_iso, to_iso
if TYPE_CHECKING:
from datetime import date
from crdt_sync import Log
@dataclass(frozen=True)
class SickDay:
"""One day's entry, as the rest of the package wants to see it.
Attributes:
day: The calendar date.
is_sick: Whether the day is currently a sick day.
actor: Which device last changed it.
"""
day: date
is_sick: bool
actor: str = ""
def _field_str(record: Record, name: str) -> str:
field = record.fields.get(name)
if field is None:
return ""
value = field[0]
return value if isinstance(value, str) else ""
def is_sick_record(record: Record) -> bool:
"""Whether ``record`` currently marks its day sick.
Anything not understood -- a tombstone, a missing or unknown state --
reads as "not sick": the gates apply.
"""
if record.deleted:
return False
return _field_str(record, FIELD_STATE) == STATE_SICK
def to_sick_day(record: Record) -> SickDay | None:
"""Build a :class:`SickDay` from ``record``, or None if its id is junk."""
try:
day = parse_iso(record.id)
except ValueError:
return None
return SickDay(
day=day,
is_sick=is_sick_record(record),
actor=_field_str(record, FIELD_ACTOR),
)
def sick_dates(log: Log) -> frozenset[date]:
"""Every date ``log`` currently marks sick (junk ids skipped)."""
days: set[date] = set()
for record in log.values():
entry = to_sick_day(record)
if entry is not None and entry.is_sick:
days.add(entry.day)
return frozenset(days)
def latest_hlc(log: Log, node_id: str) -> Hlc | None:
"""Return the greatest clock this node has issued anywhere in ``log``.
Feeding this back into :meth:`Hlc.new_tick` keeps the node's clock
monotonic across process restarts, so a fresh process never issues a
tick that loses to its own earlier write.
"""
ticks = [
hlc
for record in log.values()
for _, hlc in record.fields.values()
if hlc.node_id == node_id
]
return max(ticks) if ticks else None
def build_record(entry: SickDay, hlc: Hlc) -> Record:
"""Return the record expressing ``entry``, every field stamped ``hlc``.
One tick for every field, so a concurrent add and remove of one day
resolve together rather than interleaving state from one device and the
actor from the other.
"""
state = STATE_SICK if entry.is_sick else STATE_CLEARED
return Record(
id=to_iso(entry.day),
fields={
FIELD_STATE: (state, hlc),
FIELD_ACTOR: (entry.actor, hlc),
},
)

61
sickdays/sickdays/_paths.py Normal file
View File

@ -0,0 +1,61 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Every file this package owns, as one value that can be swapped wholesale.
The same seam freedays uses: one bundle, so ``Paths.under(tmp_path)``
redirects everything at once and a file added to this class later is
redirected by construction rather than by remembering to update a list.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING
from sickdays._constants import (
AUDIT_PATH,
DEVICE_ID_PATH,
HMAC_KEY_FILE,
LOG_PATH,
SYNC_STATE_PATH,
)
if TYPE_CHECKING:
from pathlib import Path
@dataclass(frozen=True)
class Paths:
"""Where the pool, the device id, the audit trail and sync state live."""
log: Path
device_id: Path
audit: Path
hmac_key: Path
sync_state: Path
@classmethod
def default(cls) -> Paths:
"""The real, per-user locations under ``~/.local/share/sickdays``."""
return cls(
log=LOG_PATH,
device_id=DEVICE_ID_PATH,
audit=AUDIT_PATH,
hmac_key=HMAC_KEY_FILE,
sync_state=SYNC_STATE_PATH,
)
@classmethod
def under(cls, directory: Path) -> Paths:
"""Every file, relocated under ``directory``. The test seam."""
return cls(
log=directory / "sick_days.json",
device_id=directory / "device_id",
audit=directory / "audit.jsonl",
hmac_key=directory / "hmac.key",
sync_state=directory / "sync_state.json",
)
def resolve_paths(paths: Paths | None) -> Paths:
"""Return ``paths``, or the real locations when it is ``None``."""
return paths if paths is not None else Paths.default()

74
sickdays/sickdays/_store.py Normal file
View File

@ -0,0 +1,74 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Reading and writing the pool file, failing closed and loudly.
``crdt_sync.read_log`` is deliberately not used for reading: it turns a
corrupt file into an empty log *silently*, and lets a ``PermissionError``
escape. A gate asking "is today a sick day?" must never raise, and an
unreadable pool must be visible in the journal -- it silently re-arms every
lock and alarm, which is the safe direction but a confusing one.
So there are two readers. :func:`read_pool` is for gates: anything wrong
reads as "no sick days" and is logged. :func:`read_pool_strict` is for
writers: a file that exists but cannot be parsed raises, because rewriting
it from an empty pool would destroy whatever it still holds.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from crdt_sync import load_log, write_log
from sickdays._errors import UnreadablePoolError
if TYPE_CHECKING:
from pathlib import Path
from crdt_sync import Log
_logger = logging.getLogger(__name__)
def _read(path: Path) -> Log:
"""Parse ``path``; ``{}`` when missing. Raises on anything else."""
try:
text = path.read_text(encoding="utf-8")
except FileNotFoundError:
return {}
except OSError as exc:
msg = f"the sick-day pool {path} cannot be read: {exc}"
raise UnreadablePoolError(msg) from exc
try:
return load_log(text)
except (ValueError, KeyError, TypeError, AttributeError) as exc:
msg = f"the sick-day pool {path} is corrupt: {exc!r}"
raise UnreadablePoolError(msg) from exc
def read_pool(path: Path) -> Log:
"""Return the pool at ``path``; empty (and logged) when it is unusable.
A missing file is the normal "no sick day was ever taken" and is not
logged. Anything else -- unreadable, unparsable, wrong shape -- is a
warning, and reads as *no sick days*: the gates apply.
"""
try:
return _read(path)
except UnreadablePoolError as exc:
_logger.warning("%s -- treating every day as NOT a sick day", exc)
return {}
def read_pool_strict(path: Path) -> Log:
"""Return the pool at ``path``, or raise if it exists but is unusable.
Raises:
UnreadablePoolError: The file exists but cannot be read or parsed.
"""
return _read(path)
def write_pool(path: Path, log: Log) -> None:
"""Write ``log`` to ``path`` atomically (crdt-sync's temp-then-rename)."""
write_log(path, log)

145
sickdays/sickdays/_sync.py Normal file
View File

@ -0,0 +1,145 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Pushing and pulling the pool, so a sick day marked on the phone lands on the PC.
Sync is a *separate* operation from every decision in :mod:`sickdays._api`.
Nothing that decides whether a gate fires ever waits on this: a gate reads
the last-merged local file and moves on, so an unreachable network makes the
pool stale rather than making a lock hang.
The remote is Firebase RTDB, matching every other app in the fleet, with one
directory per device -- no two devices write the same node, so there is
nothing to conflict and convergence is the CRDT layer's job.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from crdt_sync import (
ConfigError,
FileSyncStateStore,
FirebaseAuthError,
LogCodec,
RemoteSyncError,
RevisionTracking,
SyncTarget,
dump_log,
firebase_client_for,
load_log,
sync_log,
)
from sickdays._api import actor_id, load
from sickdays._constants import (
APP_NAME,
SYNC_FILENAME,
SYNC_PATH_PREFIX,
)
from sickdays._paths import Paths, resolve_paths
from sickdays._store import write_pool
if TYPE_CHECKING:
from crdt_sync import Log, RemoteStore
_logger = logging.getLogger(__name__)
class SyncUnavailableError(Exception):
"""No usable remote right now -- unconfigured, offline, or rejected.
Deliberately not a subclass of anything in :mod:`sickdays._errors`: this
is never a policy refusal, and callers treat it the way they treat a
network timeout.
"""
def get_client() -> RemoteStore:
"""Return a signed-in Firebase client, or raise :class:`SyncUnavailableError`.
``ConfigError`` subclasses ``Exception`` directly rather than
``RemoteSyncError``, so both are caught explicitly -- assuming one covers
the other has bitten two other apps in this fleet already.
"""
try:
return firebase_client_for(APP_NAME)
except ConfigError as exc:
msg = f"sick-day sync is not configured on this machine: {exc}"
raise SyncUnavailableError(msg) from exc
except FirebaseAuthError as exc:
msg = f"sick-day sync credentials were rejected: {exc}"
raise SyncUnavailableError(msg) from exc
def sync(
*,
client: RemoteStore | None = None,
paths: Paths | None = None,
) -> Log:
"""Run one pull/merge/push tick and persist the merged pool.
Args:
client: Override for the remote; ``None`` builds the Firebase one.
paths: Where the pool, uuid and sync bookkeeping live.
Returns:
The merged pool, as pushed.
Raises:
SyncUnavailableError: If no remote could be reached or authenticated.
"""
resolved = resolve_paths(paths)
target_client = client if client is not None else get_client()
node = actor_id(resolved)
local = load(resolved.log)
state_path = resolved.sync_state
try:
merged = sync_log(
SyncTarget(
client=target_client,
device_id=node,
path_prefix=SYNC_PATH_PREFIX,
),
local,
LogCodec(
encode=dump_log,
decode=load_log,
filename=SYNC_FILENAME,
commit_message="sickdays: update pool",
),
RevisionTracking(state_store=FileSyncStateStore(state_path)),
)
except RemoteSyncError as exc:
msg = f"sick-day sync could not reach the remote: {exc}"
raise SyncUnavailableError(msg) from exc
write_pool(resolved.log, merged)
return merged
def sync_quietly(*, paths: Paths | None = None) -> bool:
"""Sync, swallowing every failure. Returns whether it actually ran.
For callers on a timer that want a fresher pool but must not fail, and
must not say anything, when the sync cannot happen.
The broad catch is deliberate and is the whole point of this function.
Today ``FirebaseAuthError`` happens to subclass ``RemoteSyncError`` and
``requests``' ``JSONDecodeError`` happens to subclass
``RequestException``, so the narrow handler above would in fact catch an
expired refresh token or a malformed payload. But that is an upstream
class hierarchy, not a promise to this package -- and a background sync
whose contract is "never fail" must not have that contract hold only by
coincidence. Anything unexpected is logged with its traceback, so a real
bug is loud in the journal while the timer stays green.
"""
try:
sync(paths=paths)
except SyncUnavailableError as exc:
# The expected case -- offline, or not configured yet. Debug level:
# this is normal on a laptop and must not fill the journal.
_logger.debug("sick-day sync skipped: %s", exc)
return False
except Exception: # pylint: disable=broad-exception-caught
_logger.exception("sick-day sync failed unexpectedly")
return False
return True

2
sickdays/sickdays/tests/__init__.py Normal file
View File

@ -0,0 +1,2 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Tests for :mod:`sickdays`."""

50
sickdays/sickdays/tests/conftest.py Normal file
View File

@ -0,0 +1,50 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Redirect every on-disk path, and the free-day pool, away from real state.
Every writer resolves its paths through :func:`sickdays._paths.resolve_paths`,
so redirecting that one function in each module that imports it moves all of
them, including any file added later.
The cross-pool check would otherwise import the real freedays and read the
real ``~/.local/share/freedays`` pool, so it is stubbed to "nothing is free"
for every test; tests of the check itself pass their own predicate.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
import pytest
from sickdays._paths import Paths
if TYPE_CHECKING:
from datetime import date
from pathlib import Path
_MODULES = ("sickdays._api", "sickdays._audit", "sickdays._conflict", "sickdays._sync")
def _never_free(_day: date) -> bool:
return False
@pytest.fixture(autouse=True)
def _redirect_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Point every default path at ``tmp_path`` before any test runs."""
state = tmp_path / "state"
state.mkdir()
redirected = Paths.under(state)
def _resolve(paths: Paths | None) -> Paths:
return paths if paths is not None else redirected
for module in _MODULES:
monkeypatch.setattr(f"{module}.resolve_paths", _resolve)
monkeypatch.setattr("sickdays._api.free_day_predicate", lambda: _never_free)
@pytest.fixture
def state_dir(tmp_path: Path) -> Path:
"""The redirected state directory, for tests that assert on files."""
return tmp_path / "state"

146
sickdays/sickdays/tests/test_api.py Normal file
View File

@ -0,0 +1,146 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Adding, removing, listing and the one-time legacy import."""
from __future__ import annotations
from datetime import date
import json
from typing import TYPE_CHECKING
import pytest
from sickdays._api import (
actor_id,
add_sick_day,
import_legacy,
is_sick_day,
load,
remove_sick_day,
sick_days,
)
from sickdays._errors import (
AlreadySickError,
FreeDayConflictError,
NotSickError,
PastDateError,
)
from sickdays._paths import Paths
if TYPE_CHECKING:
from pathlib import Path
TODAY = date(2026, 10, 16)
TOMORROW = date(2026, 10, 17)
YESTERDAY = date(2026, 10, 15)
def test_every_day_is_normal_by_default() -> None:
assert not is_sick_day(TODAY)
assert sick_days() == frozenset()
def test_adding_today_makes_it_a_sick_day() -> None:
entry = add_sick_day(TODAY, now=TODAY)
assert entry.is_sick
assert is_sick_day(TODAY)
assert is_sick_day(now=TODAY)
def test_there_is_no_cap() -> None:
for offset in range(60):
add_sick_day(date.fromordinal(TODAY.toordinal() + offset), now=TODAY)
assert len(sick_days()) == 60
def test_a_past_day_cannot_be_added() -> None:
with pytest.raises(PastDateError, match="already passed"):
add_sick_day(YESTERDAY, now=TODAY)
assert not is_sick_day(YESTERDAY)
def test_adding_twice_is_refused() -> None:
add_sick_day(TOMORROW, now=TODAY)
with pytest.raises(AlreadySickError):
add_sick_day(TOMORROW, now=TODAY)
def test_a_free_day_cannot_become_a_sick_day() -> None:
with pytest.raises(FreeDayConflictError, match="free day"):
add_sick_day(TOMORROW, is_free=lambda _day: True, now=TODAY)
assert not is_sick_day(TOMORROW)
def test_the_default_check_asks_the_free_day_pool(
monkeypatch: pytest.MonkeyPatch,
) -> None:
asked: list[date] = []
def _free(day: date) -> bool:
asked.append(day)
return True
monkeypatch.setattr("sickdays._api.free_day_predicate", lambda: _free)
with pytest.raises(FreeDayConflictError):
add_sick_day(TOMORROW, now=TODAY)
assert asked == [TOMORROW]
def test_removing_makes_the_day_normal_again() -> None:
add_sick_day(TOMORROW, now=TODAY)
entry = remove_sick_day(TOMORROW, now=TODAY)
assert not entry.is_sick
assert not is_sick_day(TOMORROW)
def test_a_removed_day_can_be_added_again() -> None:
add_sick_day(TODAY, now=TODAY)
remove_sick_day(TODAY, now=TODAY)
add_sick_day(TODAY, now=TODAY)
assert is_sick_day(TODAY)
def test_a_past_day_cannot_be_removed() -> None:
import_legacy([YESTERDAY])
with pytest.raises(PastDateError, match="removed"):
remove_sick_day(YESTERDAY, now=TODAY)
assert is_sick_day(YESTERDAY)
def test_removing_a_normal_day_is_refused() -> None:
with pytest.raises(NotSickError):
remove_sick_day(TOMORROW, now=TODAY)
def test_the_legacy_import_writes_past_days() -> None:
days = [date(2026, 7, 24), date(2026, 6, 21)]
assert import_legacy(days) == sorted(days)
assert sick_days() == frozenset(days)
def test_the_legacy_import_is_idempotent(state_dir: Path) -> None:
import_legacy([date(2026, 6, 21)])
assert import_legacy([date(2026, 6, 21), date(2026, 6, 21)]) == []
lines = (state_dir / "audit.jsonl").read_text(encoding="utf-8").splitlines()
assert [json.loads(line)["action"] for line in lines] == ["import"]
def test_every_change_is_audited(state_dir: Path) -> None:
add_sick_day(TODAY, now=TODAY)
remove_sick_day(TODAY, now=TODAY)
lines = (state_dir / "audit.jsonl").read_text(encoding="utf-8").splitlines()
assert [json.loads(line)["action"] for line in lines] == ["add", "remove"]
def test_the_actor_is_this_installs_uuid() -> None:
entry = add_sick_day(TODAY, now=TODAY)
assert entry.actor == actor_id()
assert len(entry.actor) == 36
def test_an_explicit_pool_is_honoured(tmp_path: Path) -> None:
paths = Paths.under(tmp_path / "elsewhere")
add_sick_day(TOMORROW, paths=paths, now=TODAY)
assert is_sick_day(TOMORROW, log_path=paths.log)
assert sick_days(log_path=paths.log) == frozenset({TOMORROW})
assert not is_sick_day(TOMORROW)
assert load(paths.log)

89
sickdays/sickdays/tests/test_audit.py Normal file
View File

@ -0,0 +1,89 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The audit trail: best-effort, signed, and never in the way."""
from __future__ import annotations
from dataclasses import replace
from datetime import date
import json
from typing import TYPE_CHECKING
from sickdays._audit import record_event
from sickdays._hmac import verify
from sickdays._paths import Paths
if TYPE_CHECKING:
from pathlib import Path
import pytest
DAY = date(2026, 12, 24)
def _lines(path: Path) -> list[dict[str, object]]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
def test_an_event_is_appended_with_its_details(state_dir: Path) -> None:
record_event("add", DAY, actor="device-1")
(entry,) = _lines(state_dir / "audit.jsonl")
assert entry["action"] == "add"
assert entry["day"] == "2026-12-24"
assert entry["actor"] == "device-1"
def test_events_accumulate_rather_than_overwrite(state_dir: Path) -> None:
record_event("add", DAY, actor="device-1")
record_event("remove", DAY, actor="device-1")
assert len(_lines(state_dir / "audit.jsonl")) == 2
def test_the_signing_key_is_generated_on_first_use(state_dir: Path) -> None:
key = state_dir / "hmac.key"
assert not key.exists()
record_event("add", DAY, actor="device-1")
assert key.exists()
def test_the_written_entry_verifies_against_the_key(state_dir: Path) -> None:
record_event("add", DAY, actor="device-1")
(entry,) = _lines(state_dir / "audit.jsonl")
assert verify(entry, key_file=state_dir / "hmac.key")
def test_a_tampered_entry_stops_verifying(state_dir: Path) -> None:
record_event("add", DAY, actor="device-1")
(entry,) = _lines(state_dir / "audit.jsonl")
entry["day"] = "2026-12-25"
assert not verify(entry, key_file=state_dir / "hmac.key")
def test_an_existing_key_is_reused_rather_than_regenerated(state_dir: Path) -> None:
record_event("add", DAY, actor="device-1")
original = (state_dir / "hmac.key").read_bytes()
record_event("remove", DAY, actor="device-1")
assert (state_dir / "hmac.key").read_bytes() == original
def test_explicit_paths_are_honoured(tmp_path: Path) -> None:
elsewhere = tmp_path / "elsewhere"
record_event("add", DAY, actor="device-1", paths=Paths.under(elsewhere))
(entry,) = _lines(elsewhere / "audit.jsonl")
assert verify(entry, key_file=elsewhere / "hmac.key")
def test_an_unwritable_trail_warns_but_never_raises(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""A sick day must be addable even when its audit line cannot be written."""
blocked = tmp_path / "a-file" / "audit.jsonl"
blocked.parent.write_text("I am a file, not a directory", encoding="utf-8")
with caplog.at_level("WARNING"):
record_event(
"add",
DAY,
actor="device-1",
paths=replace(Paths.under(tmp_path), audit=blocked),
)
assert "could not append" in caplog.text

117
sickdays/sickdays/tests/test_cli.py Normal file
View File

@ -0,0 +1,117 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The command line: add, remove, list, check, sync."""
from __future__ import annotations
from datetime import timedelta
import runpy
import pytest
from sickdays._cli import main
from sickdays._day import to_iso, today
from sickdays._sync import SyncUnavailableError
from sickdays.tests.test_sync import FakeRemote
def _future() -> str:
return to_iso(today() + timedelta(days=30))
def _refuse() -> None:
msg = "no credentials on this machine"
raise SyncUnavailableError(msg)
def test_list_on_an_empty_pool(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["list"]) == 0
assert "no sick days" in capsys.readouterr().out
def test_add_defaults_to_today(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["add"]) == 0
assert f"{to_iso(today())} is now a sick day" in capsys.readouterr().out
assert main(["check"]) == 0
def test_list_marks_today(capsys: pytest.CaptureFixture[str]) -> None:
main(["add", "today"])
main(["add", _future()])
capsys.readouterr()
assert main(["list"]) == 0
out = capsys.readouterr().out.splitlines()
assert out == [f"{to_iso(today())} (today)", _future()]
def test_remove_makes_the_day_normal(capsys: pytest.CaptureFixture[str]) -> None:
main(["add", "tomorrow"])
assert main(["remove", "tomorrow"]) == 0
assert "normal day again" in capsys.readouterr().out
assert main(["check", "tomorrow"]) == 1
def test_a_past_day_is_refused(capsys: pytest.CaptureFixture[str]) -> None:
yesterday = to_iso(today() - timedelta(days=1))
assert main(["add", yesterday]) == 1
assert "already passed" in capsys.readouterr().err
assert main(["remove", yesterday]) == 1
def test_removing_a_normal_day_is_refused(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["remove", _future()]) == 1
assert "not a sick day" in capsys.readouterr().err
def test_a_free_day_is_refused(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr("sickdays._api.free_day_predicate", lambda: lambda _d: True)
assert main(["add", _future()]) == 1
assert "is a free day" in capsys.readouterr().err
def test_an_unparsable_date_exits_two(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["add", "next tuesday"]) == 2
assert "not a YYYY-MM-DD date" in capsys.readouterr().err
def test_a_missing_subcommand_is_rejected() -> None:
with pytest.raises(SystemExit):
main([])
def test_sync_reports_success(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr("sickdays._sync.firebase_client_for", lambda _app: FakeRemote())
assert main(["sync"]) == 0
assert "pool synced" in capsys.readouterr().out
def test_sync_failure_exits_one(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr("sickdays._sync.get_client", _refuse)
assert main(["sync"]) == 1
assert "no credentials" in capsys.readouterr().err
def test_sync_quiet_exits_zero_either_way(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr("sickdays._sync.firebase_client_for", lambda _app: FakeRemote())
assert main(["sync-quiet"]) == 0
assert "pool synced" in capsys.readouterr().out
monkeypatch.setattr("sickdays._sync.get_client", _refuse)
assert main(["sync-quiet"]) == 0
assert "no remote available" in capsys.readouterr().out
def test_python_dash_m_runs_the_cli(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setattr("sys.argv", ["sickdays", "list"])
with pytest.raises(SystemExit) as exited:
runpy.run_module("sickdays", run_name="__main__")
assert exited.value.code == 0
assert "no sick days" in capsys.readouterr().out

65
sickdays/sickdays/tests/test_conflict.py Normal file
View File

@ -0,0 +1,65 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""A date is never in both pools: the lazy free-day check and its mirror."""
from __future__ import annotations
from datetime import date
import sys
import types
from typing import TYPE_CHECKING
import pytest
from sickdays._api import add_sick_day
from sickdays._conflict import ensure_not_free, ensure_not_sick, free_day_predicate
from sickdays._errors import (
ConflictCheckUnavailableError,
FreeDayConflictError,
SickDayConflictError,
)
from sickdays._paths import Paths
if TYPE_CHECKING:
from pathlib import Path
TODAY = date(2026, 10, 16)
def test_the_predicate_is_freedays_own(monkeypatch: pytest.MonkeyPatch) -> None:
fake = types.ModuleType("freedays")
def _is_free_day(day: date) -> bool:
return day == TODAY
vars(fake)["is_free_day"] = _is_free_day
monkeypatch.setitem(sys.modules, "freedays", fake)
predicate = free_day_predicate()
assert predicate(TODAY)
assert not predicate(date(2026, 10, 17))
def test_no_freedays_means_the_add_is_refused(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setitem(sys.modules, "freedays", None)
with pytest.raises(ConflictCheckUnavailableError, match="not installed"):
free_day_predicate()
def test_ensure_not_free_passes_a_normal_day() -> None:
ensure_not_free(TODAY, lambda _day: False)
with pytest.raises(FreeDayConflictError, match="freedays release"):
ensure_not_free(TODAY, lambda _day: True)
def test_a_sick_day_cannot_be_marked_free() -> None:
ensure_not_sick(TODAY)
add_sick_day(TODAY, now=TODAY)
with pytest.raises(SickDayConflictError, match="sickdays remove"):
ensure_not_sick(TODAY)
def test_ensure_not_sick_honours_explicit_paths(tmp_path: Path) -> None:
elsewhere = Paths.under(tmp_path / "elsewhere")
add_sick_day(TODAY, paths=elsewhere, now=TODAY)
ensure_not_sick(TODAY)
with pytest.raises(SickDayConflictError):
ensure_not_sick(TODAY, paths=elsewhere)

50
sickdays/sickdays/tests/test_day.py Normal file
View File

@ -0,0 +1,50 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Date handling: the one place allowed an opinion about "today"."""
from __future__ import annotations
from datetime import date
import pytest
from sickdays._day import parse_iso, resolve, to_iso, today
def test_today_reads_the_clock_when_not_overridden() -> None:
assert isinstance(today(), date)
def test_today_returns_the_override_untouched() -> None:
fixed = date(2026, 3, 4)
assert today(fixed) == fixed
def test_parse_iso_round_trips() -> None:
assert to_iso(parse_iso("2026-12-24")) == "2026-12-24"
def test_parse_iso_tolerates_surrounding_whitespace() -> None:
assert parse_iso(" 2026-12-24 ") == date(2026, 12, 24)
@pytest.mark.parametrize("bad", ["24-12-2026", "2026-13-01", "tomorrow", ""])
def test_parse_iso_rejects_junk_and_names_it(bad: str) -> None:
with pytest.raises(ValueError, match="not a YYYY-MM-DD date"):
parse_iso(bad)
def test_resolve_understands_today() -> None:
assert resolve("today", now=date(2026, 3, 4)) == date(2026, 3, 4)
def test_resolve_understands_tomorrow_across_a_month_boundary() -> None:
assert resolve("TOMORROW", now=date(2026, 3, 31)) == date(2026, 4, 1)
def test_resolve_falls_through_to_a_plain_date() -> None:
assert resolve("2026-12-24", now=date(2026, 3, 4)) == date(2026, 12, 24)
def test_resolve_rejects_junk() -> None:
with pytest.raises(ValueError, match="not a YYYY-MM-DD date"):
resolve("next tuesday", now=date(2026, 3, 4))

97
sickdays/sickdays/tests/test_hmac.py Normal file
View File

@ -0,0 +1,97 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Signing, and the ways it is allowed to fail."""
from __future__ import annotations
from typing import TYPE_CHECKING
from sickdays._hmac import generate_key, load_key, sign, verify
if TYPE_CHECKING:
from pathlib import Path
_ENTRY = {"action": "mark", "day": "2026-12-24", "actor": "device-1"}
def test_a_generated_key_is_owner_only(tmp_path: Path) -> None:
"""A signing key the rest of the machine can read is not much of one."""
key_file = tmp_path / "hmac.key"
generate_key(key_file)
assert key_file.stat().st_mode & 0o077 == 0
def test_a_generated_key_round_trips(tmp_path: Path) -> None:
key_file = tmp_path / "hmac.key"
created = generate_key(key_file)
assert created is not None
assert load_key(key_file) == created
def test_two_generated_keys_differ(tmp_path: Path) -> None:
first = generate_key(tmp_path / "a.key")
second = generate_key(tmp_path / "b.key")
assert first != second
def test_a_key_that_cannot_be_written_returns_none(tmp_path: Path) -> None:
blocked = tmp_path / "a-file" / "hmac.key"
blocked.parent.write_text("I am a file, not a directory", encoding="utf-8")
assert generate_key(blocked) is None
def test_a_missing_key_reads_as_none(tmp_path: Path) -> None:
assert load_key(tmp_path / "absent.key") is None
def test_signing_and_verifying_agree(tmp_path: Path) -> None:
key_file = tmp_path / "hmac.key"
generate_key(key_file)
signature = sign(_ENTRY, key_file=key_file)
assert signature is not None
assert verify({**_ENTRY, "hmac": signature}, key_file=key_file)
def test_signing_is_stable_across_key_order(tmp_path: Path) -> None:
"""Sorted-key JSON: the same fields sign the same however they arrive."""
key_file = tmp_path / "hmac.key"
generate_key(key_file)
reordered = dict(reversed(list(_ENTRY.items())))
assert sign(_ENTRY, key_file=key_file) == sign(reordered, key_file=key_file)
def test_a_tampered_field_fails_verification(tmp_path: Path) -> None:
key_file = tmp_path / "hmac.key"
generate_key(key_file)
signature = sign(_ENTRY, key_file=key_file)
tampered = {**_ENTRY, "day": "2026-12-25", "hmac": signature}
assert not verify(tampered, key_file=key_file)
def test_another_key_fails_verification(tmp_path: Path) -> None:
mine = tmp_path / "mine.key"
theirs = tmp_path / "theirs.key"
generate_key(mine)
generate_key(theirs)
signature = sign(_ENTRY, key_file=mine)
assert not verify({**_ENTRY, "hmac": signature}, key_file=theirs)
def test_an_entry_without_a_signature_does_not_verify(tmp_path: Path) -> None:
key_file = tmp_path / "hmac.key"
generate_key(key_file)
assert not verify(dict(_ENTRY), key_file=key_file)
def test_a_non_string_signature_does_not_verify(tmp_path: Path) -> None:
key_file = tmp_path / "hmac.key"
generate_key(key_file)
assert not verify({**_ENTRY, "hmac": 17}, key_file=key_file)
def test_verification_without_a_key_is_a_no(tmp_path: Path) -> None:
""" "Cannot check" and "does not match" are the same answer to the caller."""
assert not verify({**_ENTRY, "hmac": "deadbeef"}, key_file=tmp_path / "absent.key")
def test_signing_without_a_key_returns_none(tmp_path: Path) -> None:
assert sign(_ENTRY, key_file=tmp_path / "absent.key") is None

70
sickdays/sickdays/tests/test_model.py Normal file
View File

@ -0,0 +1,70 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Records, and how two devices' copies of a date converge."""
from __future__ import annotations
from datetime import date
from crdt_sync import Hlc, Record, merge_logs
from sickdays._model import (
SickDay,
build_record,
is_sick_record,
latest_hlc,
sick_dates,
to_sick_day,
)
DAY = date(2026, 10, 20)
def _tick(node: str, wall_ms: int) -> Hlc:
return Hlc(wall_time_ms=wall_ms, counter=0, node_id=node)
def _record(*, sick: bool, node: str, wall_ms: int) -> Record:
return build_record(SickDay(DAY, is_sick=sick, actor=node), _tick(node, wall_ms))
def test_a_record_round_trips() -> None:
record = _record(sick=True, node="pc", wall_ms=1)
assert to_sick_day(record) == SickDay(DAY, is_sick=True, actor="pc")
def test_a_later_removal_on_another_device_wins_everywhere() -> None:
added = {"2026-10-20": _record(sick=True, node="pc", wall_ms=1)}
removed = {"2026-10-20": _record(sick=False, node="phone", wall_ms=2)}
assert sick_dates(merge_logs(added, removed)) == frozenset()
assert sick_dates(merge_logs(removed, added)) == frozenset()
def test_a_later_re_add_wins_over_an_earlier_removal() -> None:
removed = {"2026-10-20": _record(sick=False, node="pc", wall_ms=2)}
added_again = {"2026-10-20": _record(sick=True, node="phone", wall_ms=3)}
assert sick_dates(merge_logs(removed, added_again)) == {DAY}
assert sick_dates(merge_logs(added_again, removed)) == {DAY}
def test_a_tombstone_never_reads_as_sick() -> None:
record = _record(sick=True, node="pc", wall_ms=1)
tombstoned = Record(id=record.id, fields=record.fields, deleted=True)
assert not is_sick_record(tombstoned)
def test_junk_ids_and_odd_values_are_tolerated() -> None:
junk = Record(id="not-a-date", fields={"state": ("sick", _tick("pc", 1))})
assert to_sick_day(junk) is None
odd = Record(id="2026-10-20", fields={"state": (7, _tick("pc", 1))})
assert not is_sick_record(odd)
assert to_sick_day(odd) == SickDay(DAY, is_sick=False, actor="")
assert sick_dates({"x": junk, "y": odd}) == frozenset()
def test_latest_hlc_is_per_node() -> None:
log = {
"a": _record(sick=True, node="pc", wall_ms=5),
"b": _record(sick=True, node="phone", wall_ms=9),
}
assert latest_hlc(log, "pc") == _tick("pc", 5)
assert latest_hlc(log, "tablet") is None

60
sickdays/sickdays/tests/test_paths.py Normal file
View File

@ -0,0 +1,60 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""The path bundle itself, exercised unpatched.
The autouse redirect in ``conftest`` replaces ``resolve_paths`` *as imported
by* ``_api``/``_audit``/``_sync``. It does not touch ``sickdays._paths``, so
the functions below are the real ones -- which is the point: the seam every
other test relies on has to be tested somewhere, and that somewhere cannot
be behind the seam.
"""
from __future__ import annotations
from pathlib import Path
from sickdays._constants import (
AUDIT_PATH,
DEVICE_ID_PATH,
HMAC_KEY_FILE,
LOG_PATH,
STATE_DIR,
SYNC_STATE_PATH,
)
from sickdays._paths import Paths, resolve_paths
def test_the_defaults_are_the_real_per_user_locations() -> None:
defaults = Paths.default()
assert defaults.log == LOG_PATH
assert defaults.device_id == DEVICE_ID_PATH
assert defaults.audit == AUDIT_PATH
assert defaults.hmac_key == HMAC_KEY_FILE
assert defaults.sync_state == SYNC_STATE_PATH
def test_every_default_lives_under_one_state_directory() -> None:
"""Nothing may escape ``~/.local/share/sickdays``."""
for path in vars(Paths.default()).values():
assert isinstance(path, Path)
assert path.parent == STATE_DIR
def test_under_relocates_every_single_file(tmp_path: Path) -> None:
"""The property the test redirect depends on: no file is left behind."""
relocated = Paths.under(tmp_path)
for path in vars(relocated).values():
assert path.parent == tmp_path
def test_under_gives_each_file_a_distinct_name(tmp_path: Path) -> None:
names = [path.name for path in vars(Paths.under(tmp_path)).values()]
assert len(names) == len(set(names))
def test_resolve_returns_the_bundle_it_was_given(tmp_path: Path) -> None:
explicit = Paths.under(tmp_path)
assert resolve_paths(explicit) is explicit
def test_resolve_falls_back_to_the_defaults() -> None:
assert resolve_paths(None) == Paths.default()

76
sickdays/sickdays/tests/test_store.py Normal file
View File

@ -0,0 +1,76 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""An unusable pool fails closed for readers, and refuses writers, loudly."""
from __future__ import annotations
from datetime import date
from typing import TYPE_CHECKING
import pytest
from sickdays._api import add_sick_day, import_legacy, is_sick_day, remove_sick_day
from sickdays._errors import UnreadablePoolError
from sickdays._store import read_pool, read_pool_strict
if TYPE_CHECKING:
from pathlib import Path
TODAY = date(2026, 10, 16)
CORRUPT_PAYLOADS = (
"{not json",
"[1, 2, 3]",
'{"2026-10-16": {"fields": 5}}',
'{"2026-10-16": {"id": "2026-10-16", "fields": {}}}',
(
'{"2026-10-16": {"id": "x", "fields": {"state": ["sick", "nope"]}, '
'"deleted": false}}'
),
)
@pytest.mark.parametrize("payload", CORRUPT_PAYLOADS)
def test_a_corrupt_pool_reads_as_not_sick_and_is_logged(
payload: str, state_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
(state_dir / "sick_days.json").write_text(payload, encoding="utf-8")
with caplog.at_level("WARNING"):
assert not is_sick_day(TODAY)
assert "NOT a sick day" in caplog.text
assert "corrupt" in caplog.text
def test_an_unreadable_pool_reads_as_not_sick_and_is_logged(
state_dir: Path, caplog: pytest.LogCaptureFixture
) -> None:
(state_dir / "sick_days.json").mkdir()
with caplog.at_level("WARNING"):
assert not is_sick_day(TODAY)
assert "cannot be read" in caplog.text
def test_a_missing_pool_is_silent(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level("DEBUG"):
assert read_pool(tmp_path / "absent.json") == {}
assert not caplog.text
def test_strict_reading_raises_instead(tmp_path: Path) -> None:
target = tmp_path / "pool.json"
target.write_text("{not json", encoding="utf-8")
with pytest.raises(UnreadablePoolError, match="corrupt"):
read_pool_strict(target)
def test_writers_refuse_rather_than_overwrite_a_corrupt_pool(state_dir: Path) -> None:
pool = state_dir / "sick_days.json"
pool.write_text("{not json", encoding="utf-8")
with pytest.raises(UnreadablePoolError):
add_sick_day(TODAY, now=TODAY)
with pytest.raises(UnreadablePoolError):
remove_sick_day(TODAY, now=TODAY)
with pytest.raises(UnreadablePoolError):
import_legacy([TODAY])
assert pool.read_text(encoding="utf-8") == "{not json"

214
sickdays/sickdays/tests/test_sync.py Normal file
View File

@ -0,0 +1,214 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Sync: convergence across devices, and silence when the network is not there."""
from __future__ import annotations
from datetime import date
from typing import TYPE_CHECKING
from crdt_sync import ConfigError, FirebaseAuthError, RemoteSyncError, dump_log
import pytest
from sickdays._api import add_sick_day, is_sick_day, load, remove_sick_day
from sickdays._paths import Paths
from sickdays._store import write_pool
from sickdays._sync import SyncUnavailableError, get_client, sync, sync_quietly
if TYPE_CHECKING:
from pathlib import Path
TODAY = date(2026, 10, 16)
LATER = date(2026, 12, 24)
class FakeRemote:
"""An in-memory RemoteStore: a dict with the four methods sync calls."""
def __init__(self, files: dict[str, str] | None = None) -> None:
self.files: dict[str, str] = dict(files or {})
def list_directory(self, path: str) -> list[str]:
prefix = f"{path}/"
return sorted(
{
name[len(prefix) :].split("/", 1)[0]
for name in self.files
if name.startswith(prefix)
}
)
def get_file_text(self, path: str) -> str | None:
return self.files.get(path)
def put_file_text(self, path: str, text: str, *, message: str) -> None:
del message
self.files[path] = text
def delete_file(self, path: str, *, message: str = "") -> None:
del message
self.files.pop(path, None)
def can_access_remote(self) -> bool:
return True
class BrokenRemote(FakeRemote):
"""A remote that fails the way an offline one does."""
def list_directory(self, path: str) -> list[str]:
msg = f"cannot reach {path}"
raise RemoteSyncError(msg)
def test_a_local_day_is_pushed_to_the_remote() -> None:
add_sick_day(LATER, now=TODAY)
remote = FakeRemote()
sync(client=remote)
assert any("sickdays-sync/devices/" in path for path in remote.files)
def test_a_peers_day_arrives_and_becomes_sick_locally(state_dir: Path) -> None:
peer_log = {}
# Build a peer's pushed log by marking into a separate pool file.
add_sick_day(LATER, now=TODAY)
peer_log = load()
remote = FakeRemote(
{"sickdays-sync/devices/other-device/sick_days.json": dump_log(peer_log)}
)
# Start over with an empty local pool, then pull.
write_pool(Paths.under(state_dir).log, {})
assert not is_sick_day(LATER)
sync(client=remote)
assert is_sick_day(LATER)
def test_syncing_twice_changes_nothing() -> None:
add_sick_day(LATER, now=TODAY)
remote = FakeRemote()
first = sync(client=remote)
second = sync(client=remote)
assert first == second
def test_a_corrupt_peer_push_is_skipped_rather_than_fatal() -> None:
add_sick_day(LATER, now=TODAY)
remote = FakeRemote(
{"sickdays-sync/devices/broken/sick_days.json": "{not json at all"}
)
sync(client=remote)
assert is_sick_day(LATER)
def test_an_unreachable_remote_raises_sync_unavailable() -> None:
with pytest.raises(SyncUnavailableError, match="could not reach"):
sync(client=BrokenRemote())
def test_sync_quietly_reports_failure_without_raising(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
def _refuse() -> None:
msg = "not configured"
raise SyncUnavailableError(msg)
monkeypatch.setattr("sickdays._sync.get_client", _refuse)
with caplog.at_level("DEBUG"):
assert sync_quietly() is False
assert "sync skipped" in caplog.text
def test_sync_quietly_reports_success(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("sickdays._sync.get_client", FakeRemote)
assert sync_quietly() is True
def test_an_unconfigured_machine_cannot_sync(monkeypatch: pytest.MonkeyPatch) -> None:
def _unconfigured(_app: str) -> None:
msg = "no config file"
raise ConfigError(msg)
monkeypatch.setattr("sickdays._sync.firebase_client_for", _unconfigured)
with pytest.raises(SyncUnavailableError, match="not configured"):
get_client()
def test_a_rejected_credential_cannot_sync(monkeypatch: pytest.MonkeyPatch) -> None:
def _rejected(_app: str) -> None:
msg = "bad password"
raise FirebaseAuthError(msg)
monkeypatch.setattr("sickdays._sync.firebase_client_for", _rejected)
with pytest.raises(SyncUnavailableError, match="rejected"):
get_client()
def test_the_configured_client_is_used_when_none_is_passed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
remote = FakeRemote()
monkeypatch.setattr("sickdays._sync.firebase_client_for", lambda _app: remote)
add_sick_day(LATER, now=TODAY)
sync()
assert remote.files
def test_the_pool_can_be_pointed_elsewhere_wholesale(tmp_path: Path) -> None:
elsewhere = tmp_path / "other"
elsewhere.mkdir()
paths = Paths.under(elsewhere)
add_sick_day(LATER, now=TODAY, paths=paths)
sync(client=FakeRemote(), paths=paths)
assert is_sick_day(LATER, log_path=paths.log)
def test_an_expired_credential_mid_sync_does_not_fail_the_timer(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A refresh token expiring during the tick, not at client construction."""
class ExpiringRemote(FakeRemote):
def list_directory(self, path: str) -> list[str]:
msg = f"refresh token rejected while listing {path}"
raise FirebaseAuthError(msg)
monkeypatch.setattr(
"sickdays._sync.firebase_client_for", lambda _app: ExpiringRemote()
)
assert sync_quietly() is False
def test_an_unexpected_error_is_logged_with_a_traceback_and_swallowed(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
"""The contract is "never fail", and it must not hold only by coincidence.
FirebaseAuthError happens to subclass RemoteSyncError today. Something
that does not -- a bug, a changed upstream hierarchy -- must still leave
the timer green rather than turning it red every 15 minutes.
"""
class BrokenRemote(FakeRemote):
def list_directory(self, path: str) -> list[str]:
msg = f"something nobody predicted, at {path}"
raise RuntimeError(msg)
monkeypatch.setattr(
"sickdays._sync.firebase_client_for", lambda _app: BrokenRemote()
)
with caplog.at_level("ERROR"):
assert sync_quietly() is False
assert "failed unexpectedly" in caplog.text
assert "RuntimeError" in caplog.text
def test_a_removal_on_one_device_reaches_the_other() -> None:
"""The point of the state flip: un-marking survives a merge."""
remote = FakeRemote()
add_sick_day(LATER, now=TODAY)
sync(client=remote)
pushed = next(iter(remote.files))
# The peer pushed the same log; then this device removes the day.
remote.files["sickdays-sync/devices/phone/sick_days.json"] = remote.files[pushed]
remove_sick_day(LATER, now=TODAY)
sync(client=remote)
assert not is_sick_day(LATER)

19
sickdays/systemd/sickdays-sync.service Normal file
View File

@ -0,0 +1,19 @@
[Unit]
Description=Pull and push the shared sick-day pool
Documentation=file:///home/kuhy/src/utils/sickdays/README.md
# Without a network this exits 0 having done nothing, by design -- the gates
# read the last-merged local file and a stale pool is never a reason to fail.
After=network-online.target
[Service]
Type=oneshot
# `sync-quiet` rather than `sync`: this unit must never report failure just
# because the machine is offline or the credential has not been set up yet.
# A real bug still shows up in the journal via the library's own logging.
ExecStart=%h/.local/share/sickdays/venv/bin/sickdays sync-quiet
# A hung network call must not leave a unit running until the next tick.
TimeoutStartSec=120
[Install]
WantedBy=default.target

19
sickdays/systemd/sickdays-sync.timer Normal file
View File

@ -0,0 +1,19 @@
[Unit]
Description=Sync the shared sick-day pool periodically
Documentation=file:///home/kuhy/src/utils/sickdays/README.md
[Timer]
# Every 15 minutes. A sick day marked on the phone should reach the PC well
# before the next gate fires, and the gates that run most often (wake-alarm
# every minute, the lockers every 5) read the *local* file, so this interval
# bounds staleness rather than gating anything.
OnBootSec=2min
OnUnitActiveSec=15min
# A laptop that was asleep at the scheduled tick should sync on waking, not
# silently wait another 15 minutes -- the overnight case is exactly when a
# day gets marked from bed and needs to be there by morning.
Persistent=true
[Install]
WantedBy=timers.target