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

Make the demo escapable, and stop claiming the blocking was deliberate

Browse Source 
The demo promised three escapes and delivered none: the only way out was a
hard reboot. Two independent causes, and only one of them was demo-only.

launch_workout_app blocks until the app exits. Called straight from a Tk
button that froze the event loop, so the Escape binding and the close button
could not fire -- measured: zero handlers ran during the block, and the events
were never delivered, not merely delivered late. WorkoutSession drives the
same phases through `after` instead, on the main thread, so grab_release and
grab_input stay where Tk requires them and the loop keeps pumping (199 ticks
during a run that previously managed none).

The second cause survives a healthy event loop. The Flutter app takes its own
gdk_seat_grab with GDK_SEAT_CAPABILITY_ALL and never releases it, so while it
is up X delivers the keyboard to it and no binding of ours can fire. Correct
for production, fatal for a demo, and no flag turns it off -- so the demo's
escape is a file the app cannot intercept: touch ~/.cache/stop-workout-demo.

The previous commit's rationale for the blocking was wrong and is corrected
here rather than left to mislead. It claimed pumping Tk would let the recovery
watchdog steal the screen mid-workout; in fact release() stops the watchdog
first and stop() cancels its timers outright, so there were never any ticks to
race with. The blocking was incidental, which is why removing it costs the
production handoff nothing.

Verified with the real debug bundle under Xvfb: with the app up and holding
its seat grab, the event loop is alive, the escape file terminates it, and the
lock takes the grab back.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DfJC1Wu1SEgJ6X49k5Jqtg
This commit is contained in:
Krzysztof Rudnicki 2026-08-22 12:33:07 +02:00
parent f65cb150aa
commit c3424059b2
4 changed files with 446 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
screen_locker/_workout_handoff.py
View File

