kuhyx/screen-locker
1
0
Fork 0
mirror of https://github.com/kuhyx/screen-locker.git synced 2026-11-10 21:27:13 +01:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Re-arm the locker, and make a disarmed enforcer impossible to miss
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Details
Tests / test (3.10) (push) Waiting to run
Details
Tests / test (3.11) (push) Waiting to run
Details
Tests / test (3.12) (push) Waiting to run
Details
Tests / lock-popup-safety (push) Waiting to run
Details
release-apk / release (APK) (push) Waiting to run
Details

Browse Source 
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01F2b6ZjBcdrkSYnMujx4mWd
This commit is contained in:
Krzysztof Rudnicki 2026-08-17 16:47:25 +02:00
parent b1ee78be64
commit e5db25468d
20 changed files with 1411 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@ -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]

116
arm.sh Executable file
View File

@ -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 "$@"

14
early-bird-workout-check.timer
View File

@ -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

19
install_systemd.sh
View File

@ -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"

54
screen_locker/_auto_upgrade.py
View File

@ -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:

19
screen_locker/_cli.py
View File

@ -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(

227
screen_locker/_decision_log.py Normal file
View File

@ -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

76
screen_locker/_manual_sync.py
View File

@ -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

63
screen_locker/_startup_checks.py
View File

@ -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)

4
screen_locker/_status.py
View File

@ -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)

4
screen_locker/screen_lock.py
View File

@ -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:

4
screen_locker/tests/conftest.py
View File

@ -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",)),
)

213
screen_locker/tests/test_decision_log.py Normal file
View File

@ -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)

50
screen_locker/tests/test_manual_sync_part3.py
View File

@ -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"]

149
screen_locker/tests/test_systemd_units.py Normal file
View File

@ -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"

113
scripts/add_scheduled_skip.py Executable file
View File

@ -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())

184
scripts/check_enforcement_armed.py Executable file
View File

@ -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())

99
scripts/check_silent_failures.py
View File

@ -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:

28
workout-locker.timer Normal file
View File

@ -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

7
workout-sync.service
View File

@ -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