gatelock: promote the priority queue from leetcode_guard

wait_for_turn/stronger_claims already fixed leetcode_guard's grab-race
with screen_locker; screen_locker and diet_guard never got the same
fix, so a stray re-arm race between them was still live (confirmed
2026-08-21: diet_guard held the global grab 07:04-09:07 while
screen_locker built a competing window and retried its own grab
forever instead of queueing).

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

gatelock/gatelock/__init__.py

@@ -36,6 +36,13 @@ from gatelock._outputs import (
     enumerate_outputs,
     parse_xrandr_query,
 )
+from gatelock._queue import (
+    QUEUE_DEADLINE_SECONDS,
+    QUEUE_POLL_SECONDS,
+    QueueResult,
+    stronger_claims,
+    wait_for_turn,
+)
 from gatelock._recovery import RecoveryLoop, RecoveryReport
 from gatelock._root import GateRoot
 from gatelock._scrollable import ScrollableSurface
@@ -68,6 +75,8 @@ from gatelock.widgets import (
 
 __all__ = [
     "DEFAULT_WRAP",
+    "QUEUE_DEADLINE_SECONDS",
+    "QUEUE_POLL_SECONDS",
     "RANK_DIET_GUARD",
     "RANK_SCREEN_LOCKER",
     "RANK_WAKE_ALARM",
@@ -90,6 +99,7 @@ __all__ = [
     "OutputEnumerator",
     "OutputRect",
     "OutputScan",
+    "QueueResult",
     "RandrBackend",
     "RecoveryLoop",
     "RecoveryReport",
@@ -119,5 +129,7 @@ __all__ = [
     "report_fit",
     "restore_vt_switching",
     "row",
+    "stronger_claims",
+    "wait_for_turn",
     "wait_for_x_server",
 ]

gatelock/gatelock/_queue.py

@@ -0,0 +1,138 @@
+"""Wait our turn instead of fighting for the screen.
+
+The lockers form a ladder (see ``_arbiter.py``'s module docstring: wake up,
+then work out, then grind, then eat), but ranking alone does not stop two
+apps drawing competing windows at once. Every consumer publishes a claim,
+calls ``acquire_holder()``, and used to discard the result and build its
+window anyway -- the loser draws a visible, ungrabbed window *behind* the
+winner and spins in gatelock's forever-retrying grab loop.
+
+So this waits **headlessly**: no root, no surfaces, nothing on screen, until
+no live claim outranks us. Then the caller arms normally.
+
+Two rules it must not break:
+
+* **Never exit because another lock is running.** Standing down permanently
+  would turn "start the workout lock" into a way to skip the grind.
+* **Never give up and leave the machine unlocked.** The deadline is a runaway
+  backstop for a wait that should have ended; reaching it arms anyway and says
+  so at ERROR.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+import time
+from typing import TYPE_CHECKING, Final
+
+_logger: Final = logging.getLogger(__name__)
+
+QUEUE_POLL_SECONDS: Final = 2.0
+QUEUE_DEADLINE_SECONDS: Final = 6 * 60 * 60
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from gatelock._arbiter import Arbiter, Claim
+
+
+@dataclass(frozen=True)
+class QueueResult:
+    """How the wait ended."""
+
+    waited_seconds: float
+    blocked_by: tuple[str, ...]
+    """Apps we queued behind, in the order they were first seen."""
+
+    timed_out: bool
+    """``True`` means the deadline was hit and we armed anyway."""
+
+    @property
+    def queued(self) -> bool:
+        """Whether we actually had to wait for anything."""
+        return bool(self.blocked_by)
+
+
+def stronger_claims(arbiter: Arbiter) -> tuple[Claim, ...]:
+    """Live claims that outrank ours.
+
+    Rank alone, deliberately -- not gatelock's strength comparison. We are
+    *queueing*, not standing down: a higher-ranked app owns the screen right
+    now whether or not it grabs as hard as we would, and drawing over it is
+    the behaviour this module exists to remove.
+    """
+    mine = arbiter.claim.instance_id
+    return tuple(
+        claim
+        for claim in arbiter.live_claims()
+        if claim.instance_id != mine and claim.rank > arbiter.claim.rank
+    )
+
+
+def wait_for_turn(
+    arbiter: Arbiter,
+    *,
+    poll: float = QUEUE_POLL_SECONDS,
+    deadline: float = QUEUE_DEADLINE_SECONDS,
+    sleep: Callable[[float], None] | None = None,
+    now: Callable[[], float] | None = None,
+) -> QueueResult:
+    """Block, showing nothing, until no stronger locker holds the screen.
+
+    Args:
+        arbiter: Our own arbiter, already published so that lower-ranked apps
+            queue behind us in turn.
+        poll: Seconds between checks. Liveness comes from ``flock``, so a
+            SIGKILLed incumbent is noticed on the very next tick.
+        deadline: Runaway backstop in seconds.
+        sleep: Injected for tests.
+        now: Injected monotonic clock, for tests.
+
+    Returns:
+        A record of the wait.
+    """
+    rest = sleep if sleep is not None else time.sleep
+    clock = now if now is not None else time.monotonic
+
+    started = clock()
+    seen: list[str] = []
+    while True:
+        blockers = stronger_claims(arbiter)
+        if not blockers:
+            waited = clock() - started
+            if seen:
+                _logger.warning(
+                    "%s waited %.0fs behind %s; arming now",
+                    arbiter.claim.app,
+                    waited,
+                    ", ".join(seen),
+                )
+            return QueueResult(
+                waited_seconds=waited, blocked_by=tuple(seen), timed_out=False
+            )
+
+        for claim in blockers:
+            if claim.app not in seen:
+                seen.append(claim.app)
+                _logger.info(
+                    "%s is queued behind %s (rank %d) -- waiting with no window",
+                    arbiter.claim.app,
+                    claim.app,
+                    claim.rank,
+                )
+
+        elapsed = clock() - started
+        if elapsed >= deadline:
+            _logger.error(
+                "%s waited %.0fs behind %s and gave up waiting -- arming "
+                "anyway rather than leaving the machine unlocked",
+                arbiter.claim.app,
+                elapsed,
+                ", ".join(seen),
+            )
+            return QueueResult(
+                waited_seconds=elapsed, blocked_by=tuple(seen), timed_out=True
+            )
+
+        rest(poll)

gatelock/gatelock/tests/test_queue.py

@@ -0,0 +1,143 @@
+"""Tests for the layered wait."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from gatelock import RANK_DIET_GUARD, RANK_SCREEN_LOCKER, RANK_WAKE_ALARM, Arbiter
+from gatelock._queue import stronger_claims, wait_for_turn
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest
+
+_RANK_LEETCODE_GUARD = 150
+
+
+def make_arbiter(app: str, rank: int, runtime: Path) -> Arbiter:
+    arbiter = Arbiter(app, rank, grab="global", disable_vt=True, runtime_dir=runtime)
+    arbiter.publish()
+    return arbiter
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.value = 0.0
+        self.slept: list[float] = []
+
+    def now(self) -> float:
+        return self.value
+
+    def sleep(self, seconds: float) -> None:
+        self.slept.append(seconds)
+        self.value += seconds
+
+
+def test_an_empty_ladder_arms_immediately(tmp_path: Path) -> None:
+    arbiter = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+    clock = FakeClock()
+
+    result = wait_for_turn(arbiter, sleep=clock.sleep, now=clock.now)
+
+    assert not result.queued
+    assert not result.timed_out
+    assert clock.slept == []
+    arbiter.release()
+
+
+def test_we_queue_behind_the_workout_lock_and_arm_when_it_finishes(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Your rule, exactly: the workout lock shows first, then this one -- and
+    this one never exits just because that one is running."""
+    workout = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path)
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+    clock = FakeClock()
+
+    released = {"done": False}
+
+    def sleep(seconds: float) -> None:
+        clock.sleep(seconds)
+        if not released["done"]:
+            workout.release()
+            released["done"] = True
+
+    with caplog.at_level(logging.INFO):
+        result = wait_for_turn(ours, sleep=sleep, now=clock.now)
+
+    assert result.queued
+    assert result.blocked_by == ("screen_locker",)
+    assert not result.timed_out
+    assert result.waited_seconds > 0
+    ours.release()
+
+
+def test_a_lower_ranked_locker_is_not_waited_for(tmp_path: Path) -> None:
+    """diet_guard sits below us, so it queues behind us -- not the reverse."""
+    diet = make_arbiter("diet_guard", RANK_DIET_GUARD, tmp_path)
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+    clock = FakeClock()
+
+    result = wait_for_turn(ours, sleep=clock.sleep, now=clock.now)
+
+    assert not result.queued
+    diet.release()
+    ours.release()
+
+
+def test_every_higher_ranked_app_is_named_once(tmp_path: Path) -> None:
+    alarm = make_arbiter("wake_alarm", RANK_WAKE_ALARM, tmp_path)
+    workout = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path)
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+    clock = FakeClock()
+
+    def sleep(seconds: float) -> None:
+        clock.sleep(seconds)
+        alarm.release()
+        workout.release()
+
+    result = wait_for_turn(ours, sleep=sleep, now=clock.now)
+
+    assert set(result.blocked_by) == {"wake_alarm", "screen_locker"}
+    ours.release()
+
+
+def test_the_deadline_arms_anyway_rather_than_leaving_the_pc_unlocked(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The backstop must never mean "give up and let them through"."""
+    workout = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path)
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+    clock = FakeClock()
+
+    with caplog.at_level(logging.ERROR):
+        result = wait_for_turn(
+            ours, poll=10.0, deadline=30.0, sleep=clock.sleep, now=clock.now
+        )
+
+    assert result.timed_out
+    assert result.blocked_by == ("screen_locker",)
+    assert any(record.levelno == logging.ERROR for record in caplog.records)
+    workout.release()
+    ours.release()
+
+
+def test_stronger_claims_ignores_our_own(tmp_path: Path) -> None:
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+
+    assert stronger_claims(ours) == ()
+    ours.release()
+
+
+def test_a_dead_holder_stops_blocking_us(tmp_path: Path) -> None:
+    """Liveness is the kernel's flock, so a SIGKILLed locker is noticed on the
+    next tick -- no heartbeats, no staleness heuristics."""
+    workout = make_arbiter("screen_locker", RANK_SCREEN_LOCKER, tmp_path)
+    ours = make_arbiter("leetcode_guard", _RANK_LEETCODE_GUARD, tmp_path)
+
+    assert len(stronger_claims(ours)) == 1
+    workout.release()
+    assert stronger_claims(ours) == ()
+    ours.release()

gatelock/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "gatelock"
-version = "0.5.0"
+version = "0.6.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