@ -65,13 +65,15 @@ def start_workout(lock: LockWindow) -> str:
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.
Blocks until the app exits, so it is only safe for a caller with no event
loop to starve. Behind a Tk button use :class:`WorkoutSession` instead:
blocking there freezes the loop, which makes every binding on the lock --
including the escapes -- inert until the app exits.
An earlier version of this docstring claimed the blocking was deliberate,
to stop the recovery watchdog ticking mid-workout. That was wrong:
``release`` stops the watchdog first, and ``stop()`` cancels its timers
outright, so there are no ticks to race with either way.
"""
result = launch_workout_app(lock_grab_handoff(lock))
if not result.launched:

165
screen_locker/_workout_session.py Normal file
View File

@ -0,0 +1,165 @@
"""Running the workout app without freezing the lock's event loop.
``launch_workout_app`` blocks until the app exits, which is fine for a
headless caller but wrong behind a Tk button: while it blocks, the event loop
is not running, so key bindings and buttons on the lock itself are inert. In
the demo lock -- whose whole promise is that Escape always works -- that
turned a "safe" screen into one that needed a hard reboot to leave.
So the phases are driven by ``after`` instead, on the main thread. Every Tk
call (``grab_release``, ``grab_input``) stays there, which a worker thread
could not guarantee, and the loop keeps pumping the entire time the app is up.
This does not weaken the production handoff. The grab is still released only
after the ready line, and the recovery watchdog is still stopped first -- and
because ``stop()`` cancels its timers outright, there are no ticks to race
with while the app runs.
"""
from __future__ import annotations
import logging
import subprocess
from typing import TYPE_CHECKING
from screen_locker._workout_app import ProcessHooks, workout_app_binary
from screen_locker._workout_ready import READY_MARKER
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
from typing import IO, Protocol
from screen_locker._workout_app import GrabHandoff
class _Child(Protocol):
"""The half of ``subprocess.Popen`` this module drives."""
stdout: IO[str] | None
returncode: int | None
def poll(self) -> int | None: ...
def terminate(self) -> None: ...
__all__ = ["WorkoutSession"]
_logger = logging.getLogger(__name__)
# How often to look at the app while it runs. Short enough that the lock comes
# back promptly when the app exits, long enough not to spin the event loop.
_POLL_MS = 100
# 600 polls x 100ms = 60s, matching the blocking supervisor's own deadline.
_READY_TIMEOUT_POLLS = 600
class WorkoutSession:
"""One supervised run of the workout app, driven by the Tk event loop.
``on_status`` is called with a human-readable line at each transition, so a
failure is visible on the lock screen rather than only in the journal.
"""
def __init__(
self,
handoff: GrabHandoff,
*,
after: Callable[[int, Callable[[], None]], object],
on_status: Callable[[str], None],
binary: Path | None = None,
hooks: ProcessHooks | None = None,
) -> None:
self._handoff = handoff
self._after = after
self._on_status = on_status
self._binary = binary
self._hooks = hooks if hooks is not None else ProcessHooks()
self._remaining = _READY_TIMEOUT_POLLS
self._child: _Child | None = None
self._seen = ""
self._released = False
def start(self) -> None:
"""Spawn the app; the rest happens on later event-loop turns."""
app = self._binary if self._binary is not None else workout_app_binary()
if app is None:
self._fail("the desktop workout app is not built; run flutter build linux")
return
self._child = self._hooks.popen(
[str(app), "--lock-mode"],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
self._on_status("Starting the workout app…")
self._after(_POLL_MS, self._await_ready)
def _await_ready(self) -> None:
"""Watch for the ready line without ever blocking the loop."""
child = self._child
stdout = getattr(child, "stdout", None)
if stdout is None:
self._fail("the workout app exited before signalling ready")
return
# Zero timeout: ask what is there right now and return to the loop.
readable, _, _ = self._hooks.selector([stdout], [], [], 0)
if readable:
chunk = _read_available(stdout)
if not chunk:
self._fail("the workout app exited before signalling ready")
return
self._seen += chunk
if READY_MARKER in self._seen:
self._release_and_watch()
return
self._remaining -= 1
if self._remaining <= 0:
# Never release to a window that may not be up.
child.terminate()
self._fail("the workout app never signalled ready in time")
return
self._after(_POLL_MS, self._await_ready)
def _release_and_watch(self) -> None:
self._handoff.release()
self._released = True
self._on_status("The workout app has the screen.")
self._after(_POLL_MS, self._watch_until_exit)
def _watch_until_exit(self) -> None:
if self._child.poll() is None:
self._after(_POLL_MS, self._watch_until_exit)
return
self.finish()
def finish(self) -> None:
"""Take the screen back. Idempotent, so an early abort is safe."""
if self._released:
self._released = False
self._handoff.reacquire()
code = getattr(self._child, "returncode", None)
_logger.warning("The workout app exited with code %s.", code)
self._on_status("Workout app closed — checking whether it counted…")
def abort(self) -> None:
"""Kill the app and restore the lock, whatever phase we are in."""
child = self._child
if child is not None and child.poll() is None:
child.terminate()
_logger.warning("Workout app terminated by an explicit abort.")
self.finish()
def _fail(self, reason: str) -> None:
_logger.error("Could not run the PC workout: %s", reason)
self._on_status(f"Could not start the workout: {reason}")
def _read_available(stream: IO[str]) -> str:
"""Read what the pipe holds now; see ``_workout_ready`` for the why."""
buffer = getattr(stream, "buffer", None)
if buffer is None:
return stream.readline()
return buffer.read1().decode("utf-8", "replace")

209
screen_locker/tests/test_workout_session.py Normal file
View File

@ -0,0 +1,209 @@
"""Tests for running the workout app without freezing the lock's event loop.
The invariant these defend: no phase blocks. Every wait is an ``after``
callback, because a blocked event loop makes the lock's own escapes inert --
which is what once forced a hard reboot out of a "safe" demo.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
from screen_locker._workout_app import GrabHandoff, ProcessHooks
from screen_locker._workout_ready import READY_MARKER
from screen_locker._workout_session import WorkoutSession
if TYPE_CHECKING:
from collections.abc import Callable
class _FakeAfter:
"""A drainable stand-in for ``root.after``."""
def __init__(self) -> None:
self.queue: list[Callable[[], None]] = []
def __call__(self, _ms: int, callback: Callable[[], None]) -> object:
self.queue.append(callback)
return object()
def drain(self, limit: int = 5000) -> None:
for _ in range(limit):
if not self.queue:
return
self.queue.pop(0)()
class _FakeStdout:
def __init__(self, chunks: list[str]) -> None:
self._chunks = chunks
def readline(self) -> str:
return self._chunks.pop(0) if self._chunks else ""
class _FakeChild:
def __init__(self, chunks: list[str], *, exits_after: int = 1) -> None:
self.stdout = _FakeStdout(chunks)
self.returncode: int | None = None
self.terminated = False
self._polls = 0
self._exits_after = exits_after
def poll(self) -> int | None:
self._polls += 1
if self._polls >= self._exits_after:
self.returncode = 0
return 0
return None
def terminate(self) -> None:
self.terminated = True
self.returncode = -15
def _session(child: _FakeChild, after: _FakeAfter, calls: list[str]):
return WorkoutSession(
GrabHandoff(
release=lambda: calls.append("release"),
reacquire=lambda: calls.append("reacquire"),
),
after=after,
on_status=lambda text: calls.append(f"status:{text[:20]}"),
binary=Path("/nonexistent/workout_app"),
hooks=ProcessHooks(
popen=lambda *_a, **_k: child,
selector=lambda r, _w, _x, _t: (r, [], []),
),
)
class TestWorkoutSession:
"""Each phase hands control back to the event loop."""
def test_releases_the_grab_only_after_the_ready_line(self) -> None:
after, calls = _FakeAfter(), []
child = _FakeChild([f"{READY_MARKER}\n"], exits_after=2)
_session(child, after, calls).start()
assert "release" not in calls # nothing released synchronously
after.drain()
assert "release" in calls
assert calls.index("release") < calls.index("reacquire")
def test_never_releases_when_the_app_dies_before_ready(self) -> None:
after, calls = _FakeAfter(), []
_session(_FakeChild([]), after, calls).start()
after.drain()
assert "release" not in calls
def test_reacquires_when_the_app_exits(self) -> None:
after, calls = _FakeAfter(), []
child = _FakeChild([f"{READY_MARKER}\n"], exits_after=3)
_session(child, after, calls).start()
after.drain()
assert calls[-2] == "reacquire"
assert "Workout app closed" in calls[-1]
def test_abort_terminates_the_app_and_restores_the_lock(self) -> None:
"""The escape hatch: usable at any point, from outside X."""
after, calls = _FakeAfter(), []
child = _FakeChild([f"{READY_MARKER}\n"], exits_after=99)
session = _session(child, after, calls)
session.start()
for _ in range(3):
if after.queue:
after.queue.pop(0)()
session.abort()
assert child.terminated
assert "reacquire" in calls
def test_abort_before_release_does_not_reacquire(self) -> None:
"""Nothing was lent, so there is nothing to take back."""
after, calls = _FakeAfter(), []
session = _session(_FakeChild([], exits_after=99), after, calls)
session.start()
session.abort()
assert "reacquire" not in calls
def test_times_out_without_ever_blocking(self) -> None:
after, calls = _FakeAfter(), []
child = _FakeChild(["noise\n"] * 10, exits_after=99)
session = _session(child, after, calls)
session._remaining = 3
session.start()
after.drain()
assert child.terminated
assert "release" not in calls
def test_says_so_when_the_app_is_not_built(self) -> None:
after, calls = _FakeAfter(), []
session = WorkoutSession(
GrabHandoff(release=lambda: None, reacquire=lambda: None),
after=after,
on_status=calls.append,
hooks=ProcessHooks(popen=lambda *_a, **_k: None),
)
# Force the unbuilt case regardless of what this machine has built.
session._binary = None
monkey = "screen_locker._workout_session.workout_app_binary"
import unittest.mock
with unittest.mock.patch(monkey, return_value=None):
session.start()
assert any("not built" in c for c in calls)
def test_handles_a_missing_stdout_pipe(self) -> None:
after, calls = _FakeAfter(), []
child = _FakeChild([])
child.stdout = None
_session(child, after, calls).start()
after.drain()
assert "release" not in calls
class TestSessionEdges:
"""Paths a fake pipe or a happy path does not reach."""
def test_keeps_waiting_when_the_pipe_has_nothing_yet(self) -> None:
"""A quiet app is not a dead app; poll again, never block."""
after, calls = _FakeAfter(), []
child = _FakeChild([f"{READY_MARKER}\n"], exits_after=99)
session = WorkoutSession(
GrabHandoff(
release=lambda: calls.append("release"),
reacquire=lambda: calls.append("reacquire"),
),
after=after,
on_status=lambda _t: None,
binary=Path("/nonexistent/workout_app"),
hooks=ProcessHooks(
popen=lambda *_a, **_k: child,
# Never readable: the app is up but silent.
selector=lambda _r, _w, _x, _t: ([], [], []),
),
)
session._remaining = 2
session.start()
after.drain()
assert "release" not in calls
assert child.terminated
def test_abort_with_no_child_is_harmless(self) -> None:
after, calls = _FakeAfter(), []
session = _session(_FakeChild([]), after, calls)
session.abort() # never started
assert len(calls) == 1
assert "Workout app closed" in calls[0]
def test_reads_a_real_pipe_without_waiting_for_a_newline(self) -> None:
"""The production path: a text stream wrapping a binary buffer."""
import os
from screen_locker._workout_session import _read_available
read_fd, write_fd = os.pipe()
with os.fdopen(read_fd, "r") as reader, os.fdopen(write_fd, "w") as writer:
writer.write("WORKOUT_")
writer.flush()
assert _read_available(reader) == "WORKOUT_"

