gatelock 0.11.0: Claim.grab_held + held_grabs(), so a reader can tell a locker's X grab is actually held

A live claim was not "input is locked": it is published before the window
grabs, and leetcode-guard's study mode releases the grab while the claim
lives on. leisure-watch (contract C2) must pause shadow billing only while a
grab is really held.

- Claim.grab_held (always serialised; missing key from an old writer reads
  as grab != "none"). Arbitration still uses the declared grab only.
- Arbiter.set_grab_held(held=) rewrites the claim and holder copies in place
  through the owner's flocked fd (O_APPEND cleared, pwrite at 0, padded so
  the file never shrinks) -- same inode, flock liveness untouched. Never raises.
- Set true only after grab_set/grab_set_global succeeded; RecoveryLoop
  reports each tick's real answer (a stolen, un-retaken grab reads false).
  LockWindow.report_grab_held(held=) is the hook for deliberate releases.
- gatelock.held_grabs(claims_dir=None): live claims with grab_held and a real
  grab. Root callers pass /run/user/1000/gatelock/claims.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsh4Znwh2KnC7q6vbGpX1N

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

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",

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.
 

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()

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")

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

gatelock/gatelock/_recovery.py

@@ -193,13 +193,15 @@ class RecoveryLoop:
         """
         if self._parts.config.resolved_grab() != "global":
             return False
-        if self.holds_grab():
+        held = self.holds_grab()
-            return False
+        reasserted = False
-        with contextlib.suppress(tk.TclError):
+        if not held:
-            self._root.grab_set_global()
+            with contextlib.suppress(tk.TclError):
-            _logger.warning("global grab had been lost; re-acquired it")
+                self._root.grab_set_global()
-            return True
+                _logger.warning("global grab had been lost; re-acquired it")
-        return False
+                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.

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``)."""

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
+Arming order (**arm first, render second**) is explained in
-and the VT-disable happen regardless of how many outputs are live, and the
+:mod:`gatelock._arming`.
-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.
 """
 
 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:

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() == ()

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)

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