From e5db25468d55d9fdc038728ce8789a7e1dabb5e5 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 17 Aug 2026 16:47:25 +0200 Subject: [PATCH] Re-arm the locker, and make a disarmed enforcer impossible to miss The screen locker stopped locking on 2026-08-04 and nobody could tell. Not a logic bug: every predicate worked the whole time. systemd logged timers.target: Found ordering cycle: early-bird-workout-check.timer/start after graphical-session.target/start after basic.target/start after timers.target/start - after early-bird-workout-check.timer timers.target: Job early-bird-workout-check.timer/start deleted to break ordering cycle starting with timers.target/start and left the timer disabled. A timer that is WantedBy=timers.target must not be ordered After= a target that itself comes after timers.target; systemd resolves that loop by deleting the job, which is an absence, and absences page nobody. That timer is the ONLY thing that closes the early-bird window. Each morning a run inside the window banked early_bird_pending.json, logged "will re-check at 08:30", and exited 0. The re-check never came, so the marker stood all day, every day, for thirteen days. It stayed invisible because logging.basicConfig was called only in the --sync-only branch, never for --production -- the mode systemd actually runs. Journal proof: zero INFO lines from workout-locker.service since June. Every early exit logged at INFO into a logger with no handler. Meanwhile the healthy 15-minute sync kept the system looking alive. The 2026-08-15 Python split is NOT implicated: it was pure code motion, and basicConfig already sat in that branch before it. Fixes: * Drop the After= that closes the cycle; Persistent=true so a machine asleep at 08:30 still re-checks after resume. * Add workout-locker.timer (*:00,30). The service is WantedBy= graphical-session.target, i.e. one decision per login, so the 08-16 -> 08-17 day and ISO-week rollover inside one long session was never re-evaluated. * arm.sh: install units, enable timers, verify -- one command, because re-arming was a four-command sequence that was easy to half-do, and `systemctl enable` returns 0 while systemd deletes the timer job. * check_enforcement_armed.py asks systemd what it actually scheduled (enabled AND has a next trigger) and fails loudly otherwise. Wired as ExecStartPost on workout-sync.service, so a disarmed locker marks that unit failed every 15 minutes instead of scrolling past. install_systemd.sh now fails on it too. Logging (the reason this went unnoticed, so treated as a defect in its own right): * basicConfig for every mode, never one branch. * Every enforcement-abandoning exit records a decision instead of an invisible INFO: one greppable line -- DECISION lock=no reason=early_bird_window_active weekly=0/5 -- at WARNING, since per CLAUDE.md an INFO explanation of a non-action is a silent failure with extra steps. The enforced case is logged too: "no line at all" was the outage's whole signature. * A durable JSONL trail outside the repo (~/.local/share/screen_locker/), so "when did this last actually enforce?" survives journal rotation. --status and --sync-only record lock=n/a; "did not decide" must never look like "decided not to lock". * check_silent_failures.py now also catches sys.exit(0) that explains nothing. Verified both directions: it flags 3/3 exits in the pre-fix _auto_upgrade.py and passes the repo clean. Sync noise: four records replayed as WARNINGs every cycle (~384/day), which is how the actionable lines got tuned out. manual_workout:2026-08-15 is an empty stub ({date, kind, type}) -- no workout to recover, and that day's credit was already counted under a different sync_record_id, so no credit was ever lost. The three "collided" records are dedup working. Both now report at info; a WARNING from this logger again means something real. 4 -> 0 per cycle. Verified on the live system: the periodic timer fired at 16:45 and the journal shows `DECISION lock=no reason=scheduled_skip_day weekly=0/5` -- the line that did not exist for thirteen days. Both timers report enabled and scheduled, and no ordering cycle recurs. Today is a scheduled skip so the machine was not locked mid-session; enforcement resumes tomorrow 08:30. 1007 tests, 100% coverage, all hooks green. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01F2b6ZjBcdrkSYnMujx4mWd --- .pre-commit-config.yaml | 2 +- arm.sh | 116 +++++++++ early-bird-workout-check.timer | 14 +- install_systemd.sh | 19 ++ screen_locker/_auto_upgrade.py | 54 ++++- screen_locker/_cli.py | 19 +- screen_locker/_decision_log.py | 227 ++++++++++++++++++ screen_locker/_manual_sync.py | 76 +++++- screen_locker/_startup_checks.py | 63 ++++- screen_locker/_status.py | 4 + screen_locker/screen_lock.py | 4 + screen_locker/tests/conftest.py | 4 + screen_locker/tests/test_decision_log.py | 213 ++++++++++++++++ screen_locker/tests/test_manual_sync_part3.py | 50 ++++ screen_locker/tests/test_systemd_units.py | 149 ++++++++++++ scripts/add_scheduled_skip.py | 113 +++++++++ scripts/check_enforcement_armed.py | 184 ++++++++++++++ scripts/check_silent_failures.py | 99 +++++++- workout-locker.timer | 28 +++ workout-sync.service | 7 + 20 files changed, 1411 insertions(+), 34 deletions(-) create mode 100755 arm.sh create mode 100644 screen_locker/_decision_log.py create mode 100644 screen_locker/tests/test_decision_log.py create mode 100644 screen_locker/tests/test_systemd_units.py create mode 100755 scripts/add_scheduled_skip.py create mode 100755 scripts/check_enforcement_armed.py create mode 100644 workout-locker.timer diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 691cffc..63f9c08 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -192,7 +192,7 @@ repos: # the gate it was being refactored to satisfy. types_or: [python, shell, dart] - id: no-silent-failures - name: Block silent failures (swallowed exceptions must log why) + name: Block silent failures (swallowed exceptions and silent sys.exit(0) must say why) entry: python3 scripts/check_silent_failures.py language: system types: [python] diff --git a/arm.sh b/arm.sh new file mode 100755 index 0000000..47cc0e7 --- /dev/null +++ b/arm.sh @@ -0,0 +1,116 @@ +#!/bin/bash + +# ============================================================================ +# Arm the workout screen locker — install units, enable timers, verify. +# +# One command, because re-arming used to be a four-command sequence that was +# easy to half-do. On 2026-08-04 a systemd ordering cycle silently disabled +# early-bird-workout-check.timer and enforcement stopped for thirteen days with +# nothing on the machine saying so; `systemctl enable` alone would NOT have +# caught that, because enable "succeeds" while systemd deletes the timer job. +# So this script always finishes by asking systemd what it actually scheduled, +# and fails loudly when the answer is "nothing". +# +# Usage: +# ./arm.sh # install, enable, verify +# ./arm.sh --skip-today # ...but do not lock today (adds a scheduled skip) +# ./arm.sh --status # verify only; change nothing +# ============================================================================ + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +readonly SCRIPT_DIR +readonly USER_UNIT_DIR="$HOME/.config/systemd/user" + +# Units that must be on disk for enforcement to be schedulable. +readonly UNITS=( + "workout-locker.service" + "workout-locker.timer" + "early-bird-workout-check.timer" + "workout-sync.service" + "workout-sync.timer" +) + +# Only the timers get enabled; the service is triggered by them and by login. +readonly TIMERS=( + "early-bird-workout-check.timer" + "workout-locker.timer" + "workout-sync.timer" +) + +SKIP_TODAY=0 +STATUS_ONLY=0 + +usage() { + echo "Usage: $(basename "$0") [--skip-today] [--status]" + echo " --skip-today Arm the locker but exempt today (no lock until tomorrow)" + echo " --status Report whether enforcement is armed; change nothing" + exit 0 +} + +verify_armed() { + python3 "$SCRIPT_DIR/scripts/check_enforcement_armed.py" +} + +add_skip_for_today() { + # Delegated to Python rather than hand-editing JSON in bash: the file is + # the same one the lock chain reads, and a malformed write would be read as + # "not a skip" and lock anyway. + python3 "$SCRIPT_DIR/scripts/add_scheduled_skip.py" --date today +} + +main() { + if ((STATUS_ONLY)); then + verify_armed + return + fi + + echo "Installing units into $USER_UNIT_DIR" + mkdir -p "$USER_UNIT_DIR" + for unit in "${UNITS[@]}"; do + cp "$SCRIPT_DIR/$unit" "$USER_UNIT_DIR/$unit" + done + + systemctl --user daemon-reload + + if ((SKIP_TODAY)); then + # Register the exemption BEFORE the timers can fire, so arming can + # never race the first re-check and lock the machine anyway. + add_skip_for_today + fi + + echo "Enabling timers" + systemctl --user enable --now "${TIMERS[@]}" + + # The whole point: `enable` returning 0 is not proof of anything. + echo "" + verify_armed + echo "" + echo "✓ Screen locker armed." + if ((SKIP_TODAY)); then + echo " Today is exempt — enforcement begins tomorrow." + fi +} + +while [[ $# -gt 0 ]]; do + case $1 in + --skip-today) + SKIP_TODAY=1 + shift + ;; + --status) + STATUS_ONLY=1 + shift + ;; + -h | --help) + usage + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +main "$@" diff --git a/early-bird-workout-check.timer b/early-bird-workout-check.timer index 47fa3f7..0350f09 100644 --- a/early-bird-workout-check.timer +++ b/early-bird-workout-check.timer @@ -1,6 +1,13 @@ [Unit] Description=Re-check workout after early bird grace period expires (08:30 normal, 09:05 extended) -After=graphical-session.target +# NO After=graphical-session.target here. This timer is WantedBy=timers.target, +# and timers.target is ordered before basic.target before +# graphical-session.target -- so that After= closes a dependency cycle. +# systemd resolves such a cycle by DELETING this timer's job, which is exactly +# what happened on 2026-08-04 ("Job early-bird-workout-check.timer/start deleted +# to break ordering cycle"). The timer went disabled and the locker silently +# stopped enforcing for days. The service it triggers keeps its own +# After=graphical-session.target, which is where that ordering belongs. [Timer] # Fires at 08:30 for the normal early-bird window (5:00–8:30). @@ -9,7 +16,10 @@ After=graphical-session.target OnCalendar=*-*-* 08:30:00 OnCalendar=*-*-* 09:05:00 Unit=workout-locker.service -Persistent=false +# Persistent=true: a machine asleep or off at 08:30 would otherwise skip the +# re-check entirely, leaving the early-bird marker banked and the day +# unenforced. Both runs are idempotent (see above), so a catch-up run is safe. +Persistent=true AccuracySec=1s RandomizedDelaySec=0 diff --git a/install_systemd.sh b/install_systemd.sh index 86b2697..0b1deba 100755 --- a/install_systemd.sh +++ b/install_systemd.sh @@ -4,11 +4,13 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" SERVICE_FILE="$SCRIPT_DIR/workout-locker.service" EARLY_BIRD_TIMER_FILE="$SCRIPT_DIR/early-bird-workout-check.timer" +LOCKER_TIMER_FILE="$SCRIPT_DIR/workout-locker.timer" SYNC_SERVICE_FILE="$SCRIPT_DIR/workout-sync.service" SYNC_TIMER_FILE="$SCRIPT_DIR/workout-sync.timer" USER_SERVICE_DIR="$HOME/.config/systemd/user" SERVICE_NAME="workout-locker.service" EARLY_BIRD_TIMER_NAME="early-bird-workout-check.timer" +LOCKER_TIMER_NAME="workout-locker.timer" SYNC_SERVICE_NAME="workout-sync.service" SYNC_TIMER_NAME="workout-sync.timer" @@ -48,6 +50,11 @@ cp "$SERVICE_FILE" "$USER_SERVICE_DIR/$SERVICE_NAME" # Copy early bird timer cp "$EARLY_BIRD_TIMER_FILE" "$USER_SERVICE_DIR/$EARLY_BIRD_TIMER_NAME" +# Copy the periodic re-check timer. Without it the locker only ever decides at +# login and at 08:30/09:05, so a day or ISO-week boundary crossed inside a long +# session is never re-evaluated. +cp "$LOCKER_TIMER_FILE" "$USER_SERVICE_DIR/$LOCKER_TIMER_NAME" + # Copy the periodic workout-sync units. Without these, syncing happens only # once per locker start, so a workout finished after login is not seen until # the next login. @@ -69,11 +76,23 @@ systemctl --user enable "$SERVICE_NAME" # Enable the early bird re-check timer systemctl --user enable --now "$EARLY_BIRD_TIMER_NAME" +# Enable the periodic re-check timer +systemctl --user enable --now "$LOCKER_TIMER_NAME" + # Enable the periodic workout sync systemctl --user enable --now "$SYNC_TIMER_NAME" +# Verify enforcement is actually armed. enable can silently no-op when systemd +# breaks an ordering cycle by deleting the timer's job -- exactly what happened +# on 2026-08-04 -- so a successful `enable` is NOT proof the locker will run. +if ! python3 "$SCRIPT_DIR/scripts/check_enforcement_armed.py"; then + echo "✗ Enforcement is NOT armed after install — see the errors above" >&2 + exit 1 +fi + echo "✓ Workout locker service installed" echo "✓ Early bird re-check timer installed (fires daily at 08:30)" +echo "✓ Periodic re-check timer installed (every 30 min)" echo "✓ Service will start automatically on next login" echo "" echo "To start now: systemctl --user start workout-locker" diff --git a/screen_locker/_auto_upgrade.py b/screen_locker/_auto_upgrade.py index e75bac8..5845e78 100644 --- a/screen_locker/_auto_upgrade.py +++ b/screen_locker/_auto_upgrade.py @@ -12,11 +12,26 @@ import logging import sys from screen_locker import _sick_tracker +from screen_locker._decision_log import LockDecision, record_decision from screen_locker._wake_state import has_workout_skip_today _logger = logging.getLogger(__name__) +def _skip(reason: str, detail: str, **extra: object) -> None: + """Record a decision NOT to enforce, then exit. + + Every branch that abandons enforcement goes through here. Before this + existed each one just logged at INFO and called ``sys.exit(0)``, which is + how the locker managed to stop enforcing for thirteen days without leaving + a single line saying so. + """ + record_decision( + LockDecision(locked=False, reason=reason, detail=detail, extra=extra) + ) + sys.exit(0) + + class AutoUpgradeMixin: """Handles today-state detection and silent log-entry upgrading. @@ -32,8 +47,7 @@ class AutoUpgradeMixin: """Check startup conditions and exit early when appropriate.""" if verify_only: if not self._is_sick_day_today(): - _logger.info("No sick day logged today. Nothing to verify.") - sys.exit(0) + _skip("no_sick_day_to_verify", "No sick day logged today.") return self._check_non_verify_exits() @@ -41,27 +55,45 @@ class AutoUpgradeMixin: """Handle early-bird and today's log states. Return True to stop startup.""" if self._is_early_bird_pending() and not self._is_early_bird_time(): if self._try_auto_upgrade_early_bird(): - _logger.info("Auto-upgraded early_bird entry to phone_verified.") - sys.exit(0) + _skip( + "early_bird_auto_upgraded", + "Auto-upgraded early_bird entry to phone_verified.", + ) return True return False # Expired early bird, upgrade unavailable — full lock. if self._is_early_bird_pending(): - _logger.info("Early bird window still active — skipping lock.") + # The ONLY thing that closes this window is + # early-bird-workout-check.timer. When that timer was disabled by an + # ordering cycle (2026-08-04) this branch silently deferred the lock + # every single day, forever. + _skip( + "early_bird_window_active", + "Early bird window still active — deferring to the 08:30/09:05 " + "re-check timer.", + recheck_by="early-bird-workout-check.timer", + ) elif self._is_sick_day_today(): if self._try_auto_upgrade_sick_day(): - _logger.info("Auto-upgraded today's sick_day entry to phone_verified.") + _skip( + "sick_day_auto_upgraded", + "Auto-upgraded today's sick_day entry to phone_verified.", + ) else: - _logger.info("Sick day already logged today.") + _skip("sick_day", "Sick day already logged today.") elif self.has_logged_today(): - _logger.info("Workout already logged today. Skipping screen lock.") + _skip("workout_logged_today", "Workout already logged today.") elif has_workout_skip_today(): - _logger.info("Wake alarm earned workout skip. Skipping screen lock.") + _skip("wake_alarm_skip", "Wake alarm earned a workout skip.") elif self._is_early_bird_time(): self._save_early_bird_pending() - _logger.info("Early bird time — skipping lock, will re-check at 08:30.") + _skip( + "early_bird_banked", + "Early bird time — banked the pending marker, will re-check " + "at 08:30/09:05.", + recheck_by="early-bird-workout-check.timer", + ) else: return False - sys.exit(0) return True def _try_auto_upgrade_sick_day(self) -> bool: diff --git a/screen_locker/_cli.py b/screen_locker/_cli.py index 1a83ce6..0d5c9c2 100644 --- a/screen_locker/_cli.py +++ b/screen_locker/_cli.py @@ -13,6 +13,7 @@ from pathlib import Path import sys from typing import TYPE_CHECKING +from screen_locker._decision_log import record_no_decision from screen_locker._status import run_status if TYPE_CHECKING: @@ -38,6 +39,16 @@ def _headless_locker(locker_cls: type[ScreenLocker]) -> ScreenLocker: def main(locker_cls: type[ScreenLocker], argv: list[str]) -> None: """Dispatch on ``argv`` and run the requested mode.""" + # Configure logging for EVERY mode, not just --sync-only. This used to sit + # inside the --sync-only branch, so `--production` -- the mode systemd + # actually runs -- had no handler and fell back to lastResort, which drops + # everything below WARNING. The result: zero INFO lines from + # workout-locker.service since June, and a thirteen-day enforcement outage + # that left no trace at all. Never narrow this back to one branch. + logging.basicConfig( + level=logging.INFO, + format="%(levelname)s %(name)s: %(message)s", + ) if "--status" in argv: run_status(_headless_locker(locker_cls)) @@ -46,11 +57,11 @@ def main(locker_cls: type[ScreenLocker], argv: list[str]) -> None: # apply their credit, with no Tk and no lock screen. Sync used to run # ONLY at process start, so a workout finished after login was not seen # until the next login — the timer closes that window. - logging.basicConfig( - level=logging.INFO, - format="%(levelname)s %(name)s: %(message)s", - ) _headless_locker(locker_cls).sync_now() + # Say so explicitly: --sync-only NEVER locks, so a reader tracing "why + # didn't it lock?" through the journal must not mistake a sync run for + # an enforcement run that decided to skip. + record_no_decision("--sync-only") sys.exit(0) locker = locker_cls( diff --git a/screen_locker/_decision_log.py b/screen_locker/_decision_log.py new file mode 100644 index 0000000..0aabe90 --- /dev/null +++ b/screen_locker/_decision_log.py @@ -0,0 +1,227 @@ +"""One structured, durable record of every lock decision the locker makes. + +This module exists because of a multi-day outage that produced *no evidence*. +Between 2026-08-04 and 2026-08-17 the locker stopped enforcing, and the only +trace in the journal was its absence: ``workout-locker.service`` started, ran +for six seconds, exited ``0/SUCCESS``, and said nothing about why. Every +early-exit reason was logged at ``INFO``, and ``logging.basicConfig`` was only +ever called on the ``--sync-only`` path, so ``--production`` emitted no INFO at +all -- zero such lines since June. + +Two things follow from that, and this module provides both: + +* **A single grep-able line per run.** ``DECISION lock=no reason=... weekly=0/5`` + is one line a human or an agent can find without knowing the code, and it is + emitted at ``WARNING`` when enforcement was skipped, because per CLAUDE.md an + ``INFO``-level explanation of a non-action is a silent failure with extra + steps. +* **A durable JSONL trail.** The journal rotates and is per-boot; a decision + history that outlives it is what makes "when did this last actually enforce?" + answerable in one command instead of a forensic session. + +The file is deliberately outside the repo (see ``DECISION_LOG_FILE``): the +2026-07 privacy incident purged workout state JSONs from git history, and they +must never be re-tracked. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +import json +import logging +from pathlib import Path +from typing import Final + +__all__ = [ + "DECISION_LOG_FILE", + "DECISION_LOG_MAX_ENTRIES", + "LockDecision", + "read_decisions", + "record_decision", +] + +_logger = logging.getLogger(__name__) + +# Outside the repo on purpose -- workout state is private and git-untracked. +DECISION_LOG_FILE: Path = ( + Path.home() / ".local" / "share" / "screen_locker" / "decisions.jsonl" +) + +# Trimmed on write so the recurring timers cannot grow this without bound. +# Writers are the 30-minute locker timer plus the 15-minute sync run (which +# records "made no decision"), i.e. ~100 lines/day -> roughly a month of +# history. That comfortably covers the 13-day blind spot this file exists to +# make impossible. +DECISION_LOG_MAX_ENTRIES: Final[int] = 3000 + + +@dataclass(frozen=True) +class LockDecision: + """Why the locker did or did not enforce on one run. + + ``reason`` is a stable machine-readable slug (``early_bird_pending``, + ``weekly_minimum_met``, ``enforced``...), not prose, so it can be grepped + and counted across runs. ``detail`` carries the human sentence. + """ + + locked: bool + reason: str + detail: str = "" + weekly_count: int | None = None + weekly_required: int | None = None + extra: dict[str, object] = field(default_factory=dict) + + def as_line(self) -> str: + """Render the single grep-able summary line. + + Keys are ``key=value`` pairs so the line stays readable by eye and + parseable by ``awk``/``grep -o`` without a JSON decoder. + """ + parts = [ + "DECISION", + f"lock={'yes' if self.locked else 'no'}", + f"reason={self.reason}", + ] + if self.weekly_required is not None: + parts.append(f"weekly={self.weekly_count}/{self.weekly_required}") + for key, value in self.extra.items(): + parts.append(f"{key}={value}") + if self.detail: + parts.append(f"detail={self.detail!r}") + return " ".join(parts) + + def as_record(self) -> dict[str, object]: + """Render the durable JSON object written to the trail.""" + record: dict[str, object] = { + "timestamp": datetime.now(tz=timezone.utc).isoformat(), + "locked": self.locked, + "reason": self.reason, + } + if self.detail: + record["detail"] = self.detail + if self.weekly_required is not None: + record["weekly_count"] = self.weekly_count + record["weekly_required"] = self.weekly_required + record.update(self.extra) + return record + + +def _trimmed(lines: list[str], new_line: str) -> list[str]: + """Return the retained history with ``new_line`` appended.""" + if len(lines) >= DECISION_LOG_MAX_ENTRIES: + lines = lines[-(DECISION_LOG_MAX_ENTRIES - 1) :] + return [*lines, new_line] + + +def record_decision(decision: LockDecision, *, log_file: Path | None = None) -> None: + """Log ``decision`` to the journal and append it to the durable trail. + + Never raises: a failure to *record* why the locker acted must not stop the + locker from acting. A write failure is reported at ``warning`` rather than + swallowed, so the recording gap is itself visible. + """ + line = decision.as_line() + # Not-enforcing is the state that hid the outage, so it is the state that + # gets the louder level. An enforced lock is self-evident on screen. + if decision.locked: + _logger.info("%s", line) + else: + _logger.warning("%s", line) + + _append_record(decision.as_record(), log_file=log_file) + + +def _append_record(record: dict[str, object], *, log_file: Path | None = None) -> None: + """Append one record to the durable trail, trimming old history. + + Never raises: failing to *record* why the locker acted must not stop the + locker from acting. A write failure is reported at ``warning`` rather than + swallowed, so the gap in the trail is itself visible. + """ + target = DECISION_LOG_FILE if log_file is None else log_file + try: + target.parent.mkdir(parents=True, exist_ok=True) + existing = ( + target.read_text(encoding="utf-8").splitlines() if target.exists() else [] + ) + kept = _trimmed( + [entry for entry in existing if entry.strip()], + json.dumps(record), + ) + target.write_text("\n".join(kept) + "\n", encoding="utf-8") + except OSError as exc: + _logger.warning( + "Could not append to the decision log at %s (%s) — this run's " + "decision is in the journal only, so the durable trail now has a " + "gap", + target, + exc, + ) + + +def record_no_decision(mode: str, *, log_file: Path | None = None) -> None: + """Note that a run finished WITHOUT evaluating the lock at all. + + ``--sync-only`` and ``--status`` never reach the decision ladder. Recording + that explicitly is what keeps the trail honest: during the outage the + journal was full of healthy 15-minute sync runs, which made the system look + alive while no enforcement decision had been made for days. "This run did + not decide" and "this run decided not to lock" must never look alike. + + Written at ``info`` because these modes run constantly and are expected; + the durable trail is what makes them auditable, not the log level. + """ + _logger.info( + "DECISION lock=n/a reason=mode_makes_no_decision mode=%s — this mode " + "never evaluates the lock", + mode, + ) + _append_record( + { + "timestamp": datetime.now(tz=timezone.utc).isoformat(), + "locked": None, + "reason": "mode_makes_no_decision", + "mode": mode, + }, + log_file=log_file, + ) + + +def read_decisions(*, log_file: Path | None = None) -> list[dict[str, object]]: + """Return the recorded decisions, oldest first. + + Corrupt lines are skipped with a warning rather than aborting the read: a + partially damaged trail is still worth far more than no trail. + """ + target = DECISION_LOG_FILE if log_file is None else log_file + try: + raw = target.read_text(encoding="utf-8") + except FileNotFoundError: + # Genuinely benign and expected before the first run ever records a + # decision, but still said out loud: "no history" and "history I could + # not read" must never look the same to whoever is investigating. + _logger.warning( + "No decision log at %s yet — no lock decision has been recorded " + "on this machine (this is normal before the first locker run)", + target, + ) + return [] + except OSError as exc: + _logger.warning("Could not read the decision log at %s (%s)", target, exc) + return [] + + records: list[dict[str, object]] = [] + for line in raw.splitlines(): + if not line.strip(): + continue + try: + parsed = json.loads(line) + except json.JSONDecodeError as exc: + _logger.warning("Skipping a corrupt decision-log line (%s)", exc) + continue + if isinstance(parsed, dict): + records.append(parsed) + else: + _logger.warning("Skipping a decision-log line that is not an object") + return records diff --git a/screen_locker/_manual_sync.py b/screen_locker/_manual_sync.py index 57d203c..028b22c 100644 --- a/screen_locker/_manual_sync.py +++ b/screen_locker/_manual_sync.py @@ -44,6 +44,36 @@ _logger = logging.getLogger(__name__) # ``workout_data`` — makes re-ingestion of the same record idempotent. _SYNC_ID_FIELD = "sync_record_id" +# The raw user inputs every real manual workout carries. A payload holding none +# of them is a metadata stub, not a damaged workout: there is nothing to +# recover and no amount of retrying will change that. +_SUBSTANTIVE_FIELDS = frozenset( + { + "sport", + "start_time", + "end_time", + "location_name", + "transport_method", + "cost", + "rpe", + "went_well", + "to_improve", + "overall_feeling", + } +) + + +def is_empty_stub(payload: Mapping[str, object]) -> bool: + """True when ``payload`` carries no workout content at all. + + Observed in the wild as ``{"type", "kind", "date"}`` and nothing else -- + a record the phone created without ever attaching the workout. Every sync + cycle re-reported it as malformed, so a permanent, unfixable condition + produced an unbounded stream of identical warnings (four every 15 minutes), + which is how the genuinely actionable lines got lost in the noise. + """ + return not (_SUBSTANTIVE_FIELDS & set(payload)) + def _coerce_int(value: object) -> int: """Coerce a JSON scalar to int; raise for a non-numeric (skips the record).""" @@ -100,6 +130,39 @@ def reconstruct_draft(payload: Mapping[str, object]) -> ManualWorkoutDraft | Non return None +def _draft_or_report( + record_id: str, payload: Mapping[str, object] +) -> ManualWorkoutDraft | None: + """Rebuild the draft, or report why it cannot be rebuilt and return None. + + The two failure modes are deliberately reported at different levels, because + only one of them is actionable: + + * an **empty stub** carries no workout at all (seen as ``{type, kind, + date}``) -- nothing to recover, nothing to fix, so re-reporting it at + ``warning`` on every 15-minute sync cycle only teaches the reader to + ignore this logger; + * a **malformed** payload has real workout fields that would not parse, + which means genuine credit may be going missing -- that stays loud. + """ + if is_empty_stub(payload): + _logger.info( + "Manual record %s is an empty stub (keys: %s) — no workout data to " + "ingest; skipping permanently, no credit is being lost", + record_id, + sorted(payload), + ) + return None + draft = reconstruct_draft(payload) + if draft is None: + _logger.warning( + "Manual record %s is malformed — it HAS workout fields but they " + "could not be parsed, so real credit may be lost", + record_id, + ) + return draft + + def _already_ingested(logs: dict[str, list[dict]], record_id: str) -> bool: """True if any logged entry already carries this sync record id. @@ -154,9 +217,8 @@ def ingest_manual_records( continue if _already_ingested(load_workout_log(log_file), record_id): continue - draft = reconstruct_draft(payload) + draft = _draft_or_report(record_id, payload) if draft is None: - _logger.warning("Manual record %s is malformed — skipped", record_id) continue error = validate_manual_workout(draft) if error is not None: @@ -169,9 +231,13 @@ def ingest_manual_records( entry[_SYNC_ID_FIELD] = record_id result = write_signed_entry(log_file, date, entry) if not result.appended: - _logger.warning( - "Manual record %s collided with an existing workout_id — " - "skipped without credit", + # This is dedup working, not credit being lost: the workout is + # already in the log under this same workout_id. It was logged at + # warning and re-reported on every 15-minute cycle, which read like + # a recurring fault and buried the lines that did need attention. + _logger.info( + "Manual record %s is already logged under the same workout_id " + "— skipped as a duplicate (its credit is already counted)", record_id, ) continue diff --git a/screen_locker/_startup_checks.py b/screen_locker/_startup_checks.py index 5030585..5bace04 100644 --- a/screen_locker/_startup_checks.py +++ b/screen_locker/_startup_checks.py @@ -21,12 +21,17 @@ from screen_locker._constants import ( SHUTDOWN_BASE_FILE, SICK_DAY_STATE_FILE, ) +from screen_locker._decision_log import ( + LockDecision, + record_decision, +) from screen_locker._extra_benefits import process_week_transition from screen_locker._shutdown_base import reset_to_base_if_new_day from screen_locker._sync_mixin import SyncMixin from screen_locker._temperature import fetch_current_temp_with_status from screen_locker._weekly_check import ( WEEKLY_WORKOUT_MINIMUM, + count_weekly_workouts, has_weekly_minimum, is_relaxed_day, ) @@ -45,11 +50,36 @@ class StartupChecksMixin(SyncMixin): and this keeps the god-class's ancestor count from growing. """ + def _record_decision( + self, *, locked: bool, reason: str, detail: str, **extra: object + ) -> None: + """Record one lock decision, annotated with this week's progress. + + The weekly count is attached to every decision because "was it right to + skip?" is unanswerable without it -- during the 2026-08 outage the + journal could not distinguish "5/5, correctly skipped" from "0/5, should + have locked". + """ + record_decision( + LockDecision( + locked=locked, + reason=reason, + detail=detail, + weekly_count=count_weekly_workouts(self.log_file), + weekly_required=WEEKLY_WORKOUT_MINIMUM, + extra=extra, + ) + ) + + def _record_skip(self, reason: str, detail: str, **extra: object) -> None: + """Record a decision not to enforce, then exit the process.""" + self._record_decision(locked=False, reason=reason, detail=detail, **extra) + sys.exit(0) + def _check_non_verify_exits(self) -> None: """Check all normal (non-verify) startup early-exit conditions.""" if self._is_scheduled_skip_today(): - _logger.info("Today is a scheduled skip day. Skipping screen lock.") - sys.exit(0) + self._record_skip("scheduled_skip_day", "Today is a scheduled skip day.") return # Award streak / shutdown-bonus / EB-extension rewards from last week # before the daily reset, so a Monday transition's bonus is recorded @@ -74,16 +104,19 @@ class StartupChecksMixin(SyncMixin): return # Day-of-week routing: Tue/Wed/Thu relaxed (optional), Fri-Mon enforced. if is_relaxed_day(): - _logger.info("Relaxed day (Tue-Thu) - showing optional workout prompt.") + self._record_decision( + locked=False, + reason="relaxed_day", + detail="Relaxed day (Tue-Thu) — showing the optional prompt.", + ) self._relaxed_day_mode = True return # Fri-Mon: skip lock when weekly minimum is already met. if has_weekly_minimum(self.log_file): - _logger.info( - "Weekly minimum of %d workouts met. Skipping screen lock.", - WEEKLY_WORKOUT_MINIMUM, + self._record_skip( + "weekly_minimum_met", + f"Weekly minimum of {WEEKLY_WORKOUT_MINIMUM} workouts met.", ) - sys.exit(0) return # Only remaining same-day skip: genuine extreme heat. Sick days go # through the justification flow instead; there is no banked @@ -91,6 +124,15 @@ class StartupChecksMixin(SyncMixin): # maximizing weekly workouts, so it was removed in favor of a # shutdown-time-only reward (see _apply_weekly_shutdown_bonus). self._check_heat_skip_exit() + # Nothing excused today: falling through here means the lock WILL be + # built. Recorded explicitly so the trail shows enforcement happening, + # not merely the absence of a skip -- "no line at all" was precisely + # the signature of the 2026-08 outage. + self._record_decision( + locked=True, + reason="enforced", + detail="No exemption applied — building the lock screen.", + ) def _check_heat_skip_exit(self) -> None: """Exit early if today qualifies for the extreme-heat skip dialog. @@ -120,7 +162,8 @@ class StartupChecksMixin(SyncMixin): ) if self._show_heat_skip_dialog(check.temp_celsius): self._save_heat_skip_log(check.temp_celsius) - _logger.info( - "User skipped workout due to heat (%.0f°C).", check.temp_celsius + self._record_skip( + "heat_skip", + f"User skipped the workout due to heat ({check.temp_celsius:.0f}°C).", + temp_celsius=round(check.temp_celsius, 1), ) - sys.exit(0) diff --git a/screen_locker/_status.py b/screen_locker/_status.py index 6b99317..e7ee762 100644 --- a/screen_locker/_status.py +++ b/screen_locker/_status.py @@ -9,6 +9,7 @@ import sys from typing import TYPE_CHECKING from screen_locker._constants import EXTRA_BENEFITS_FILE +from screen_locker._decision_log import record_no_decision from screen_locker._extra_benefits import ( current_streak, has_extended_early_bird, @@ -162,4 +163,7 @@ def run_status(locker: ScreenLocker) -> None: _mw, _ts, _morning = cfg print(f" Shutdown tonight : {_mw:02d}:00") + # --status is a read-out, not an enforcement run; make that explicit so the + # journal never reads as though the locker evaluated and declined to lock. + record_no_decision("--status") sys.exit(0) diff --git a/screen_locker/screen_lock.py b/screen_locker/screen_lock.py index 6ed68b9..bebe176 100755 --- a/screen_locker/screen_lock.py +++ b/screen_locker/screen_lock.py @@ -169,6 +169,10 @@ class ScreenLocker( self._lock.close() else: self.root.destroy() + # The lock screen coming down is the end of enforcement for this run, + # so it belongs in the journal next to the decision that raised it — + # otherwise "the lock closed" and "the lock never opened" look alike. + _logger.warning("Lock window closed — enforcement for this run is over.") sys.exit(0) def run(self) -> None: diff --git a/screen_locker/tests/conftest.py b/screen_locker/tests/conftest.py index 1b6b77c..ef1a293 100644 --- a/screen_locker/tests/conftest.py +++ b/screen_locker/tests/conftest.py @@ -162,6 +162,10 @@ _ISOLATED_STATE: tuple[tuple[str, tuple[str, ...]], ...] = ( ), ), ("scheduled_skips.json", ("_log_mixin.SCHEDULED_SKIPS_FILE",)), + # The durable lock-decision trail. Written on EVERY locker run, so without + # this the suite would append test decisions to the user's real + # enforcement history in ~/.local/share/screen_locker/. + ("decisions.jsonl", ("_decision_log.DECISION_LOG_FILE",)), ) diff --git a/screen_locker/tests/test_decision_log.py b/screen_locker/tests/test_decision_log.py new file mode 100644 index 0000000..4fa7bf0 --- /dev/null +++ b/screen_locker/tests/test_decision_log.py @@ -0,0 +1,213 @@ +"""Tests for screen_locker._decision_log — the durable lock-decision trail. + +This module exists because the 2026-08 enforcement outage left no evidence: +every skip logged at INFO on a path with no logging configured, so the journal +recorded thirteen days of nothing. These tests pin the two properties that make +that impossible to repeat -- a skip is loud, and every decision is written to a +file that outlives the journal. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import patch + +from screen_locker._decision_log import ( + DECISION_LOG_MAX_ENTRIES, + LockDecision, + read_decisions, + record_decision, + record_no_decision, +) + +if TYPE_CHECKING: + import pytest + + +class TestLockDecisionRendering: + """The one-line summary must stay greppable and complete.""" + + def test_skip_line_reports_reason_and_weekly_progress(self) -> None: + """A skip renders lock=no plus the counts that justify it.""" + line = LockDecision( + locked=False, + reason="early_bird_window_active", + weekly_count=0, + weekly_required=5, + ).as_line() + assert "DECISION" in line + assert "lock=no" in line + assert "reason=early_bird_window_active" in line + assert "weekly=0/5" in line + + def test_enforced_line_reports_lock_yes(self) -> None: + """An enforced run renders lock=yes.""" + assert "lock=yes" in LockDecision(locked=True, reason="enforced").as_line() + + def test_extra_fields_and_detail_are_rendered(self) -> None: + """Extra context and the human detail both reach the line.""" + line = LockDecision( + locked=False, + reason="heat_skip", + detail="too hot", + extra={"temp_celsius": 34.0}, + ).as_line() + assert "temp_celsius=34.0" in line + assert "too hot" in line + + def test_weekly_is_omitted_when_unknown(self) -> None: + """No weekly counts → no misleading 'weekly=None/None'.""" + assert "weekly=" not in LockDecision(locked=True, reason="enforced").as_line() + + +class TestRecordDecision: + """Recording must be loud for skips and durable for everything.""" + + def test_skip_is_logged_at_warning( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A decision NOT to enforce must be visible at warning. + + This is the exact regression: an INFO-level skip is invisible under the + default logging configuration and hid the outage for thirteen days. + """ + with caplog.at_level("INFO"): + record_decision( + LockDecision(locked=False, reason="weekly_minimum_met"), + log_file=tmp_path / "d.jsonl", + ) + record = next(r for r in caplog.records if "DECISION" in r.message) + assert record.levelname == "WARNING" + + def test_enforced_is_logged_at_info( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """An enforced lock is self-evident on screen, so INFO suffices.""" + with caplog.at_level("INFO"): + record_decision( + LockDecision(locked=True, reason="enforced"), + log_file=tmp_path / "d.jsonl", + ) + record = next(r for r in caplog.records if "DECISION" in r.message) + assert record.levelname == "INFO" + + def test_decision_is_appended_to_the_durable_trail(self, tmp_path: Path) -> None: + """Each decision lands in the file as one JSON object.""" + target = tmp_path / "d.jsonl" + record_decision( + LockDecision( + locked=False, reason="sick_day", weekly_count=2, weekly_required=5 + ), + log_file=target, + ) + record_decision(LockDecision(locked=True, reason="enforced"), log_file=target) + records = read_decisions(log_file=target) + assert [r["reason"] for r in records] == ["sick_day", "enforced"] + assert records[0]["weekly_count"] == 2 + assert records[0]["locked"] is False + assert "timestamp" in records[0] + + def test_parent_directory_is_created(self, tmp_path: Path) -> None: + """A missing data directory is created rather than losing the record.""" + target = tmp_path / "nested" / "deeper" / "d.jsonl" + record_decision(LockDecision(locked=True, reason="enforced"), log_file=target) + assert target.exists() + + def test_history_is_trimmed_to_the_cap(self, tmp_path: Path) -> None: + """An every-30-minutes timer must not grow the trail without bound.""" + target = tmp_path / "d.jsonl" + target.write_text( + "\n".join( + json.dumps({"reason": f"old{i}"}) + for i in range(DECISION_LOG_MAX_ENTRIES + 50) + ) + + "\n" + ) + record_decision(LockDecision(locked=True, reason="newest"), log_file=target) + records = read_decisions(log_file=target) + assert len(records) == DECISION_LOG_MAX_ENTRIES + assert records[-1]["reason"] == "newest" + + def test_write_failure_is_reported_not_swallowed( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A trail that cannot be written says so — a gap must be visible.""" + target = tmp_path / "d.jsonl" + with ( + patch.object(Path, "write_text", side_effect=OSError("disk full")), + caplog.at_level("WARNING"), + ): + record_decision( + LockDecision(locked=True, reason="enforced"), log_file=target + ) + assert any("decision log" in r.message for r in caplog.records) + + def test_write_failure_does_not_stop_the_locker(self, tmp_path: Path) -> None: + """Failing to record must never prevent the lock from happening.""" + with patch.object(Path, "write_text", side_effect=OSError("nope")): + record_decision( + LockDecision(locked=True, reason="enforced"), + log_file=tmp_path / "d.jsonl", + ) + + +class TestRecordNoDecision: + """Modes that never evaluate the lock must be distinguishable.""" + + def test_records_mode_and_null_lock(self, tmp_path: Path) -> None: + """'did not decide' must not look like 'decided not to lock'.""" + target = tmp_path / "d.jsonl" + record_no_decision("--sync-only", log_file=target) + (record,) = read_decisions(log_file=target) + assert record["locked"] is None + assert record["reason"] == "mode_makes_no_decision" + assert record["mode"] == "--sync-only" + + +class TestReadDecisions: + """Reading must degrade gracefully, never crash an investigation.""" + + def test_missing_file_returns_empty_and_says_so( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """No history yet is normal, but is still stated out loud.""" + with caplog.at_level("WARNING"): + assert read_decisions(log_file=tmp_path / "nope.jsonl") == [] + assert any("No decision log" in r.message for r in caplog.records) + + def test_unreadable_file_is_reported( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """An OSError is surfaced rather than silently returning [].""" + target = tmp_path / "d.jsonl" + target.write_text("{}\n") + with ( + patch.object(Path, "read_text", side_effect=OSError("locked")), + caplog.at_level("WARNING"), + ): + assert read_decisions(log_file=target) == [] + assert any("Could not read" in r.message for r in caplog.records) + + def test_corrupt_lines_are_skipped_not_fatal( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A damaged trail still yields its readable records.""" + target = tmp_path / "d.jsonl" + target.write_text('{"reason": "good"}\nnot json at all\n\n') + with caplog.at_level("WARNING"): + records = read_decisions(log_file=target) + assert [r["reason"] for r in records] == ["good"] + assert any("corrupt" in r.message for r in caplog.records) + + def test_non_object_lines_are_skipped( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """A bare JSON scalar is not a decision record.""" + target = tmp_path / "d.jsonl" + target.write_text('[1, 2, 3]\n{"reason": "good"}\n') + with caplog.at_level("WARNING"): + records = read_decisions(log_file=target) + assert [r["reason"] for r in records] == ["good"] + assert any("not an object" in r.message for r in caplog.records) diff --git a/screen_locker/tests/test_manual_sync_part3.py b/screen_locker/tests/test_manual_sync_part3.py index f643bdc..e52a975 100644 --- a/screen_locker/tests/test_manual_sync_part3.py +++ b/screen_locker/tests/test_manual_sync_part3.py @@ -22,6 +22,8 @@ from screen_locker.tests.test_manual_sync import _manual_record if TYPE_CHECKING: from pathlib import Path + import pytest + _DATE = "2026-07-13" _TT_DRAFT = ManualWorkoutDraft( @@ -120,3 +122,51 @@ class TestIngestManualRecordsCallback: ) assert ingested == [] assert calls == [] + + +class TestEmptyStubRecords: + """A payload with no workout content is skipped quietly, not as a fault. + + Observed in the wild as ``{"type", "kind", "date"}`` -- the phone created + the record but never attached the workout. It replayed as a WARNING on + every 15-minute sync cycle forever, which is how genuinely actionable + lines from this logger got tuned out. + """ + + def test_is_empty_stub_detects_a_metadata_only_payload(self) -> None: + """No substantive field present → stub.""" + assert _manual_sync.is_empty_stub( + {"type": "manual_workout", "kind": "manual_workout", "date": _DATE} + ) + + def test_is_empty_stub_is_false_when_any_workout_field_exists(self) -> None: + """One real field is enough to make it a (possibly broken) workout.""" + assert not _manual_sync.is_empty_stub( + {"kind": "manual_workout", "sport": "run"} + ) + + def test_stub_is_skipped_without_a_warning( + self, tmp_path: Path, caplog: pytest.LogCaptureFixture + ) -> None: + """The stub is reported at info, so it stops polluting the warnings.""" + log_file = tmp_path / "workout_log.json" + log_file.write_text(json.dumps({})) + with caplog.at_level("INFO"): + ingested = ingest_manual_records( + log_file, + [ + ( + "manual_workout:2026-08-15", + { + "kind": "manual_workout", + "type": "manual_workout", + "date": _DATE, + }, + ) + ], + ) + assert ingested == [] + stub_records = [r for r in caplog.records if "empty stub" in r.message] + assert stub_records, "the stub must still be reported" + assert all(r.levelname == "INFO" for r in stub_records) + assert not [r for r in caplog.records if r.levelname == "WARNING"] diff --git a/screen_locker/tests/test_systemd_units.py b/screen_locker/tests/test_systemd_units.py new file mode 100644 index 0000000..d146b40 --- /dev/null +++ b/screen_locker/tests/test_systemd_units.py @@ -0,0 +1,149 @@ +"""Tests for the systemd unit files that ARM enforcement. + +These exist because of a real outage: on 2026-08-04 systemd logged + + timers.target: Found ordering cycle: early-bird-workout-check.timer/start + after graphical-session.target/start after basic.target/start after + timers.target/start - after early-bird-workout-check.timer + timers.target: Job early-bird-workout-check.timer/start deleted to break + ordering cycle starting with timers.target/start + +and left the timer *disabled*. The Python lock-decision chain was perfectly +healthy the whole time — it simply was never invoked again, so the screen +locker stopped locking for days without a single failing unit or error line. + +A timer that is ``WantedBy=timers.target`` must not be ordered ``After=`` a +target that itself comes after ``timers.target``; that closes a dependency +loop and systemd resolves it by silently deleting the job. Parsing is plain +text on purpose: this must run in CI, where no systemd is present. The +complementary *runtime* check ("is it actually enabled and scheduled right +now?") cannot run in a container and lives in +``scripts/check_enforcement_armed.py`` instead. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] + +# Targets that are ordered AFTER timers.target. A timer pulled in by +# timers.target may never declare After= on any of these. +_TARGETS_AFTER_TIMERS = frozenset( + { + "graphical-session.target", + "basic.target", + "graphical.target", + "multi-user.target", + "default.target", + } +) + +# Every timer unit that schedules the locker, i.e. every file whose breakage +# silently disarms enforcement. +_ENFORCEMENT_TIMERS = ("early-bird-workout-check.timer", "workout-locker.timer") + + +def _directive_values(unit_text: str, key: str) -> list[str]: + """Return every value assigned to ``key``, ignoring comments. + + systemd allows a directive to repeat (``OnCalendar=`` does, deliberately), + so this returns a list rather than a single value. + """ + values: list[str] = [] + for raw_line in unit_text.splitlines(): + line = raw_line.strip() + if not line or line.startswith(("#", ";")): + continue + name, sep, value = line.partition("=") + if sep and name.strip() == key: + values.append(value.strip()) + return values + + +@pytest.mark.parametrize("timer_name", _ENFORCEMENT_TIMERS) +class TestEnforcementTimersAreCycleFree: + """Every locker timer must be installable without an ordering cycle.""" + + def test_timer_file_exists(self, timer_name: str) -> None: + """The unit file ships in the repo, so install_systemd.sh can copy it.""" + assert (REPO_ROOT / timer_name).is_file(), ( + f"{timer_name} is missing; enforcement cannot be scheduled without it" + ) + + def test_no_ordering_cycle_with_timers_target(self, timer_name: str) -> None: + """No After= on a target that is itself ordered after timers.target. + + This is the exact regression from 2026-08-04. Without this assertion the + failure is invisible: the unit file looks reasonable, nothing errors, and + the timer just quietly never fires. + """ + text = (REPO_ROOT / timer_name).read_text(encoding="utf-8") + wanted_by = set() + for value in _directive_values(text, "WantedBy"): + wanted_by.update(value.split()) + if "timers.target" not in wanted_by: + pytest.skip(f"{timer_name} is not pulled in by timers.target") + + ordered_after = set() + for value in _directive_values(text, "After"): + ordered_after.update(value.split()) + + offenders = sorted(ordered_after & _TARGETS_AFTER_TIMERS) + assert not offenders, ( + f"{timer_name} declares After={offenders} while being " + f"WantedBy=timers.target. timers.target is ordered before " + f"basic.target before graphical-session.target, so this closes an " + f"ordering cycle and systemd deletes the timer job to break it — " + f"silently disabling enforcement (see 2026-08-04)." + ) + + def test_triggers_the_locker_service(self, timer_name: str) -> None: + """The timer must actually start the locker, not some other unit.""" + text = (REPO_ROOT / timer_name).read_text(encoding="utf-8") + assert _directive_values(text, "Unit") == ["workout-locker.service"], ( + f"{timer_name} must trigger workout-locker.service" + ) + + def test_survives_a_missed_trigger(self, timer_name: str) -> None: + """Persistent=true, so a suspend across the trigger still re-checks. + + A laptop asleep at 08:30 would otherwise skip the re-check entirely and + leave the early-bird marker banked for the rest of the day. + """ + assert _directive_values( + text := (REPO_ROOT / timer_name).read_text(encoding="utf-8"), "Persistent" + ) == ["true"], ( + f"{timer_name} must set Persistent=true so a missed trigger is " + f"still run after resume; text was:\n{text}" + ) + + +class TestEarlyBirdTimerWindows: + """The early-bird timer must keep closing both grace windows.""" + + def test_covers_normal_and_extended_windows(self) -> None: + """08:30 closes the normal window; 09:05 closes the extended one.""" + text = (REPO_ROOT / "early-bird-workout-check.timer").read_text( + encoding="utf-8" + ) + calendars = _directive_values(text, "OnCalendar") + assert "*-*-* 08:30:00" in calendars + assert "*-*-* 09:05:00" in calendars + + +class TestPeriodicLockerTimer: + """A day/week boundary crossed mid-session must be re-evaluated.""" + + def test_repeats_through_the_day(self) -> None: + """OnCalendar repeats, so enforcement is not login-only. + + The 2026-08-16 -> 08-17 rollover was missed precisely because the + service is WantedBy=graphical-session.target (one shot per login) and + the only other trigger was the broken early-bird timer. + """ + text = (REPO_ROOT / "workout-locker.timer").read_text(encoding="utf-8") + calendars = _directive_values(text, "OnCalendar") + assert calendars, "workout-locker.timer must define at least one OnCalendar" diff --git a/scripts/add_scheduled_skip.py b/scripts/add_scheduled_skip.py new file mode 100755 index 0000000..34648e9 --- /dev/null +++ b/scripts/add_scheduled_skip.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""Add a date to ``scheduled_skips.json`` — the locker's first early exit. + +A scheduled skip is the supported way to exempt a specific day without +weakening enforcement for any other day: the lock chain checks it first, and +(since the 2026-08 logging work) records the exemption as a decision, so an +unexpected quiet day is traceable to the date that caused it instead of looking +like the enforcer silently died again. + +Kept out of ``arm.sh`` because the file it edits is the one the lock chain +reads: a malformed write is read as "not a skip" and the machine locks anyway, +so this belongs somewhere the repo's linters and tests actually apply. + +Usage: + python3 scripts/add_scheduled_skip.py --date today + python3 scripts/add_scheduled_skip.py --date 2026-08-20 + python3 scripts/add_scheduled_skip.py --list +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +import json +from pathlib import Path +import sys + +SKIPS_FILE = ( + Path(__file__).resolve().parent.parent / "screen_locker" / ("scheduled_skips.json") +) + + +def _today() -> str: + """Return today's local date as YYYY-MM-DD. + + Local, not UTC: the skip must match the day the user is living in, and the + lock chain's own ``_today_str`` is local for the same reason. + """ + return datetime.now().astimezone().strftime("%Y-%m-%d") + + +def _report(message: str) -> None: + """Write one line to stdout (matches the other scripts in this directory).""" + sys.stdout.write(f"{message}\n") + + +def load_skips(path: Path) -> list[str]: + """Return the recorded skip dates, or [] when the file is absent.""" + if not path.exists(): + return [] + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + # Refuse rather than silently starting a fresh list: overwriting an + # unreadable file would drop skips the user still expects to hold. + msg = f"could not read {path}: {exc}" + msg = f"ERROR: {msg}" + raise SystemExit(msg) from exc + if not isinstance(data, list): + msg = f"ERROR: {path} does not contain a JSON list" + raise SystemExit(msg) + return [str(entry) for entry in data] + + +def add_skip(path: Path, date: str) -> bool: + """Add ``date`` to the skips file. True if it was newly added.""" + skips = load_skips(path) + if date in skips: + return False + skips.append(date) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(sorted(skips), indent=2) + "\n", encoding="utf-8") + return True + + +def main(argv: list[str] | None = None) -> int: + """Add the requested date, or list what is already recorded.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--date", + help="date to skip: 'today' or YYYY-MM-DD", + ) + parser.add_argument( + "--list", + action="store_true", + help="print the recorded skip dates and exit", + ) + args = parser.parse_args(argv) + + if args.list: + for entry in sorted(load_skips(SKIPS_FILE)): + _report(entry) + return 0 + + if not args.date: + parser.error("--date is required (or use --list)") + + date = _today() if args.date == "today" else args.date + try: + datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=timezone.utc) + except ValueError: + msg = f"ERROR: {date!r} is not a YYYY-MM-DD date" + raise SystemExit(msg) from None + + if add_skip(SKIPS_FILE, date): + _report(f"Added scheduled skip for {date} — the locker will not lock that day.") + else: + _report(f"{date} is already a scheduled skip; nothing to do.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/check_enforcement_armed.py b/scripts/check_enforcement_armed.py new file mode 100755 index 0000000..584ce45 --- /dev/null +++ b/scripts/check_enforcement_armed.py @@ -0,0 +1,184 @@ +#!/usr/bin/env python3 +"""Fail loudly when the screen locker is not actually armed. + +The 2026-08 outage was not a logic bug. Every predicate worked; the timer that +*invokes* them had been disabled by a systemd ordering cycle, and nothing on the +machine considered that worth mentioning. ``systemctl status`` showed no failed +unit, because a job deleted to break a cycle is not a failure -- it is an +absence, and absences do not page anyone. + +This script turns that absence into a loud, deterministic failure. It answers +one question -- *would enforcement actually happen?* -- and answers it from +systemd's own view rather than from the unit files on disk, because the files +were correct-looking the entire time the timer sat disabled. + +Run it from the periodic sync timer (``workout-sync.service``), so the check +rides along with a unit already proven healthy instead of adding another timer +that could itself silently die. + +Output goes through ``sys.stderr.write`` rather than ``print`` to match the +other gate scripts in this directory (see ``check_silent_failures.py``), which +keeps the repo free of T201 waivers. + +Usage: + python3 scripts/check_enforcement_armed.py # exit 1 if disarmed + python3 scripts/check_enforcement_armed.py --quiet # only report problems +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +import logging +import shutil +import subprocess +import sys + +_logger = logging.getLogger(__name__) + +# Every unit that must be armed for a workout-less day to end in a lock. +# workout-locker.service itself is WantedBy=graphical-session.target and so is +# only ever a login one-shot; the timers are what make enforcement recur. +REQUIRED_TIMERS = ( + "early-bird-workout-check.timer", + "workout-locker.timer", +) + +_TIMEOUT_SECONDS = 10 + + +@dataclass(frozen=True) +class TimerState: + """What systemd currently believes about one timer.""" + + name: str + enabled: bool + enabled_raw: str + scheduled: bool + + @property + def armed(self) -> bool: + """True only when the timer is both enabled and actually scheduled. + + Both halves matter. ``is-enabled`` alone passes for a timer whose job + systemd deleted to break an ordering cycle; ``list-timers`` alone passes + for a transient ``start``ed timer that will not survive a reboot. + """ + return self.enabled and self.scheduled + + def describe(self) -> str: + """Return a one-line human summary of this timer's state.""" + if self.armed: + return f"OK {self.name}: enabled and scheduled" + problems = [] + if not self.enabled: + problems.append(f"not enabled (is-enabled={self.enabled_raw or 'unknown'})") + if not self.scheduled: + problems.append("no next trigger in list-timers") + return f"DISARMED {self.name}: {', '.join(problems)}" + + +def _report(message: str) -> None: + """Write one line to stderr, so a disarmed locker is visible in the unit.""" + sys.stderr.write(f"{message}\n") + + +def _run(args: list[str]) -> tuple[int, str]: + """Run ``args``, returning (returncode, stdout). Never raises.""" + try: + # Explicit argument list, never shell=True. + proc = subprocess.run( + args, + capture_output=True, + text=True, + timeout=_TIMEOUT_SECONDS, + check=False, + ) + except (OSError, subprocess.SubprocessError): + # A check that cannot run must not look like a check that passed. + _logger.exception("Could not run %s", " ".join(args)) + return 1, "" + return proc.returncode, proc.stdout + + +def _scheduled_timers() -> set[str]: + """Return the timer names systemd currently has a next trigger for.""" + _, out = _run(["systemctl", "--user", "list-timers", "--all", "--no-pager"]) + scheduled: set[str] = set() + for line in out.splitlines(): + for name in REQUIRED_TIMERS: + # A timer with no next trigger prints "n/a" in the NEXT column. + if name in line and " n/a " not in f" {line} ": + scheduled.add(name) + return scheduled + + +def collect_states() -> list[TimerState]: + """Return the current arming state of every required timer.""" + scheduled = _scheduled_timers() + states = [] + for name in REQUIRED_TIMERS: + code, out = _run(["systemctl", "--user", "is-enabled", name]) + raw = out.strip() + states.append( + TimerState( + name=name, + enabled=code == 0 and raw == "enabled", + enabled_raw=raw, + scheduled=name in scheduled, + ) + ) + return states + + +def main(argv: list[str] | None = None) -> int: + """Report arming state; return 1 when enforcement would not happen.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--quiet", + action="store_true", + help="print only when something is wrong", + ) + args = parser.parse_args(argv) + + if shutil.which("systemctl") is None: + # Reachable in containers/CI. Say so explicitly rather than passing: + # "could not check" must never be recorded as "checked and fine". + _logger.error("systemctl not found — the arming check could not run") + _report( + "ERROR: systemctl not found — cannot verify that the screen locker " + "is armed. This check did NOT pass; it did not run." + ) + return 1 + + states = collect_states() + disarmed = [state for state in states if not state.armed] + + if disarmed: + _logger.error( + "Screen locker is NOT armed: %s", + ", ".join(state.name for state in disarmed), + ) + _report( + "ERROR: the screen locker is NOT armed — a day without a workout " + "would pass unenforced." + ) + for state in states: + _report(f" {state.describe()}") + _report( + "\nRe-arm with:\n" + " systemctl --user daemon-reload\n" + " systemctl --user enable --now " + + " ".join(state.name for state in disarmed) + ) + return 1 + + if not args.quiet: + for state in states: + _report(f" {state.describe()}") + _report("Screen locker enforcement is armed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/check_silent_failures.py b/scripts/check_silent_failures.py index a25bd2a..e7b20d4 100755 --- a/scripts/check_silent_failures.py +++ b/scripts/check_silent_failures.py @@ -50,6 +50,96 @@ def _handler_is_loud(handler: ast.ExceptHandler) -> bool: ) +def _is_silent_exit(node: ast.AST) -> bool: + """Return True if ``node`` is a bare ``sys.exit(0)`` / ``exit(0)`` call.""" + if not isinstance(node, ast.Call): + return False + func = node.func + is_exit = (isinstance(func, ast.Attribute) and func.attr == "exit") or ( + isinstance(func, ast.Name) and func.id == "exit" + ) + if not is_exit or len(node.args) != 1: + return False + arg = node.args[0] + return isinstance(arg, ast.Constant) and arg.value == 0 + + +def _find_silent_exits(tree: ast.AST) -> list[ast.Call]: + """Return every ``sys.exit(0)`` not preceded by a loud explanation. + + A process that exits 0 has declared success, so nothing downstream will ever + flag it — which is precisely how the screen locker stopped enforcing for + thirteen days in 2026-08 while every unit reported ``0/SUCCESS``. Each such + exit abandons the work the process was started to do, so it must say why at + a level a human actually sees. + + The check is per-enclosing-block: an exit is considered explained when a + loud log call or a ``record_decision(...)``-style helper appears in the same + body. That keeps it cheap and, more importantly, keeps it free of an escape + hatch -- there is no comment that silences it. + """ + offenders: dict[int, ast.Call] = {} + for block in ast.walk(tree): + for body_attr in ("body", "orelse", "finalbody"): + statements = getattr(block, body_attr, None) + if not isinstance(statements, list): + continue + for index, statement in enumerate(statements): + exits = [ + inner for inner in ast.walk(statement) if _is_silent_exit(inner) + ] + if not exits: + continue + # Scope the search to the exit's OWN statement plus the ones + # directly before it in the same body. Scanning the whole + # enclosing body was useless: a sibling method's warning() call + # elsewhere in the class marked every exit "explained", which is + # how the original _auto_upgrade.py passed this check while + # containing exactly the bug it is meant to catch. + scope = [*statements[:index], statement] + explained = any( + _is_loud_log_call(inner) or _is_explaining_call(inner) + for candidate in scope + for inner in ast.walk(candidate) + ) + if explained: + continue + for call in exits: + offenders.setdefault(call.lineno, call) + return [call for _, call in sorted(offenders.items())] + + +# Helpers that record the reason for stopping. Calling one counts as speaking +# up, the same as logging loudly. +_EXPLAINING_CALLS = frozenset( + { + "record_decision", + "record_no_decision", + "_record_decision", + "_record_skip", + "_skip", + } +) + +# Deliberately NOT relaxed to accept `info`. An INFO line beside sys.exit(0) +# is structurally indistinguishable from the bug this check exists to catch -- +# the 2026-08 outage looked exactly like `_logger.info(...); sys.exit(0)`. Since +# the AST cannot tell "finished my job" from "abandoned enforcement", the rule +# stays strict and the handful of genuine completion exits record themselves +# through _EXPLAINING_CALLS instead. A check that cannot distinguish must fail +# closed, not open. + + +def _is_explaining_call(node: ast.AST) -> bool: + """Return True if ``node`` calls a helper that records why work stopped.""" + if not isinstance(node, ast.Call): + return False + func = node.func + if isinstance(func, ast.Attribute): + return func.attr in _EXPLAINING_CALLS + return isinstance(func, ast.Name) and func.id in _EXPLAINING_CALLS + + def _handler_label(handler: ast.ExceptHandler) -> str: """Return a readable name for what the handler catches.""" if handler.type is None: @@ -70,12 +160,19 @@ def _check_source(path: Path, source: str) -> list[str]: exc.lineno, ) return [f"{path}:{exc.lineno}: could not parse ({exc.msg})"] - return [ + problems = [ f"{path}:{node.lineno}: {_handler_label(node)} swallows the error without " f"logging why — re-raise, or log at warning/error/exception/critical" for node in ast.walk(tree) if isinstance(node, ast.ExceptHandler) and not _handler_is_loud(node) ] + problems.extend( + f"{path}:{node.lineno}: sys.exit(0) abandons the work this process was " + f"started for without saying why — log at warning/error, or record the " + f"reason (see screen_locker._decision_log.record_decision)" + for node in _find_silent_exits(tree) + ) + return problems def main() -> int: diff --git a/workout-locker.timer b/workout-locker.timer new file mode 100644 index 0000000..ab54c93 --- /dev/null +++ b/workout-locker.timer @@ -0,0 +1,28 @@ +[Unit] +Description=Periodically re-evaluate the workout lock decision +# Deliberately no After=graphical-session.target: this timer is +# WantedBy=timers.target, and ordering it after a target that itself comes +# after timers.target closes a cycle systemd breaks by deleting the job. +# See early-bird-workout-check.timer for the outage that taught us this. + +[Timer] +# workout-locker.service is WantedBy=graphical-session.target, i.e. it runs +# ONCE per graphical login. That meant a day boundary or an ISO-week rollover +# crossed inside a long session was never re-evaluated: on 2026-08-16 the +# locker correctly exited (weekly minimum met), the new week began on 08-17, +# and nothing re-checked it because the session never ended. +# +# This timer closes that hole. Every run is idempotent -- the service's own +# early-exit ladder handles "already logged", "window still active", +# "weekly minimum met" and "relaxed day" -- so a re-check costs nothing when +# there is nothing to do. +OnCalendar=*-*-* *:00,30:00 +# Catch up after suspend/downtime rather than waiting for the next half hour. +Persistent=true +# A lock that appears at a hard-predictable second feels like a glitch; a small +# spread also avoids stacking with workout-sync.timer on the same tick. +RandomizedDelaySec=60 +Unit=workout-locker.service + +[Install] +WantedBy=timers.target diff --git a/workout-sync.service b/workout-sync.service index 4c1c225..bf110f4 100644 --- a/workout-sync.service +++ b/workout-sync.service @@ -13,3 +13,10 @@ Type=oneshot # did once per process start -- so a workout finished after login was invisible # until the next login. ExecStart=/usr/bin/python3 -m screen_locker.screen_lock --sync-only +# Piggyback the arming check on this already-healthy 15-minute timer rather +# than adding another timer that could itself be silently disabled -- which is +# the exact failure this check exists to catch (2026-08-04: an ordering cycle +# disabled early-bird-workout-check.timer and nothing noticed for 13 days). +# ExecStartPost without "-" means a disarmed locker marks this unit failed, so +# it surfaces in `systemctl --user --failed` instead of scrolling past. +ExecStartPost=/usr/bin/python3 %h/screen-locker/scripts/check_enforcement_armed.py --quiet