79
scripts/demo_workout_handoff.py
View File

@ -1,15 +1,35 @@
"""Try the workout-app handoff by hand, on a lock you can always escape.
Deliberately NOT the production lock screen. This builds a demo-mode lock --
``grab="local"``, VT switching left enabled -- so a wedged Flutter app can
never trap you: Ctrl+Alt+F2, the window manager, and the Escape binding below
all still work. Production uses a global grab and disables VT switching, which
is exactly what makes it unsafe to poke at by hand.
Deliberately NOT the production lock screen: ``grab="local"`` and VT switching
left enabled, so nothing here can outlive a Ctrl+Alt+F2.
One consequence worth knowing before you read anything into a green run: under
a local grab gatelock's watchdog early-returns and never re-takes the grab, so
this exercises the launch, the ready handshake and the reacquire, but NOT the
1000ms steal-back that the global grab makes possible.
The escapes are the point, and the first version of this script got them
wrong in two independent ways -- it promised three and delivered none, and the
only way out was a hard reboot. Both are fixed here, and both are worth
knowing about because only one of them was a demo-only problem:
1. ``launch_workout_app`` blocks until the app exits. Called straight from a
button, it froze the Tk event loop, so the Escape binding and the close
button could not fire. The run is driven by :class:`WorkoutSession` now,
which uses ``after`` and keeps the loop pumping.
2. The Flutter app takes its OWN seat grab (``gdk_seat_grab`` with
``GDK_SEAT_CAPABILITY_ALL``) and never releases it. That is correct for
production and fatal for a demo: while the app is up, X delivers keyboard
input to it, so no Tk binding of ours can fire no matter how healthy the
event loop is. So the escape hatch below is deliberately NOT a key binding
-- it is a file the app cannot intercept.
While the workout app is on screen, escape it from any other terminal or VT:
touch ~/.cache/stop-workout-demo
A watchdog polls that path and terminates the app, which hands the screen
back. Ctrl+Alt+F2 also still works, because neither this script nor
``enter_lock_mode`` disables VT switching.
One thing a green run here does NOT prove: under a local grab gatelock's
watchdog early-returns and never re-takes the grab, so the 1000ms steal-back
that makes ``_recovery.stop()`` load-bearing in production is not exercised.
python3 scripts/demo_workout_handoff.py
"""
@ -26,7 +46,12 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from gatelock import LockConfig, LockWindow
from screen_locker._workout_app import workout_app_binary
from screen_locker._workout_handoff import start_workout
from screen_locker._workout_handoff import lock_grab_handoff
from screen_locker._workout_session import WorkoutSession
# Polled while the app holds the seat grab; the one escape X cannot swallow.
# Under $HOME rather than /tmp: predictable for the user, and not world-writable.
ESCAPE_FILE = Path.home() / ".cache" / "stop-workout-demo"
class _DemoHooks:
@ -35,6 +60,7 @@ class _DemoHooks:
def __init__(self) -> None:
self.status: tk.Label | None = None
self.lock: LockWindow | None = None
self.session: WorkoutSession | None = None
def build_surface(self, parent: tk.Misc, _surface: object) -> None:
tk.Label(
@ -46,7 +72,10 @@ class _DemoHooks:
).pack(pady=(120, 8))
self.status = tk.Label(
parent,
text="Press “Start workout” to hand this screen to the app.",
text=(
"Press “Start workout”. To escape while the app holds the "
f"screen, run: touch {ESCAPE_FILE}"
),
font=("TkDefaultFont", 12),
bg="#1B1D21",
fg="#9AA0A6",
@ -73,11 +102,29 @@ class _DemoHooks:
def _start(self) -> None:
if self.lock is None or self.status is None: # pragma: no cover
return
self.status.config(text="Launching… the screen is handed over shortly.")
self.status.update_idletasks()
# Blocks for the whole workout, by design -- see start_workout's
# docstring. The demo window stops repainting until the app exits.
self.status.config(text=start_workout(self.lock))
ESCAPE_FILE.unlink(missing_ok=True)
self.session = WorkoutSession(
lock_grab_handoff(self.lock),
after=self.lock.root.after,
on_status=self._set_status,
)
self.session.start()
self._poll_escape_file()
def _set_status(self, text: str) -> None:
if self.status is not None: # pragma: no branch
self.status.config(text=text)
def _poll_escape_file(self) -> None:
"""The escape the Flutter app's seat grab cannot intercept."""
if self.session is None or self.lock is None: # pragma: no cover
return
if ESCAPE_FILE.exists():
ESCAPE_FILE.unlink(missing_ok=True)
self.session.abort()
self.session = None
return
self.lock.root.after(200, self._poll_escape_file)
def _close(self) -> None:
if self.lock is not None: # pragma: no cover