gatelock: preempt a weaker holder blocking a stronger claim's grab

wait_for_turn only helps a weaker locker arriving after a stronger
one; it does nothing when the weaker one is already the incumbent.
On 2026-08-21 diet_guard (rank 100) held the grab for ~3 minutes
against a newly-arriving screen_locker (rank 200), which just
retried and logged instead of ever making progress.

Fix: when a grab is blocked by a holder that outranks-us-not, SIGTERM
it once. The holder's own signal handler already converts SIGTERM
into its normal close() teardown (VT restore, arbiter release,
app-specific on_close) -- this reuses that path rather than adding a
new one, and never signals upward (never disturbs an incumbent that
outranks us), so it cannot become a way to bypass a stronger lock.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018M5gKyYzxWG6QkCaj1wWfk

gatelock/gatelock/_window.py

@@ -28,6 +28,7 @@ import atexit
 import contextlib
 from dataclasses import dataclass
 import logging
+import os
 import signal
 import tkinter as tk
 from typing import TYPE_CHECKING, Literal, Protocol
@@ -43,7 +44,7 @@ from gatelock._vt import disable_vt_switching, restore_vt_switching
 if TYPE_CHECKING:
     from types import FrameType
 
-    from gatelock._arbiter import Arbiter
+    from gatelock._arbiter import Arbiter, Claim
     from gatelock._surfaces import SurfaceInfo
 
 _logger = logging.getLogger(__name__)
@@ -86,6 +87,12 @@ class LockConfig:
             fails. 0 means "try once, then fall back to a local grab". Left
             unset (None), a "global" grab retries forever every 200ms.
         grab_log_every: Log a warning every N failed retry-forever attempts.
+        preempt_weaker_holder: SIGTERM a lower-ranked incumbent that is
+            blocking our grab, instead of retrying against it forever. Only
+            the *direction* the arbiter's own ranking already sanctions: we
+            never signal a holder that outranks us. SIGTERM (not SIGKILL) so
+            the holder's own signal handler runs its normal close() path --
+            same teardown as a clean dismiss, just triggered externally.
         app_name: This app's name, used in arbitration logs so a blocked app
             can say who is actually holding the screen.
         rank: Arbitration priority; higher wins. See the ``RANK_*`` constants
@@ -130,6 +137,7 @@ class LockConfig:
     disable_vt: bool | None = None
     grab_retry_ms: int | None = None
     grab_log_every: int = 25
+    preempt_weaker_holder: bool = True
     app_name: str = "gatelock"
     rank: int = RANK_SCREEN_LOCKER
     recovery_tick_ms: int = 1000
@@ -310,6 +318,7 @@ class LockWindow:
         self._vt_disabled = False
         self._closed = False
         self._focus_notified = False
+        self._preempted_pids: set[int] = set()
         self._surfaces = SurfaceSet(root, config, hooks)
         self._enumerator = OutputEnumerator(root)
         self._detector = OutputChangeDetector(root)
@@ -419,6 +428,13 @@ class LockWindow:
         v0.1.1 always blamed "a fullscreen game". On 2026-07-25 the holder was
         the other locker, and that guess sent the diagnosis in the wrong
         direction for the length of the outage.
+
+        A holder that outranks us is left alone entirely -- retrying is the
+        correct behavior there, matching :func:`gatelock.wait_for_turn`'s own
+        rule for an app that has not armed yet. A *weaker* holder is preempted
+        below, once per pid, rather than left to block us for as long as it
+        takes to finish on its own (diet_guard held the grab for ~3 minutes
+        against a higher-ranked screen_locker on 2026-08-21).
         """
         holder = self._arbiter.describe_holder() if self._arbiter else None
         if holder is None:
@@ -437,6 +453,41 @@ class LockWindow:
             holder.pid,
             holder.started,
         )
+        self._maybe_preempt(holder)
+
+    def _maybe_preempt(self, holder: Claim) -> None:
+        """SIGTERM a weaker holder once, so it stands down instead of blocking.
+
+        SIGTERM (not SIGKILL): the holder's own signal handler
+        (``LockWindow._install_signal_handlers``) raises ``SystemExit(0)``,
+        which unwinds through its ``run()``'s ``finally`` into its own
+        ``close()`` -- the identical teardown a clean dismiss gets (VT
+        restored, arbiter claim released, app-specific ``on_close`` hook
+        run), just triggered externally instead of by the user.
+        """
+        if not self._config.preempt_weaker_holder or self._arbiter is None:
+            return
+        if holder.rank >= self._arbiter.claim.rank:
+            return
+        if holder.pid in self._preempted_pids:
+            return
+        self._preempted_pids.add(holder.pid)
+        _logger.warning(
+            "preempting weaker holder %r (rank %d, pid %d) so rank %d can arm",
+            holder.app,
+            holder.rank,
+            holder.pid,
+            self._arbiter.claim.rank,
+        )
+        try:
+            os.kill(holder.pid, signal.SIGTERM)
+        except ProcessLookupError:
+            _logger.info(
+                "holder pid %d was already gone by the time we signalled it",
+                holder.pid,
+            )
+        except OSError:
+            _logger.exception("could not signal holder pid %d", holder.pid)
 
     def _notify_focus_ready(self) -> None:
         """Tell the app it can focus its first input widget now.

gatelock/gatelock/tests/test_window_surfaces.py

@@ -3,9 +3,10 @@
 from __future__ import annotations
 
 import queue
+import signal
 from unittest.mock import MagicMock, patch
 
