screen-locker/screen_locker/_compliance_state.py
Krzysztof kuhy Rudnicki d9119a6582
Some checks are pending
pre-commit / pre-commit (push) Waiting to run
Tests / test (3.10) (push) Waiting to run
Tests / test (3.11) (push) Waiting to run
Tests / test (3.12) (push) Waiting to run
Add manual (unverified) workout logging with evidence form
Adds a rate-limited manual-workout entry point (StatusWindow + locked
retry screen) for activities like table tennis that ADB/RunnerUp can't
verify, gated by a 2/7d + 5/30d budget and a detailed evidence form
rather than free text. Full credit toward the weekly minimum, debt
clearing, and commitment prompt, matching phone/RunnerUp verification.

Also stops tracking personal runtime state (workout_log.json,
sick_history.json, extra_benefits_state.json, scheduled_skips.json,
shutdown_base.json, early_bird_pending.json) in git — these are
regenerated locally and never belong in a public/private repo's
history.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AbPcvAvt1JhiRzEyPDE5dA
2026-07-05 20:08:52 +02:00

384 lines
13 KiB
Python

"""Pure, read-only predicates mirroring the lock-decision chain.
Single source of truth for the leaf checks walked by
``screen_lock.ScreenLocker._check_non_verify_exits`` and
``_auto_upgrade.AutoUpgradeMixin._check_today_state_exits``. The existing
mixins delegate their read-only checks to the functions here; this module
never touches ADB, sudo, subprocess, or the network, and never writes to
disk.
``explain_lock_decision`` re-derives the same ordered chain for the status
view. One branch is a deliberate simplification: when an early-bird pending
marker has expired, the real code attempts a live phone/RunnerUp
auto-upgrade whose outcome can't be known without ADB. Rather than guess,
this module always continues the trace as if that attempt had *not yet
resolved* (mirroring the real code's "auto-upgrade failed" fallthrough) and
surfaces the pending opportunity separately via ``AutoUpgradeOpportunity``.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import json
import logging
from typing import TYPE_CHECKING
from gatelock.log_integrity import compute_entry_hmac, verify_entry_hmac
from screen_locker._constants import (
EARLY_BIRD_END_HOUR,
EARLY_BIRD_END_MINUTE,
EARLY_BIRD_START_HOUR,
)
from screen_locker._sick_tracker import is_sick_day as _is_sick_day
if TYPE_CHECKING:
from pathlib import Path
from screen_locker._sick_tracker import SickHistory
_logger = logging.getLogger(__name__)
def _today_str() -> str:
"""Return today's date as ``YYYY-MM-DD`` in UTC."""
return datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")
@dataclass(frozen=True)
class PredicateResult:
"""One evaluated step in a :class:`LockExplanation` trace."""
name: str
fired: bool
reason: str
@dataclass(frozen=True)
class AutoUpgradeOpportunity:
"""Describes whether the real locker would attempt a live auto-upgrade."""
would_attempt: bool
via: str # "early_bird_expired" | "sick_day" | "none"
reason: str
@dataclass(frozen=True)
class LockExplanation:
"""Read-only reconstruction of what the lock-decision chain would do."""
fired: bool
stage: str
reason: str
trace: tuple[PredicateResult, ...]
auto_upgrade: AutoUpgradeOpportunity
heat_skip_evaluated: bool
_NO_UPGRADE = AutoUpgradeOpportunity(
would_attempt=False,
via="none",
reason="No pending auto-upgrade opportunity.",
)
def is_scheduled_skip_today(
scheduled_skips_file: Path, *, today: str | None = None
) -> bool:
"""Return True if *today* is listed in *scheduled_skips_file*."""
if not scheduled_skips_file.exists():
return False
try:
with scheduled_skips_file.open() as f:
skips = json.load(f)
except (OSError, json.JSONDecodeError):
return False
return (today or _today_str()) in skips
def has_logged_today(log_file: Path, *, today: str | None = None) -> bool:
"""Return True if a validly-signed workout is logged for *today*."""
if not log_file.exists():
return False
try:
with log_file.open() as f:
logs = json.load(f)
except (OSError, json.JSONDecodeError):
return False
entry = logs.get(today or _today_str())
if entry is None:
return False
if verify_entry_hmac(entry):
return True
if compute_entry_hmac({"_probe": True}) is None and "hmac" not in entry:
_logger.info("HMAC key unavailable — accepting unsigned entry")
return True
_logger.warning("HMAC verification failed for today's log entry")
return False
def is_early_bird_pending(pending_file: Path, *, today: str | None = None) -> bool:
"""Return True if today has an unresolved early-bird pending marker."""
if not pending_file.exists():
return False
try:
with pending_file.open() as f:
state = json.load(f)
except (OSError, json.JSONDecodeError):
return False
if not isinstance(state, dict) or state.get("date") != (today or _today_str()):
return False
if verify_entry_hmac(state):
return True
if compute_entry_hmac({"_probe": True}) is None and "hmac" not in state:
_logger.info("HMAC key unavailable — accepting unsigned pending marker")
return True
_logger.warning("HMAC verification failed for early-bird pending marker")
return False
def is_sick_day_today(history: SickHistory, *, today: str | None = None) -> bool:
"""Return True if today is recorded as a sick day.
Thin wrapper around ``_sick_tracker.is_sick_day`` kept here so callers of
the status layer have one predicate module to import from.
"""
return _is_sick_day(history, today=today)
def _early_bird_window_open(*, extended: bool, local_minutes: int) -> bool:
"""Deliberate, independent reimplementation of ``_is_early_bird_time``.
Not shared with ``EarlyBirdMixin._is_early_bird_time`` because too many
existing tests are pinned to its ``_get_local_time_minutes`` patch point.
Kept here as a small, separate, pure re-implementation for the status
layer only — see the module docstring in ``_early_bird.py``.
"""
start = EARLY_BIRD_START_HOUR * 60
end = 9 * 60 if extended else EARLY_BIRD_END_HOUR * 60 + EARLY_BIRD_END_MINUTE
return start <= local_minutes < end
def describe_auto_upgrade_opportunity(
*,
early_bird_pending: bool,
early_bird_window_open: bool,
is_sick_day: bool,
) -> AutoUpgradeOpportunity:
"""Describe what the real ``_try_auto_upgrade_*`` methods would attempt.
Read-only stand-in for ``AutoUpgradeMixin._check_today_state_exits``'s
branching — never calls phone/RunnerUp verification and never writes to
``workout_log.json``.
"""
if early_bird_pending and not early_bird_window_open:
return AutoUpgradeOpportunity(
would_attempt=True,
via="early_bird_expired",
reason=(
"Early-bird window closed with no workout logged — the next "
"lock run will try phone/RunnerUp verification before "
"falling back to a full lock."
),
)
if is_sick_day:
return AutoUpgradeOpportunity(
would_attempt=True,
via="sick_day",
reason=(
"Sick day logged — the next lock run will try phone/RunnerUp "
"verification to upgrade it to a real workout."
),
)
return _NO_UPGRADE
def _stage(
*,
fired: bool,
stage: str,
reason: str,
trace: list[PredicateResult],
auto_upgrade: AutoUpgradeOpportunity,
) -> LockExplanation:
"""Build a terminal :class:`LockExplanation` from the trace so far."""
return LockExplanation(
fired=fired,
stage=stage,
reason=reason,
trace=tuple(trace),
auto_upgrade=auto_upgrade,
heat_skip_evaluated=False,
)
def explain_lock_decision(
*,
log_file: Path,
scheduled_skips_file: Path,
early_bird_pending_file: Path,
sick_history: SickHistory,
extended_early_bird: bool,
weekly_minimum_met: bool,
relaxed_day: bool,
wake_skip: bool = False,
now: datetime | None = None,
) -> LockExplanation:
"""Re-derive the lock-decision chain without ADB, sudo, or writes.
Mirrors ``_check_non_verify_exits`` / ``_check_today_state_exits`` in
order. Heat-skip is never evaluated (it needs a live ``wttr.in`` call) —
reported via ``heat_skip_evaluated=False``, not guessed.
"""
instant = now if now is not None else datetime.now(tz=timezone.utc)
today_str = instant.astimezone(timezone.utc).strftime("%Y-%m-%d")
local_dt = instant.astimezone()
local_minutes = local_dt.hour * 60 + local_dt.minute
scheduled = is_scheduled_skip_today(scheduled_skips_file, today=today_str)
pending = is_early_bird_pending(early_bird_pending_file, today=today_str)
window_open = _early_bird_window_open(
extended=extended_early_bird, local_minutes=local_minutes
)
expired = pending and not window_open
early_bird_open_and_pending = pending and window_open
sick_today = is_sick_day_today(sick_history, today=today_str)
logged = has_logged_today(log_file, today=today_str)
auto_upgrade = describe_auto_upgrade_opportunity(
early_bird_pending=pending,
early_bird_window_open=window_open,
is_sick_day=sick_today,
)
trace: list[PredicateResult] = []
def _check(
name: str,
*,
fired: bool,
reason_true: str,
reason_false: str,
terminal_reason: str | None = None,
) -> LockExplanation | None:
"""Record one trace step; return an explanation if it terminates the chain."""
reason = reason_true if fired else reason_false
trace.append(PredicateResult(name, fired, reason))
if fired and terminal_reason is not None:
return _stage(
fired=False,
stage=name,
reason=terminal_reason,
trace=trace,
auto_upgrade=auto_upgrade,
)
return None
result = _check(
"scheduled_skip",
fired=scheduled,
reason_true="Today is in scheduled_skips.json (unconditional manual skip).",
reason_false="Today is not a manually scheduled skip day.",
terminal_reason="Manually scheduled skip day — no lock, no workout required.",
)
if result is not None:
return result
# Recorded but never terminal on its own: an expired pending marker still
# falls through toward a full lock, with the upgrade opportunity surfaced
# separately via `auto_upgrade` (see module docstring).
_check(
"early_bird_pending_expired",
fired=expired,
reason_true="Logged in during the early-bird window and it has since closed "
"with no workout confirmed yet.",
reason_false="No expired early-bird pending marker.",
)
result = _check(
"early_bird_window_open",
fired=early_bird_open_and_pending,
reason_true="Logged in during the early-bird window earlier today; still "
"waiting to see a workout before the window closes.",
reason_false="Not pending inside an open early-bird window.",
terminal_reason=(
"Early-bird window still open — lock skipped while waiting for a workout."
),
)
if result is not None:
return result
result = _check(
"sick_day",
fired=sick_today,
reason_true="Today is marked as a sick day in sick_history.json.",
reason_false="Today is not marked as a sick day.",
terminal_reason="Sick day logged today — lock skipped.",
)
if result is not None:
return result
result = _check(
"already_logged",
fired=logged,
reason_true="A validly-signed workout is already logged for today.",
reason_false="No workout logged for today yet.",
terminal_reason="Workout already logged today — lock skipped.",
)
if result is not None:
return result
result = _check(
"wake_alarm_skip",
fired=wake_skip,
reason_true="Wake-alarm earned a workout skip for today.",
reason_false="No wake-alarm workout skip earned today.",
terminal_reason="Wake-alarm earned a workout skip — lock skipped.",
)
if result is not None:
return result
result = _check(
"early_bird_time_fresh",
fired=window_open,
reason_true="Currently inside the early-bird window (05:00-08:30/09:00).",
reason_false="Not currently inside the early-bird window.",
terminal_reason="Inside the early-bird window — lock skipped, pending marker "
"would be saved for the 08:30/09:00 re-check.",
)
if result is not None:
return result
result = _check(
"relaxed_day",
fired=relaxed_day,
reason_true="Today is a relaxed day (Tue/Wed/Thu).",
reason_false="Today is not a relaxed day.",
terminal_reason="Relaxed day (Tue/Wed/Thu) — workout optional, lock skipped.",
)
if result is not None:
return result
result = _check(
"weekly_minimum_met",
fired=weekly_minimum_met,
reason_true="Weekly workout minimum already reached.",
reason_false="Weekly workout minimum not yet reached.",
terminal_reason="Weekly workout minimum already met — lock skipped.",
)
if result is not None:
return result
return _stage(
fired=True,
stage="full_lock_pending_heat_check",
reason=(
"No skip condition applies. The real locker still checks live "
"Warsaw temperature before showing the full lock — not "
"evaluated here."
),
trace=trace,
auto_upgrade=auto_upgrade,
)