diff --git a/pyproject.toml b/pyproject.toml index 93ec88c..3b845bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,16 @@ unfixable = [] # -truth reuse status_view.py above is exempted for — and _make_bare_verifier # imports ScreenLocker inside the function to avoid an import cycle. "screen_locker/_status_view_verify.py" = ["PLC0415", "SLF001"] +# _workout_handoff.py must pause gatelock's recovery watchdog while the Flutter +# app holds the screen. The loop re-takes any grab it finds missing every +# 1000ms -- by design, so that pulling a cable is not an escape -- and +# holds_grab() is `grab_current() is not None`, which is False once another X +# client holds it. Left running it would steal the screen back from the workout +# app within a second. gatelock exposes no public pause (only close(), which +# also restores VT switching and would hand the user Ctrl+Alt+F2), and adding +# one there would contradict _recovery.py's monotonicity proof that no +# transition ever decreases the grab. +"screen_locker/_workout_handoff.py" = ["SLF001"] # explain_lock_decision takes one arg per real file path + sick history, and # returns early per lock-decision-chain stage (see module docstring) — both # are the deliberate design, not something to collapse into a config object. diff --git a/screen_locker/_workout_handoff.py b/screen_locker/_workout_handoff.py new file mode 100644 index 0000000..bbe97ce --- /dev/null +++ b/screen_locker/_workout_handoff.py @@ -0,0 +1,81 @@ +"""Handing the lock screen to the workout app, and taking it back. + +The Tk locker cannot simply ``close()`` to let the app take over: ``close()`` +also restores VT switching, which would hand the user Ctrl+Alt+F2 and an exit +from the very obligation the lock exists to enforce. Only the *grab* moves. + +The recovery watchdog has to be stopped for the duration, and that is the part +that is easy to get wrong. It is built never to release -- it re-takes a grab +it finds missing, deliberately, so that pulling a cable is not an escape. Left +running it would treat the Flutter app exactly like an intruder and steal the +screen back within one verify tick, mid-workout. So the order is: stop the +watchdog, release the grab, run the app, re-take the grab, restart the +watchdog. The Tk window itself stays mapped underneath the whole time, which +is what keeps the screen covered if the app dies. +""" + +from __future__ import annotations + +import contextlib +import logging +import tkinter as tk +from typing import TYPE_CHECKING + +from screen_locker._workout_app import GrabHandoff, launch_workout_app + +if TYPE_CHECKING: + from gatelock import LockWindow + +__all__ = ["lock_grab_handoff", "start_workout"] + +_logger = logging.getLogger(__name__) + + +def lock_grab_handoff(lock: LockWindow) -> GrabHandoff: + """Build the release/reacquire pair for a live lock window. + + Neither direction touches VT switching or the arbiter claim: this is a + loan of the grab, not a teardown. + """ + + def release() -> None: + # Stop first: a watchdog tick between the release and the app's own + # grab would take the screen straight back off it. + lock._recovery.stop() + with contextlib.suppress(tk.TclError): + lock.root.grab_release() + _logger.warning( + "Released the X grab to the workout app; the lock window stays " + "mapped and VT switching stays disabled." + ) + + def reacquire() -> None: + # Unconditional and idempotent: whatever happened to the app, this + # machine must not be left with nothing holding the screen. + lock.grab_input() + _logger.warning("Re-took the X grab from the workout app.") + + return GrabHandoff(release=release, reacquire=reacquire) + + +def start_workout(lock: LockWindow) -> str: + """Run the workout app as the lock surface; return a status for the UI. + + Never raises: this is wired to a button on a screen the user cannot leave, + so a failure has to come back as a message rather than a traceback that + would leave the grab in an unknown state. + + Blocks until the app exits, which stops the Tk event loop for the whole + workout. That is deliberate rather than merely tolerated: the Flutter + window is covering the screen, and the alternative -- pumping Tk + underneath -- is what would let the watchdog tick and take the screen back + mid-set. The cost is that a monitor hotplug during the workout is not + handled until the app exits, when ``reacquire`` re-runs ``grab_input()`` + and the watchdog re-covers every live output. + """ + result = launch_workout_app(lock_grab_handoff(lock)) + if not result.launched: + _logger.error("Could not start the PC workout: %s", result.reason) + return f"Could not start the workout: {result.reason}" + _logger.warning("The workout app finished: %s", result.reason) + return "Workout app closed — checking whether it counted…" diff --git a/screen_locker/tests/test_workout_handoff.py b/screen_locker/tests/test_workout_handoff.py new file mode 100644 index 0000000..9255289 --- /dev/null +++ b/screen_locker/tests/test_workout_handoff.py @@ -0,0 +1,114 @@ +"""Tests for lending the lock screen's grab to the workout app. + +The invariant: the grab moves, nothing else does. VT switching stays disabled +and the arbiter claim stays held, because ``close()`` would undo both and hand +the user Ctrl+Alt+F2 out of the obligation the lock exists to enforce. +""" + +from __future__ import annotations + +import tkinter as tk + +from screen_locker._workout_handoff import lock_grab_handoff, start_workout + + +class _FakeRecovery: + """Stands in for gatelock's watchdog, which re-grabs every 1000ms.""" + + def __init__(self) -> None: + self.running = True + self.events: list[str] = [] + + def stop(self) -> None: + self.running = False + self.events.append("stop") + + +class _FakeLock: + """A lock window that records the order of grab operations.""" + + def __init__(self, *, release_raises: bool = False) -> None: + self._recovery = _FakeRecovery() + self.events: list[str] = [] + self.closed = False + self._release_raises = release_raises + outer = self + + class _Root: + def grab_release(self) -> None: + outer.events.append("grab_release") + if outer._release_raises: + msg = "no grab to release" + raise tk.TclError(msg) + + self.root = _Root() + + def grab_input(self) -> None: + self.events.append("grab_input") + self._recovery.running = True + + def close(self) -> None: + self.closed = True + + +class TestLockGrabHandoff: + """Releasing and re-taking, in the only safe order.""" + + def test_stops_the_watchdog_before_releasing(self) -> None: + """A tick between release and the app's own grab would steal it back.""" + lock = _FakeLock() + lock_grab_handoff(lock).release() + assert lock._recovery.events == ["stop"] + assert lock.events == ["grab_release"] + assert not lock._recovery.running + + def test_reacquire_restores_the_grab_and_the_watchdog(self) -> None: + handoff = lock_grab_handoff(lock := _FakeLock()) + handoff.release() + handoff.reacquire() + assert lock.events == ["grab_release", "grab_input"] + assert lock._recovery.running + + def test_never_closes_the_lock_window(self) -> None: + """close() also restores VT switching -- that is the escape hatch.""" + handoff = lock_grab_handoff(lock := _FakeLock()) + handoff.release() + handoff.reacquire() + assert not lock.closed + + def test_tolerates_having_no_grab_to_release(self) -> None: + """Tk raises when releasing a grab we do not hold; not fatal.""" + handoff = lock_grab_handoff(lock := _FakeLock(release_raises=True)) + handoff.release() + handoff.reacquire() + assert lock.events == ["grab_release", "grab_input"] + + +class TestStartWorkout: + """The button handler never raises onto a screen with no way out.""" + + def test_reports_a_failed_launch_as_a_message(self, monkeypatch: object) -> None: + from screen_locker import _workout_handoff + + class _Result: + launched = False + reason = "not built" + + monkeypatch.setattr( + _workout_handoff, "launch_workout_app", lambda _h: _Result() + ) + message = start_workout(_FakeLock()) + assert "Could not start the workout" in message + assert "not built" in message + + def test_reports_a_finished_run(self, monkeypatch: object) -> None: + from screen_locker import _workout_handoff + + class _Result: + launched = True + reason = "exited with code 0" + + monkeypatch.setattr( + _workout_handoff, "launch_workout_app", lambda _h: _Result() + ) + assert "checking whether it counted" in start_workout(_FakeLock())