diff --git a/gatelock/README.md b/gatelock/README.md index 9ef74b1..04a78b1 100644 --- a/gatelock/README.md +++ b/gatelock/README.md @@ -80,6 +80,17 @@ evening night-lockdown already happened?" from night-lockdown's re-arm for the rest of such a day; diet-guard does not. Missing or garbage means "no lockdown" (garbage warns), so the gate locks as before. +`gatelock.held_grabs(claims_dir=None) -> tuple[Claim, ...]` answers "is a +locker's X grab ACTUALLY held right now?": live claims (flock-proved) whose +`grab_held` is true and whose declared grab is not `none`. `grab_held` turns +true only after `grab_set`/`grab_set_global` succeeded, follows each recovery +tick, and is cleared/restored around a deliberate release via +`LockWindow.report_grab_held(held=...)` (or `Arbiter.set_grab_held`) -- +leetcode-guard's study mode. It is rewritten in place under the owner's flock. +A claim from a pre-0.11 writer has no key and reads as `grab != "none"`. Root +has no `XDG_RUNTIME_DIR`, so a root reader passes the user's dir +(`/run/user/1000/gatelock/claims`). + ## Development ```bash diff --git a/gatelock/gatelock/__init__.py b/gatelock/gatelock/__init__.py index 4790af7..1e9cc3d 100644 --- a/gatelock/gatelock/__init__.py +++ b/gatelock/gatelock/__init__.py @@ -14,6 +14,7 @@ from gatelock._arbiter import ( RANK_WAKE_ALARM, Arbiter, ) +from gatelock._claimfile import held_grabs from gatelock._claims import ArbiterVerdict, Claim, default_runtime_dir, grab_strength from gatelock._config import GrabPolicy, LockConfig, SpaceStep, TypeRole from gatelock._detect import OutputChangeDetector @@ -124,6 +125,7 @@ __all__ = [ "escape_text_tab_trap", "grab_strength", "heading", + "held_grabs", "make_button", "measure_fit", "mirror_text_widgets", diff --git a/gatelock/gatelock/_arbiter.py b/gatelock/gatelock/_arbiter.py index ba293b1..8f7c7ce 100644 --- a/gatelock/gatelock/_arbiter.py +++ b/gatelock/gatelock/_arbiter.py @@ -31,12 +31,14 @@ arms anyway. See :meth:`Arbiter.evaluate`. from __future__ import annotations +from dataclasses import replace from datetime import UTC, datetime import logging import os from typing import IO, TYPE_CHECKING, Final import uuid +from gatelock._claimfile import CLAIMS_DIRNAME, rewrite_in_place from gatelock._claims import ( ArbiterVerdict, Claim, @@ -55,12 +57,8 @@ RANK_WAKE_ALARM: Final = 300 RANK_SCREEN_LOCKER: Final = 200 RANK_DIET_GUARD: Final = 100 -_CLAIMS_DIRNAME: Final = "claims" _HOLDER_FILENAME: Final = "holder.lock" -# Ordering for the strength check. A stand-down is only safe towards an -# incumbent whose grab is at least this strong. - def _utc_now_iso() -> str: """Return an ISO-8601 UTC timestamp, sortable as a plain string.""" @@ -89,7 +87,7 @@ class Arbiter: runtime_dir: Override for the state directory. Tests pass a tmp_path. """ self._root = runtime_dir if runtime_dir is not None else default_runtime_dir() - self._claims_dir = self._root / _CLAIMS_DIRNAME + self._claims_dir = self._root / CLAIMS_DIRNAME self._claim = Claim( app=app, rank=rank, @@ -134,6 +132,19 @@ class Arbiter: handle.flush() self._claim_handle = handle + def set_grab_held(self, *, held: bool) -> None: + """Record whether this app's X grab is ACTUALLY held right now. + + Rewrites the claim (and the holder copy, if held) in place under the + existing flock -- see :mod:`gatelock._claimfile`. Idempotent, and + never raises: it runs inside grab and study-mode transitions. + """ + if self._claim.grab_held is held: + return + self._claim = replace(self._claim, grab_held=held) + for handle in (self._claim_handle, self._holder_handle): + rewrite_in_place(handle, self._claim.to_json()) + def live_claims(self) -> tuple[Claim, ...]: """Return every claim whose owning process is still alive. diff --git a/gatelock/gatelock/_arming.py b/gatelock/gatelock/_arming.py index 233c8f1..d90c67a 100644 --- a/gatelock/gatelock/_arming.py +++ b/gatelock/gatelock/_arming.py @@ -51,6 +51,10 @@ _TOPMOST_ON = True _DEFAULT_GRAB_RETRY_MS = 200 +def _no_grab_listener() -> None: + """Default for :attr:`ArmingCollaborators.notify_grab_held`.""" + + @dataclass(frozen=True) class ArmingCollaborators: """Everything the arming sequence needs from its owning window. @@ -64,6 +68,8 @@ class ArmingCollaborators: notify_focus_ready: Called once the lock is mapped and grabbed. log_grab_blocked: Called with the attempt count when a global grab is still blocked, to name the holder and stand a weaker one down. + notify_grab_held: Called once a grab call has actually succeeded, so + the claim's ``grab_held`` turns true only then. """ config: LockConfig @@ -73,6 +79,7 @@ class ArmingCollaborators: recovery: RecoveryLoop notify_focus_ready: Callable[[], None] log_grab_blocked: Callable[[int], None] + notify_grab_held: Callable[[], None] = _no_grab_listener def setup(root: tk.Tk, collab: ArmingCollaborators) -> bool: @@ -142,6 +149,7 @@ def grab_input(root: tk.Tk, collab: ArmingCollaborators) -> None: elif grab == "local": with contextlib.suppress(tk.TclError): root.grab_set() + collab.notify_grab_held() collab.detector.start() collab.recovery.start() root.after(100, collab.notify_focus_ready) @@ -165,6 +173,7 @@ def acquire_global_grab( _logger.warning("Global grab failed, falling back to local grab") with contextlib.suppress(tk.TclError): root.grab_set() + collab.notify_grab_held() return effective_retry_ms = retry_ms or _DEFAULT_GRAB_RETRY_MS if not attempt % collab.config.grab.log_every: @@ -174,6 +183,7 @@ def acquire_global_grab( lambda: acquire_global_grab(root, collab, attempt=attempt + 1), ) return + collab.notify_grab_held() with contextlib.suppress(tk.TclError): root.focus_force() collab.notify_focus_ready() diff --git a/gatelock/gatelock/_claimfile.py b/gatelock/gatelock/_claimfile.py new file mode 100644 index 0000000..12460f5 --- /dev/null +++ b/gatelock/gatelock/_claimfile.py @@ -0,0 +1,89 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Whether a locker's X grab is ACTUALLY held, on disk, for any reader. + +A claim is published *before* its window grabs, and leetcode-guard's study +mode releases the grab while the claim lives on -- so "a live claim exists" is +not "input is locked". :attr:`gatelock.Claim.grab_held` is the second answer, +and this module is how it is written and read. + +**Rewritten in place, never replaced.** Liveness is the ``flock`` the owner +holds on its open claim file (see :mod:`gatelock._arbiter`). A temp file plus +``os.replace`` would swap in a fresh inode nobody has locked, and the next +reader would take its lock and reap it as dead. So the flip overwrites the +same inode through the owner's own descriptor: ``O_APPEND`` (the handle was +opened ``a+``) is cleared so ``pwrite`` lands at offset 0, and the new text is +space-padded to the current size so the file never shrinks -- a reader sees +the old document or the new one, never an empty or truncated one. JSON allows +trailing whitespace, so the padding parses. + +**Never raises.** The flip runs inside grab and study-mode transitions; an +exception there would roll a lock back. A failed write is logged and the +in-memory claim still changes. +""" + +from __future__ import annotations + +import fcntl +import logging +import os +from typing import IO, TYPE_CHECKING, Final + +from gatelock._claims import default_runtime_dir, live_claims + +if TYPE_CHECKING: + from pathlib import Path + + from gatelock._claims import Claim + +_logger = logging.getLogger(__name__) + +CLAIMS_DIRNAME: Final = "claims" + + +def rewrite_in_place(handle: IO[str] | None, text: str) -> bool: + """Overwrite the flock-held file behind `handle` with `text`, same inode. + + Args: + handle: The owner's open (and locked) claim or holder handle; None + when the arbiter never published or has released. + text: The new JSON document. + + Returns: + Whether the file was rewritten. + """ + if handle is None: + return False + try: + handle.flush() + fd = handle.fileno() + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + fcntl.fcntl(fd, fcntl.F_SETFL, flags & ~os.O_APPEND) + data = text.encode("utf-8") + os.pwrite(fd, data.ljust(os.fstat(fd).st_size, b" "), 0) + except (OSError, ValueError) as exc: + _logger.warning("could not rewrite claim file in place: %s", exc) + return False + return True + + +def held_grabs(claims_dir: Path | None = None) -> tuple[Claim, ...]: + """Every live claim whose X grab is actually held right now. + + Readable by root: the enforcer runs as root, where ``XDG_RUNTIME_DIR`` is + unset and the default below resolves to root's own (empty) dir -- so root + callers must pass the user's dir, ``/run/user/1000/gatelock/claims``. + Dead owners' claims are reaped as a side effect, as in + :func:`gatelock._claims.live_claims`. + + Args: + claims_dir: Directory holding claim files; defaults to the one + :class:`gatelock.Arbiter` publishes into for this user. + + Returns: + Live claims with ``grab_held`` and a declared grab other than "none", + in arbitration order. + """ + directory = ( + claims_dir if claims_dir is not None else default_runtime_dir() / CLAIMS_DIRNAME + ) + return tuple(c for c in live_claims(directory) if c.grab_held and c.grab != "none") diff --git a/gatelock/gatelock/_claims.py b/gatelock/gatelock/_claims.py index a9c5e08..1b62dc3 100644 --- a/gatelock/gatelock/_claims.py +++ b/gatelock/gatelock/_claims.py @@ -57,6 +57,9 @@ class Claim: arbiters, and the whole test suite does exactly that to exercise real ``flock`` contention without spawning subprocesses. """ + grab_held: bool = False + """Whether the X grab is ACTUALLY held now (false before it lands and while + released, e.g. leetcode-guard study mode). See :mod:`gatelock._claimfile`.""" def to_json(self) -> str: """Serialise this claim for its on-disk file.""" @@ -69,6 +72,7 @@ class Claim: "grab": self.grab, "disable_vt": self.disable_vt, "instance_id": self.instance_id, + "grab_held": self.grab_held, } ) @@ -86,14 +90,17 @@ class Claim: if not isinstance(raw, dict): return None try: + grab = str(raw["grab"]) return cls( app=str(raw["app"]), rank=int(raw["rank"]), pid=int(raw["pid"]), started=str(raw["started"]), - grab=str(raw["grab"]), + grab=grab, disable_vt=bool(raw["disable_vt"]), instance_id=str(raw["instance_id"]), + # A pre-0.11 writer never wrote the key: trust its declared grab. + grab_held=bool(raw.get("grab_held", grab != "none")), ) except KeyError, TypeError, ValueError: return None diff --git a/gatelock/gatelock/_recovery.py b/gatelock/gatelock/_recovery.py index f96fb9f..f1cec80 100644 --- a/gatelock/gatelock/_recovery.py +++ b/gatelock/gatelock/_recovery.py @@ -193,13 +193,15 @@ class RecoveryLoop: """ if self._parts.config.resolved_grab() != "global": return False - if self.holds_grab(): - return False - with contextlib.suppress(tk.TclError): - self._root.grab_set_global() - _logger.warning("global grab had been lost; re-acquired it") - return True - return False + held = self.holds_grab() + reasserted = False + if not held: + with contextlib.suppress(tk.TclError): + self._root.grab_set_global() + _logger.warning("global grab had been lost; re-acquired it") + reasserted = held = True + self._parts.on_grab_held(held=held) + return reasserted def holds_grab(self) -> bool: """Whether the grab is held by a window this application owns. diff --git a/gatelock/gatelock/_recovery_types.py b/gatelock/gatelock/_recovery_types.py index 151f35d..53da1de 100644 --- a/gatelock/gatelock/_recovery_types.py +++ b/gatelock/gatelock/_recovery_types.py @@ -9,7 +9,7 @@ against a report without pulling in the machinery that builds one. from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol if TYPE_CHECKING: from gatelock._config import LockConfig @@ -19,6 +19,18 @@ if TYPE_CHECKING: from gatelock._surfaces import SurfaceDelta, SurfaceSet +class GrabHeldListener(Protocol): + """Told whether the grab is really ours (``Claim.grab_held``).""" + + def __call__(self, *, held: bool) -> None: + """Record the grab state.""" + + +def ignore_grab_held(*, held: bool) -> None: + """Default :class:`GrabHeldListener`: no arbiter to tell.""" + del held + + @dataclass(frozen=True) class RecoveryReport: """What one tick observed and corrected.""" @@ -46,3 +58,5 @@ class RecoveryCollaborators: enumerator: OutputEnumerator detector: OutputChangeDetector hooks: LockWindowHooks + on_grab_held: GrabHeldListener = ignore_grab_held + """Told each tick whether the grab is really ours (``Claim.grab_held``).""" diff --git a/gatelock/gatelock/_window.py b/gatelock/gatelock/_window.py index 7f81d31..22368ab 100644 --- a/gatelock/gatelock/_window.py +++ b/gatelock/gatelock/_window.py @@ -15,12 +15,8 @@ detection in :mod:`gatelock._detect`, re-assertion in :mod:`gatelock._recovery`, and cross-app priority in :mod:`gatelock._arbiter`. -Arming order matters and is deliberate: **arm first, render second.** The grab -and the VT-disable happen regardless of how many outputs are live, and the -surfaces are then built for whatever the user can actually see -- possibly -nothing. Zero live outputs means "lock without showing", never "decline to -lock". The inverse of that rule is what left the machine unlocked on -2026-07-25. +Arming order (**arm first, render second**) is explained in +:mod:`gatelock._arming`. """ from __future__ import annotations @@ -95,6 +91,7 @@ class LockWindow: enumerator=enumerator, detector=detector, hooks=hooks, + on_grab_held=self.report_grab_held, ), ) # The arming bundle is also where the window reaches its @@ -107,6 +104,7 @@ class LockWindow: recovery=recovery, notify_focus_ready=self._notify_focus_ready, log_grab_blocked=self._log_grab_blocked, + notify_grab_held=lambda: self.report_grab_held(held=True), ) @property @@ -130,6 +128,15 @@ class LockWindow: """The output-change watcher; see :attr:`recovery` for when to stop it.""" return self._arming.detector + def report_grab_held(self, *, held: bool) -> None: + """Publish whether the X grab is ACTUALLY held (``Claim.grab_held``). + + Called on every successful grab and recovery tick; an app releasing the + grab on purpose (leetcode study mode) calls it False/True around that. + """ + if self._arbiter is not None: + self._arbiter.set_grab_held(held=held) + # -- window mechanics ----------------------------------------------------- def setup(self) -> None: diff --git a/gatelock/gatelock/tests/test_claim_grab_held.py b/gatelock/gatelock/tests/test_claim_grab_held.py new file mode 100644 index 0000000..67dad35 --- /dev/null +++ b/gatelock/gatelock/tests/test_claim_grab_held.py @@ -0,0 +1,214 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for ``Claim.grab_held``: whether a locker's X grab is ACTUALLY held. + +A claim is published before the window grabs, and leetcode-guard's study mode +releases the grab while its claim lives on -- so "a claim exists" is not "the +input is locked". ``grab_held`` is the second answer, rewritten in place in +the flock-held claim file, and :func:`gatelock.held_grabs` is how a reader +(root included) asks for it. +""" + +from __future__ import annotations + +import fcntl +import json +import os +import subprocess +import sys +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest + +from gatelock import Arbiter, Claim, held_grabs +from gatelock._arbiter import RANK_SCREEN_LOCKER + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + +_BASE = { + "app": "leetcode_guard", + "rank": 150, + "pid": 1, + "started": "2026-10-19T08:00:00+00:00", + "disable_vt": True, + "instance_id": "abc", +} + + +def _arbiter(runtime: Path, grab: str = "global") -> Arbiter: + """A published arbiter under `runtime`.""" + arbiter = Arbiter( + "leetcode_guard", + RANK_SCREEN_LOCKER, + grab=grab, + disable_vt=True, + runtime_dir=runtime, + ) + arbiter.publish() + return arbiter + + +@pytest.fixture +def arb(tmp_path: Path) -> Iterator[Arbiter]: + """A published global-grab arbiter, released afterwards.""" + arbiter = _arbiter(tmp_path / "rt") + yield arbiter + arbiter.release() + + +def _claim_file(arbiter: Arbiter) -> Path: + (path,) = arbiter._claims_dir.glob("*.json") + return path + + +class TestSerialisation: + """``grab_held`` is always written, and inferred for old writers.""" + + def test_defaults_false_and_is_always_serialised(self) -> None: + """A fresh claim has not grabbed anything yet.""" + claim = Claim(**_BASE, grab="global") + assert claim.grab_held is False + assert json.loads(claim.to_json())["grab_held"] is False + + @pytest.mark.parametrize( + "case", [("global", True), ("local", True), ("none", False)] + ) + def test_old_writer_without_the_key_infers_from_grab( + self, case: tuple[str, bool] + ) -> None: + """JSON from a pre-0.11 writer reads as held whenever it declared a grab.""" + grab, held = case + claim = Claim.from_json(json.dumps({**_BASE, "grab": grab})) + assert claim is not None + assert claim.grab_held is held + + def test_explicit_key_wins_over_inference(self) -> None: + """A new writer saying False is believed even with a global grab.""" + text = json.dumps({**_BASE, "grab": "global", "grab_held": False}) + claim = Claim.from_json(text) + assert claim is not None + assert claim.grab_held is False + + def test_round_trip(self) -> None: + """to_json/from_json preserve the flag.""" + claim = Claim(**_BASE, grab="global", grab_held=True) + assert Claim.from_json(claim.to_json()) == claim + + +class TestSetGrabHeld: + """``Arbiter.set_grab_held`` rewrites the live claim in place.""" + + def test_published_claim_starts_not_held(self, arb: Arbiter) -> None: + """Published before the grab, so it must not claim one yet.""" + (claim,) = arb.live_claims() + assert claim.grab_held is False + + def test_flip_is_visible_to_readers_and_back(self, arb: Arbiter) -> None: + """True after the grab, False again on release, read off disk.""" + arb.set_grab_held(held=True) + assert arb.claim.grab_held is True + assert [c.grab_held for c in arb.live_claims()] == [True] + arb.set_grab_held(held=False) + assert [c.grab_held for c in arb.live_claims()] == [False] + + def test_rewrite_keeps_inode_and_flock(self, arb: Arbiter) -> None: + """Same file, still locked: liveness stays the kernel's flock.""" + path = _claim_file(arb) + inode = path.stat().st_ino + for held in (True, False, True): + arb.set_grab_held(held=held) + assert path.stat().st_ino == inode + with path.open("r", encoding="utf-8") as handle: + with pytest.raises(BlockingIOError): + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + assert Claim.from_json(handle.read()) == arb.claim + + def test_another_process_sees_it_held(self, arb: Arbiter) -> None: + """A separate process (the root enforcer, in production) reads it.""" + arb.set_grab_held(held=True) + code = ( + "import sys; from pathlib import Path; from gatelock import held_grabs; " + "print(len(held_grabs(Path(sys.argv[1]))))" + ) + # Popen, not run: the suite-wide hermetic fixture patches subprocess.run. + with subprocess.Popen( + [sys.executable, "-c", code, str(arb._claims_dir)], + stdout=subprocess.PIPE, + text=True, + env={**os.environ, "PYTHONPATH": os.pathsep.join(sys.path)}, + ) as proc: + stdout, _ = proc.communicate(timeout=20) + assert stdout.strip() == "1" + assert proc.returncode == 0 + + def test_unchanged_value_does_not_touch_disk(self, arb: Arbiter) -> None: + """Recovery reports every tick; only a transition is written.""" + with patch("gatelock._arbiter.rewrite_in_place") as rewrite: + arb.set_grab_held(held=False) + assert rewrite.call_count == 0 + arb.set_grab_held(held=True) + written = rewrite.call_count + arb.set_grab_held(held=True) + assert written > 0 + assert rewrite.call_count == written + + def test_holder_file_follows_too(self, arb: Arbiter) -> None: + """The holder.lock copy of the claim is kept in step.""" + assert arb.acquire_holder() + arb.set_grab_held(held=True) + holder = arb._root / "holder.lock" + claim = Claim.from_json(holder.read_text(encoding="utf-8")) + assert claim is not None + assert claim.grab_held is True + + def test_unpublished_and_released_never_raise(self, tmp_path: Path) -> None: + """No handle (or a closed one) is an in-memory update, not a crash.""" + arbiter = Arbiter("x", 1, grab="global", disable_vt=False, runtime_dir=tmp_path) + arbiter.set_grab_held(held=True) + assert arbiter.claim.grab_held is True + arbiter.publish() + arbiter.release() + arbiter.set_grab_held(held=False) + assert arbiter.claim.grab_held is False + + def test_a_closed_handle_is_logged_not_raised( + self, arb: Arbiter, caplog: pytest.LogCaptureFixture + ) -> None: + """A handle closed under us (ValueError) degrades to a warning.""" + assert arb._claim_handle is not None + arb._claim_handle.close() + arb.set_grab_held(held=True) + assert "could not rewrite" in caplog.text + + +class TestHeldGrabs: + """``held_grabs`` lists live claims whose grab is held right now.""" + + def test_filters_unheld_and_grabless(self, tmp_path: Path) -> None: + """Only a held, real grab counts.""" + runtime = tmp_path / "rt" + held, unheld, grabless = ( + _arbiter(runtime, g) for g in ("global", "local", "none") + ) + held.set_grab_held(held=True) + grabless.set_grab_held(held=True) + try: + assert [c.instance_id for c in held_grabs(runtime / "claims")] == [ + held.claim.instance_id + ] + finally: + for arbiter in (held, unheld, grabless): + arbiter.release() + + def test_default_dir_is_the_runtime_claims_dir(self) -> None: + """No argument reads the same dir Arbiter publishes into (hermetic env).""" + arbiter = Arbiter("x", 1, grab="global", disable_vt=False) + arbiter.publish() + arbiter.set_grab_held(held=True) + try: + assert held_grabs() == (arbiter.claim,) + finally: + arbiter.release() + assert held_grabs() == () diff --git a/gatelock/gatelock/tests/test_grab_held_wiring.py b/gatelock/gatelock/tests/test_grab_held_wiring.py new file mode 100644 index 0000000..3496cf7 --- /dev/null +++ b/gatelock/gatelock/tests/test_grab_held_wiring.py @@ -0,0 +1,146 @@ +# Copyright (c) 2026 Krzysztof Rudnicki +"""Tests for WHEN ``grab_held`` flips: only on a grab that actually succeeded. + +The arming sequence reports a held grab after ``grab_set``/``grab_set_global`` +returns; the recovery loop reports each tick's real answer (so a grab a game +stole reads False until it is re-taken); ``LockWindow.report_grab_held`` is the +public hook an app that suspends its lock (leetcode-guard) uses. +""" + +from __future__ import annotations + +from dataclasses import replace +import tkinter as tk +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, call + +from gatelock._config import GrabPolicy, LockConfig +from gatelock._window import LockWindow +from gatelock.tests.conftest import BOTH + +if TYPE_CHECKING: + from gatelock._recovery import RecoveryLoop + from gatelock._surfaces import SurfaceSet + + +def _window(root: MagicMock, config: LockConfig) -> tuple[LockWindow, MagicMock]: + arbiter = MagicMock() + return LockWindow(root, config, MagicMock(), arbiter=arbiter), arbiter + + +class TestArming: + """The first grab reports held only once it succeeded.""" + + def test_global_success_reports_held(self, mock_root: MagicMock) -> None: + """grab_set_global returned: held.""" + window, arbiter = _window(mock_root, LockConfig(mode="hard")) + window._acquire_global_grab(attempt=1) + arbiter.set_grab_held.assert_called_once_with(held=True) + + def test_global_failure_with_retry_reports_nothing( + self, mock_root: MagicMock + ) -> None: + """Still retrying (a game holds it): not held, nothing written.""" + mock_root.grab_set_global.side_effect = tk.TclError("held") + window, arbiter = _window(mock_root, LockConfig(mode="hard")) + window._acquire_global_grab(attempt=1) + arbiter.set_grab_held.assert_not_called() + + def test_local_fallback_reports_held_only_on_success( + self, mock_root: MagicMock + ) -> None: + """retry_ms=0 falls back to a local grab; held iff that worked.""" + mock_root.grab_set_global.side_effect = tk.TclError("held") + config = LockConfig(mode="hard", grab=GrabPolicy(retry_ms=0)) + window, arbiter = _window(mock_root, config) + window._acquire_global_grab(attempt=1) + arbiter.set_grab_held.assert_called_once_with(held=True) + + mock_root.grab_set.side_effect = tk.TclError("also gone") + window, arbiter = _window(mock_root, config) + window._acquire_global_grab(attempt=1) + arbiter.set_grab_held.assert_not_called() + + def test_configured_local_grab(self, mock_root: MagicMock) -> None: + """A local-grab lock reports held after grab_set, not before.""" + config = LockConfig(mode="soft", grab=GrabPolicy(kind="local")) + window, arbiter = _window(mock_root, config) + window.grab_input() + arbiter.set_grab_held.assert_called_once_with(held=True) + + mock_root.grab_set.side_effect = tk.TclError("no") + window, arbiter = _window(mock_root, config) + window.grab_input() + arbiter.set_grab_held.assert_not_called() + + +def _spy(recovery: RecoveryLoop) -> MagicMock: + """Swap the loop's grab-state callback for a mock.""" + spy = MagicMock() + recovery._parts = replace(recovery._parts, on_grab_held=spy) + return spy + + +class TestRecovery: + """Each tick reports whether the grab is really ours.""" + + def test_reports_true_while_held( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Held grab -> True (deduplicated downstream).""" + recovery, enumerator, _ = loop + spy = _spy(recovery) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = mock_root + recovery.tick() + spy.assert_called_once_with(held=True) + + def test_stolen_and_not_retaken_reports_false( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Lost and the re-grab failed: False, the truth.""" + recovery, enumerator, _ = loop + spy = _spy(recovery) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError("game") + assert recovery.tick().grab_reasserted is False + spy.assert_called_once_with(held=False) + + def test_retaken_reports_true( + self, loop: tuple[RecoveryLoop, MagicMock, SurfaceSet], mock_root: MagicMock + ) -> None: + """Lost then re-taken in the same tick: True.""" + recovery, enumerator, _ = loop + spy = _spy(recovery) + enumerator.scan.return_value = BOTH + mock_root.grab_current.return_value = None + assert recovery.tick().grab_reasserted is True + spy.assert_called_once_with(held=True) + + +class TestWindowHook: + """``LockWindow.report_grab_held`` forwards to the arbiter, if any.""" + + def test_forwards_both_directions(self, mock_root: MagicMock) -> None: + """Study mode's release and resume land on the claim.""" + window, arbiter = _window(mock_root, LockConfig(mode="hard")) + window.report_grab_held(held=False) + window.report_grab_held(held=True) + assert arbiter.set_grab_held.call_args_list == [ + call(held=False), + call(held=True), + ] + + def test_recovery_reports_reach_the_arbiter(self, mock_root: MagicMock) -> None: + """The loop the window builds is wired to the same hook.""" + window, arbiter = _window(mock_root, LockConfig(mode="hard")) + mock_root.grab_current.return_value = None + mock_root.grab_set_global.side_effect = tk.TclError("game") + window.recovery._reassert_grab() + arbiter.set_grab_held.assert_called_once_with(held=False) + + def test_no_arbiter_is_a_noop(self, mock_root: MagicMock) -> None: + """Arbitration disabled: nothing to write, nothing raised.""" + window = LockWindow(mock_root, LockConfig(mode="hard"), MagicMock()) + window.report_grab_held(held=True) diff --git a/gatelock/pyproject.toml b/gatelock/pyproject.toml index 5b9187e..34259ef 100644 --- a/gatelock/pyproject.toml +++ b/gatelock/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "gatelock" -version = "0.10.0" +version = "0.11.0" description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps" requires-python = ">=3.14" dependencies = [] # pure stdlib — tkinter is bundled with Python