screen-locker/screen_locker/_armed_state.py

"""Whether systemd would actually run the locker, read from systemd itself.

The 2026-08 outage was not a logic bug. Every predicate worked; the timer that
*invokes* them had been disabled by an 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 module answers one question -- *would enforcement actually happen?* -- 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.

It lives in the package rather than in ``scripts/`` because two surfaces need
the same answer: ``scripts/check_enforcement_armed.py`` (the gate that rides
along with ``workout-sync.service``) and the Health view of the local web UI.
Two copies of this logic would be free to disagree about whether the locker is
alive, which is the failure it exists to detect.
"""

from __future__ import annotations

from dataclasses import dataclass
import logging
import shutil
import subprocess

_logger = logging.getLogger(__name__)

# Every unit that must be armed for a workout-less day to end in a lock.
# The timers are what make enforcement recur.
REQUIRED_TIMERS = (
    "early-bird-workout-check.timer",
    "workout-locker.timer",
)

# workout-locker.service is WantedBy=graphical-session.target: a login
# one-shot that closes the window between a boot and the next timer tick.
# It is checked here because on 2026-08-30 it had silently drifted to
# `disabled`, the machine rebooted at 14:06, and nothing enforced until
# 14:30 -- while this very module reported "armed" the whole time, because
# it only ever looked at the two timers.
REQUIRED_SERVICES = ("workout-locker.service",)

# The target the login run hangs off. Checked as well as `is-enabled`
# because ~/.config/i3/config notes that i3 "does not reliably activate
# graphical-session.target" -- an enabled service anchored to an inactive
# target is exactly the correct-looking-but-disarmed state this module was
# written to catch.
_LOGIN_TARGET = "graphical-session.target"

_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.

        Returns:
            True when this timer would really fire.
        """
        return self.enabled and self.scheduled

    def describe(self) -> str:
        """Return a one-line human summary of this unit's state.

        Returns:
            An ``OK``/``DISARMED`` line naming every reason it is not armed.
        """
        if self.armed:
            return f"OK       {self.name}: {self._ready_phrase}"
        problems = []
        if not self.enabled:
            problems.append(f"not enabled (is-enabled={self.enabled_raw or 'unknown'})")
        if not self.scheduled:
            problems.append(self._missing_phrase)
        return f"DISARMED {self.name}: {', '.join(problems)}"

    @property
    def _ready_phrase(self) -> str:
        """Return the wording for an armed unit of this kind."""
        if self.name.endswith(".timer"):
            return "enabled and scheduled"
        return f"enabled and anchored to an active {_LOGIN_TARGET}"

    @property
    def _missing_phrase(self) -> str:
        """Return the wording for the not-scheduled half of this kind."""
        if self.name.endswith(".timer"):
            return "no next trigger in list-timers"
        return f"{_LOGIN_TARGET} is not active, so the login run cannot fire"


def systemctl_available() -> bool:
    """Return whether ``systemctl`` can be run at all.

    Reachable in containers and CI. Callers must treat False as "did not
    check", never as "checked and fine".

    Returns:
        True when ``systemctl`` is on PATH.
    """
    return shutil.which("systemctl") is not None


def _run(args: list[str]) -> tuple[int, str]:
    """Run ``args``, returning ``(returncode, stdout)``. Never raises.

    Args:
        args: The argument vector, passed without a shell.

    Returns:
        The process return code and its stdout; ``(1, "")`` if it could not run.
    """
    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.

    Returns:
        The subset of :data:`REQUIRED_TIMERS` with a scheduled next run.
    """
    _, 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


LOCKER_UNIT = "workout-locker.service"


def locker_unit_active() -> bool | None:
    """Return whether the locker unit is running right now.

    "The lock would fire" is a statement about the *decision*; it says
    nothing about whether anything is currently enforcing it. On 2026-08-30
    the machine had rebooted, no locker process existed, and the status page
    read exactly the same as it does with a lock on screen.

    Returns:
        True/False, or None when systemd could not be asked -- which must
        never be rendered as "not running".
    """
    if not systemctl_available():
        return None
    code, out = _run(["systemctl", "--user", "is-active", LOCKER_UNIT])
    if code and not out.strip():
        return None
    return out.strip() == "active"


def _login_target_active() -> bool:
    """Return whether the target the login run hangs off is active.

    Returns:
        True when ``graphical-session.target`` is active.
    """
    code, out = _run(["systemctl", "--user", "is-active", _LOGIN_TARGET])
    return not code and out.strip() == "active"


def collect_states() -> list[TimerState]:
    """Return the current arming state of every unit enforcement depends on.

    Returns:
        One :class:`TimerState` per entry in :data:`REQUIRED_TIMERS` and
        :data:`REQUIRED_SERVICES`.
    """
    scheduled = _scheduled_timers()
    login_target_up = _login_target_active()
    states = []
    for name in (*REQUIRED_TIMERS, *REQUIRED_SERVICES):
        code, out = _run(["systemctl", "--user", "is-enabled", name])
        raw = out.strip()
        states.append(
            TimerState(
                name=name,
                enabled=not code and raw == "enabled",
                enabled_raw=raw,
                # For a timer this is "systemd has a next trigger for it";
                # for the login-run service it is "the target it is wanted by
                # is actually up". Both answer the same question: would this
                # unit really fire?
                scheduled=(
                    name in scheduled if name.endswith(".timer") else login_target_up
                ),
            )
        )
    return states