gatelock 0.9.0: morning_session, the one reader of wake-alarm's carrot
Some checks failed
dependency freshness / dependencies on newest stable (push) Waiting to run
file length / file-length (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
palette drift / palette (push) Waiting to run
shared gate tests / gate unit tests (push) Waiting to run
gatelock pre-commit / pre-commit (push) Has been cancelled
gatelock tests / test (push) Has been cancelled

screen-locker and leetcode-guard each carried a near-identical copy of the
reader for wake-alarm's signed morning_session.json; book-guard is about to
need it too. The policy stays in wake-alarm -- this only answers "signed,
dated today, now < exempt_until". Path and retry budget are arguments so
each consumer's test redirect keeps working.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FBbkN4ZB4aUPJPzR2prRan
This commit is contained in:
Krzysztof Rudnicki 2026-10-02 13:59:33 +02:00
parent 694c013b6b
commit 60fba8a738
4 changed files with 312 additions and 1 deletions

View File

@ -65,6 +65,13 @@ overridden as a unit.
projects; `DEFAULT_HMAC_KEY_FILE` (`/etc/workout-locker/hmac.key`) is
unchanged so existing signed history keeps verifying.
`gatelock.morning_session.morning_skip` is the one reader of wake-alarm's
signed `morning_session.json` (the wake-early carrot: no lock until the signed
`exempt_until`, 11:00 for a morning completed in time). screen-locker,
leetcode-guard and book-guard all call it; the policy itself stays in
wake-alarm. Path and retry budget are arguments so each consumer's tests keep
redirecting their own constant.
## Development
```bash

View File

@ -0,0 +1,149 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""Read the morning-session carrot wake-alarm signed for today.
The phone's morning (weigh-in, shower, dress, desk) is published to the PC
and distilled by wake-alarm into ``morning_session.json``, HMAC-signed, always
dated today, with an ``exempt_until`` while the morning is live or once it was
completed in time (11:00). Every gate that honours the carrot reads it here.
This module never re-derives that policy: the whole decision is *signature
ok, dated today, now < exempt_until*. Missing, stale, failed or tampered all
mean "no skip" -- the gate runs as it always has.
A run deferral, never ledger state: nothing is written, so a deleted file
cannot mint an unlocked day.
The one wrinkle is a PC booted mid-morning: the file is missing or yesterday's
until wake-alarm's ``Persistent=true`` timer catches up, which needs the
network. The arming run waits for that, bounded; status paths never do.
The path and the retry budget are parameters, not module state, so each
consumer's test suite keeps redirecting its own constant.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import UTC, datetime
import json
import logging
from pathlib import Path
import time
from typing import TYPE_CHECKING, Final
from gatelock.log_integrity import verify_entry_hmac
if TYPE_CHECKING:
from collections.abc import Callable
_logger = logging.getLogger(__name__)
MORNING_SESSION_FILE: Final = (
Path.home() / ".local" / "state" / "wake_alarm" / "morning_session.json"
)
"""Where wake-alarm writes it. Keep in step with ``wake_alarm/_constants.py``."""
# Only inside this local window is a non-today file worth waiting for: the
# refresher runs 05:00-11:00, so outside it "not today's" is simply the truth.
MORNING_WINDOW: Final = ((5, 0), (11, 0))
MORNING_RETRY_SECONDS: Final = 30.0
MORNING_POLL_SECONDS: Final = 5.0
@dataclass(frozen=True)
class MorningSkip:
"""A skip the morning session earned: what the phone said, and until when."""
outcome: str
exempt_until: datetime
def __str__(self) -> str:
"""Say which morning earned it and when the gate comes back."""
return f"{self.outcome} session, no lock until {self.exempt_until:%H:%M}"
def _read_verified(path: Path) -> dict[str, object] | None:
"""The file's entry when it exists, parses and verifies; else None.
Missing is debug (it is the normal state outside the morning); anything
else is a warning, because it means the refresher or the key is broken.
"""
if not path.exists():
_logger.debug("No morning session file at %s", path)
return None
try:
entry = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as exc:
_logger.warning("Cannot read %s: %s", path, exc)
return None
if not isinstance(entry, dict) or not verify_entry_hmac(entry):
_logger.warning("Morning session file is not a signed object")
return None
return entry
def _load_today(path: Path, now: datetime) -> dict[str, object] | None:
"""Today's verified entry, or None. ``now`` is already local."""
entry = _read_verified(path)
if entry is None or entry.get("date") != now.date().isoformat():
return None
return entry
def _skip_from(entry: dict[str, object], now: datetime) -> MorningSkip | None:
"""The skip an entry grants at ``now``, if its window is still open."""
until = entry.get("exempt_until")
if not isinstance(until, str):
return None
try:
exempt_until = datetime.fromisoformat(until)
except ValueError:
_logger.warning("Morning session exempt_until unreadable: %r", until)
return None
if exempt_until.tzinfo is None or now >= exempt_until:
# A naive instant cannot be compared with a local one, and guessing
# its zone would be inventing a carrot the signer never granted.
return None
return MorningSkip(outcome=str(entry.get("outcome")), exempt_until=exempt_until)
def _in_window(now: datetime) -> bool:
(start_h, start_m), (end_h, end_m) = MORNING_WINDOW
minutes = now.hour * 60 + now.minute
return start_h * 60 + start_m <= minutes < end_h * 60 + end_m
def _local_now() -> datetime:
return datetime.now(tz=UTC).astimezone()
def morning_skip(
path: Path = MORNING_SESSION_FILE,
now: datetime | None = None,
*,
wait: bool,
retry_seconds: float = MORNING_RETRY_SECONDS,
sleep: Callable[[float], None] = time.sleep,
) -> MorningSkip | None:
"""The skip the morning session earned at ``now``, or None.
With ``wait`` (the arming path only), a file that is not today's during
the morning window is retried for up to ``retry_seconds`` so a freshly
booted PC gives wake-alarm's catch-up run a chance to land.
"""
now = (now or _local_now()).astimezone()
entry = _load_today(path, now)
waited = 0.0
while entry is None and wait and _in_window(now) and waited < retry_seconds:
sleep(MORNING_POLL_SECONDS)
waited += MORNING_POLL_SECONDS
now = _local_now()
entry = _load_today(path, now)
if entry is None:
if waited:
_logger.warning(
"No morning session for today after %.0fs; "
"is wake-alarm-session.timer running?",
waited,
)
return None
return _skip_from(entry, now)

View File

@ -0,0 +1,155 @@
# Copyright (c) 2026 Krzysztof Rudnicki
"""gatelock.morning_session -- the carrot file, read and never re-derived."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import json
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from gatelock import morning_session
from gatelock.log_integrity import compute_entry_hmac
from gatelock.morning_session import MorningSkip, morning_skip
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
NOW = datetime(2026, 9, 19, 9, 0).astimezone()
@pytest.fixture
def path(tmp_path: Path) -> Path:
"""Where the tests write the signed file -- never the real one."""
return tmp_path / "morning_session.json"
@pytest.fixture(autouse=True)
def _signing_key(tmp_path: Path) -> Iterator[None]:
key = tmp_path / "hmac.key"
key.write_bytes(b"1" * 32)
with patch("gatelock.log_integrity.DEFAULT_HMAC_KEY_FILE", key):
yield
def _write(path: Path, entry: dict[str, object], *, sign: bool = True) -> None:
"""Write an entry; signed unless told otherwise."""
if sign:
entry["hmac"] = compute_entry_hmac(entry)
path.write_text(json.dumps(entry))
def _entry(
outcome: str = "completed",
exempt_until: str | None = "11:00",
date: str = "2026-09-19",
) -> dict[str, object]:
entry: dict[str, object] = {"date": date, "outcome": outcome}
if exempt_until is not None:
entry["exempt_until"] = (
datetime.fromisoformat(f"{date}T{exempt_until}").astimezone().isoformat()
)
return entry
def _today_entry() -> dict[str, object]:
today = datetime.now(tz=UTC).astimezone().strftime("%Y-%m-%d")
return _entry(date=today, exempt_until="23:59")
class TestVerdict:
"""One rule: signed, today, not yet expired."""
def test_completed_grants_until_the_signed_instant(self, path: Path) -> None:
_write(path, _entry())
skip = morning_skip(path, NOW, wait=False)
assert skip == MorningSkip("completed", NOW.replace(hour=11))
assert str(skip) == "completed session, no lock until 11:00"
def test_expired_and_absent_exemptions_grant_nothing(self, path: Path) -> None:
_write(path, _entry())
assert morning_skip(path, NOW + timedelta(hours=3), wait=False) is None
_write(path, _entry("failed", exempt_until=None))
assert morning_skip(path, NOW, wait=False) is None
_write(path, {**_entry(), "exempt_until": "later"})
assert morning_skip(path, NOW, wait=False) is None
def test_a_naive_exempt_until_grants_nothing(self, path: Path) -> None:
_write(path, {**_entry(), "exempt_until": "2026-09-19T11:00:00"})
assert morning_skip(path, NOW, wait=False) is None
def test_yesterdays_file_is_not_today(self, path: Path) -> None:
_write(path, _entry(date="2026-09-18"))
assert morning_skip(path, NOW, wait=False) is None
def test_missing_unreadable_tampered_are_all_nothing(self, path: Path) -> None:
assert morning_skip(path, NOW, wait=False) is None
_write(path, _entry(), sign=False)
assert morning_skip(path, NOW, wait=False) is None
path.write_text("[1]")
assert morning_skip(path, NOW, wait=False) is None
path.write_text("{")
assert morning_skip(path, NOW, wait=False) is None
path.unlink()
path.mkdir()
assert morning_skip(path, NOW, wait=False) is None
def test_default_now_is_the_wall_clock(self, path: Path) -> None:
_write(path, _today_entry())
assert morning_skip(path, wait=False) is not None
def test_default_path_is_wake_alarms_state_file(self) -> None:
assert morning_session.MORNING_SESSION_FILE.parts[-3:] == (
"state",
"wake_alarm",
"morning_session.json",
)
class TestBootRetry:
"""The arming path waits for the refresher, bounded; status never does."""
def test_waits_inside_the_window_then_warns(
self, path: Path, caplog: pytest.LogCaptureFixture
) -> None:
naps: list[float] = []
# The window is re-checked against the real clock after each nap.
with patch.object(morning_session, "MORNING_WINDOW", ((0, 0), (23, 59))):
skip = morning_skip(
path, NOW, wait=True, retry_seconds=10.0, sleep=naps.append
)
assert skip is None
assert naps == [5.0, 5.0]
assert "is wake-alarm-session.timer running" in caplog.text
def test_a_file_landing_mid_wait_is_honoured(self, path: Path) -> None:
def land(_seconds: float) -> None:
_write(path, _today_entry())
with patch.object(morning_session, "MORNING_WINDOW", ((0, 0), (23, 59))):
skip = morning_skip(path, NOW, wait=True, retry_seconds=10.0, sleep=land)
assert skip is not None
def test_never_waits_outside_the_window_or_without_wait(self, path: Path) -> None:
naps: list[float] = []
late = NOW.replace(hour=14)
assert (
morning_skip(path, late, wait=True, retry_seconds=10.0, sleep=naps.append)
is None
)
assert (
morning_skip(path, NOW, wait=False, retry_seconds=10.0, sleep=naps.append)
is None
)
assert naps == []
def test_a_zero_budget_never_waits(self, path: Path) -> None:
naps: list[float] = []
assert (
morning_skip(path, NOW, wait=True, retry_seconds=0.0, sleep=naps.append)
is None
)
assert naps == []

View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "gatelock"
version = "0.8.2"
version = "0.9.0"
description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps"
requires-python = ">=3.14"
dependencies = [] # pure stdlib — tkinter is bundled with Python