-from gatelock._arbiter import RANK_DIET_GUARD, Claim
+from gatelock._arbiter import RANK_DIET_GUARD, RANK_SCREEN_LOCKER, Claim
 from gatelock._detect import _RandrEventSource
 from gatelock._outputs import Output, OutputRect
 from gatelock._window import LockConfig
@@ -61,6 +62,15 @@ class TestGrabBlockedLogging:
     def test_names_the_holding_gatelock_app(self, mock_root: MagicMock) -> None:
         """The 2026-07-25 diagnosis, available immediately."""
         arbiter = MagicMock()
+        arbiter.claim = Claim(
+            app="screen_locker",
+            rank=RANK_SCREEN_LOCKER,
+            pid=1,
+            started="2026-07-25T12:49:40+00:00",
+            grab="global",
+            disable_vt=True,
+            instance_id="ours",
+        )
         arbiter.describe_holder.return_value = Claim(
             app="diet_guard",
             rank=RANK_DIET_GUARD,
@@ -70,7 +80,9 @@ class TestGrabBlockedLogging:
             disable_vt=True,
             instance_id="t",
         )
-        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window, _hooks = make_window(
+            mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=False)
+        )
         window._arbiter = arbiter
         with patch("gatelock._window._logger") as logger:
             window._log_grab_blocked(25)
@@ -96,6 +108,123 @@ class TestGrabBlockedLogging:
         assert "fullscreen game" in str(logger.warning.call_args)
 
 
+class TestPreemptWeakerHolder:
+    """A weaker incumbent must not block a stronger claim indefinitely."""
+
+    def _arbiter_with(
+        self, *, our_rank: int, holder_rank: int, holder_pid: int
+    ) -> MagicMock:
+        arbiter = MagicMock()
+        arbiter.claim = Claim(
+            app="ours",
+            rank=our_rank,
+            pid=1,
+            started="2026-08-21T12:00:00+00:00",
+            grab="global",
+            disable_vt=True,
+            instance_id="ours",
+        )
+        arbiter.describe_holder.return_value = Claim(
+            app="holder",
+            rank=holder_rank,
+            pid=holder_pid,
+            started="2026-08-21T07:04:07+00:00",
+            grab="global",
+            disable_vt=True,
+            instance_id="theirs",
+        )
+        return arbiter
+
+    def test_signals_a_weaker_holder(self, mock_root: MagicMock) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
+        )
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window._arbiter = arbiter
+        with patch("gatelock._window.os.kill") as kill:
+            window._log_grab_blocked(25)
+        kill.assert_called_once_with(4075, signal.SIGTERM)
+
+    def test_never_signals_a_stronger_holder(self, mock_root: MagicMock) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_DIET_GUARD, holder_rank=RANK_SCREEN_LOCKER, holder_pid=4137
+        )
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window._arbiter = arbiter
+        with patch("gatelock._window.os.kill") as kill:
+            window._log_grab_blocked(25)
+        kill.assert_not_called()
+
+    def test_signals_the_same_holder_only_once(self, mock_root: MagicMock) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
+        )
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window._arbiter = arbiter
+        with patch("gatelock._window.os.kill") as kill:
+            window._log_grab_blocked(25)
+            window._log_grab_blocked(50)
+        kill.assert_called_once_with(4075, signal.SIGTERM)
+
+    def test_disabled_by_config(self, mock_root: MagicMock) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
+        )
+        window, _hooks = make_window(
+            mock_root, config=LockConfig(mode="hard", preempt_weaker_holder=False)
+        )
+        window._arbiter = arbiter
+        with patch("gatelock._window.os.kill") as kill:
+            window._log_grab_blocked(25)
+        kill.assert_not_called()
+
+    def test_no_arbiter_means_no_preemption(self, mock_root: MagicMock) -> None:
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        with patch("gatelock._window.os.kill") as kill:
+            window._maybe_preempt(
+                Claim(
+                    app="holder",
+                    rank=RANK_DIET_GUARD,
+                    pid=4075,
+                    started="2026-08-21T07:04:07+00:00",
+                    grab="global",
+                    disable_vt=True,
+                    instance_id="theirs",
+                )
+            )
+        kill.assert_not_called()
+
+    def test_a_holder_that_is_already_gone_is_reported_at_info(
+        self, mock_root: MagicMock
+    ) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
+        )
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window._arbiter = arbiter
+        with (
+            patch("gatelock._window.os.kill", side_effect=ProcessLookupError),
+            patch("gatelock._window._logger") as logger,
+        ):
+            window._log_grab_blocked(25)
+        assert logger.info.called
+
+    def test_an_unsignalable_holder_is_reported_not_raised(
+        self, mock_root: MagicMock
+    ) -> None:
+        arbiter = self._arbiter_with(
+            our_rank=RANK_SCREEN_LOCKER, holder_rank=RANK_DIET_GUARD, holder_pid=4075
+        )
+        window, _hooks = make_window(mock_root, config=LockConfig(mode="hard"))
+        window._arbiter = arbiter
+        with (
+            patch("gatelock._window.os.kill", side_effect=OSError("nope")),
+            patch("gatelock._window._logger") as logger,
+        ):
+            window._log_grab_blocked(25)
+        assert logger.exception.called
+
+
 class TestFocusNotifiedOnce:
     """Both call sites are load-bearing, so the guard is on the callee."""
 

gatelock/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "gatelock"
-version = "0.6.0"
+version = "0.7.0"
 description = "Shared per-output lock-window + HMAC log-integrity backend for blocking-overlay apps"
 requires-python = ">=3.10"
 dependencies = []  # pure stdlib — tkinter is bundled with Python