gatelock 0.8.2: wait_for_x_server waits for as long as it takes

The 60 s deadline was the 2026-09-13 leetcode-guard miss with a smaller
number: a Persistent=true catch-up starts one second into a boot, before X
exists, and a bounded wait that expires exits the unit while systemd has
already stamped the slot as fired. No X server means the user cannot use
the machine either, so an unbounded wait is not a bypass. The wait now
re-states itself at WARNING every five minutes so a long wait never reads
as a healthy, silent unit. `timeout_s` stays as an opt-in for callers that
want a bound.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BxkZDHDmoVSxUT4w72xSiE

gatelock/gatelock/_guards.py

@@ -15,11 +15,21 @@ So the rule now is: **output count never gates arming.** Zero live outputs
 means lock without showing, not decline to lock. This function may only answer
 "can we talk to an X server at all", because without one there is no Tk and
 genuinely nothing to do. It must never ask how many outputs are live.
+
+The clock does not gate arming either (2026-09-13). The wait used to give up
+after 60 s, and a boot-time ``Persistent=true`` catch-up starts one second
+after the user manager, long before X exists: on a slow boot the deadline
+passed, the unit exited, and nothing locked until the next timer slot -- which
+systemd had already marked as fired. No X server means the user cannot use the
+machine either, so waiting forever costs nothing and cannot be a bypass. The
+wait now polls until the server answers and re-states itself at WARNING every
+five minutes, so ``systemctl status`` never shows a healthy, silent unit.
 """
 
 from __future__ import annotations
 
 import logging
+import os
 import sys
 import time
 import tkinter as tk
@@ -30,8 +40,11 @@ if TYPE_CHECKING:
 
 _logger = logging.getLogger(__name__)
 
-_DEFAULT_TIMEOUT_S = 60.0
 _DEFAULT_INTERVAL_S = 1.0
+# How often a still-missing X server is re-stated at WARNING. Same reasoning
+# as ``_queue.QUEUE_HEARTBEAT_SECONDS``: a long silent wait is
+# indistinguishable from a hung unit.
+X_WAIT_HEARTBEAT_S = 300.0
 
 
 def assert_not_under_pytest(what: str) -> None:
@@ -66,7 +79,7 @@ def assert_not_under_pytest(what: str) -> None:
 
 def wait_for_x_server(
     *,
-    timeout_s: float = _DEFAULT_TIMEOUT_S,
+    timeout_s: float | None = None,
     interval_s: float = _DEFAULT_INTERVAL_S,
     sleep: Callable[[float], None] = time.sleep,
     monotonic: Callable[[], float] = time.monotonic,
@@ -80,26 +93,42 @@ def wait_for_x_server(
     is a reason to lock silently, never a reason to skip locking.
 
     Args:
-        timeout_s: Give up after this long.
+        timeout_s: Give up after this long. ``None`` (the default) waits for
+            as long as it takes; see the module docstring for why a bound is
+            the wrong tool here.
         interval_s: Seconds between attempts.
         sleep: Injected for tests.
         monotonic: Injected for tests.
         probe: Injected for tests; defaults to opening a throwaway Tk root.
 
     Returns:
-        True if an X server answered, False on timeout.
+        True once an X server answered, False only if ``timeout_s`` was set
+        and passed.
     """
     attempt = probe if probe is not None else _probe_x_server
-    deadline = monotonic() + timeout_s
+    started = monotonic()
+    next_heartbeat = X_WAIT_HEARTBEAT_S
     while True:
         if attempt():
+            elapsed = monotonic() - started
+            if elapsed >= interval_s:
+                _logger.warning("X server answered after %.0fs; arming now", elapsed)
             return True
-        if monotonic() >= deadline:
+        elapsed = monotonic() - started
+        if timeout_s is not None and elapsed >= timeout_s:
             _logger.error(
                 "no X server answered within %.0fs; cannot build a lock window",
                 timeout_s,
             )
             return False
+        if elapsed >= next_heartbeat:
+            _logger.warning(
+                "no X server on %s after %.0fs -- still waiting, nothing is locked",
+                os.environ.get("DISPLAY", "<unset>"),
+                elapsed,
+            )
+            while next_heartbeat <= elapsed:
+                next_heartbeat += X_WAIT_HEARTBEAT_S
         sleep(interval_s)
 
 

gatelock/gatelock/tests/test_guards.py

@@ -3,18 +3,46 @@
 
 from __future__ import annotations
 
+import logging
 import sys
 import tkinter as tk
+from typing import TYPE_CHECKING
 from unittest.mock import MagicMock, patch
 
 import pytest
 
 from gatelock._guards import (
+    X_WAIT_HEARTBEAT_S,
     _probe_x_server,
     assert_not_under_pytest,
     wait_for_x_server,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class FakeClock:
+    """A monotonic clock that only moves when something sleeps."""
+
+    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 _false_then_true(after: int) -> Iterator[bool]:
+    for _ in range(after):
+        yield False
+    while True:
+        yield True
+
 
 class TestAssertNotUnderPytest:
     """Refuse to build a real grabbing window inside a test run."""
@@ -59,18 +87,73 @@ class TestWaitForXServer:
         )
         assert sleeps == [1.0, 1.0]
 
-    def test_times_out(self) -> None:
-        """A display that never arrives gives up and says so."""
-        clock = iter([0.0, 100.0])
-        assert (
-            wait_for_x_server(
-                probe=lambda: False,
-                sleep=lambda _s: None,
-                monotonic=lambda: next(clock),
-                timeout_s=1.0,
+    def test_an_explicit_timeout_gives_up_and_says_so(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Opt-in only: a caller that asks for a bound gets one."""
+        clock = FakeClock()
+        with caplog.at_level(logging.ERROR):
+            assert (
+                wait_for_x_server(
+                    probe=lambda: False,
+                    sleep=clock.sleep,
+                    monotonic=clock.now,
+                    timeout_s=2.5,
+                )
+                is False
             )
-            is False
-        )
+        assert clock.slept == [1.0, 1.0, 1.0]
+        assert "cannot build a lock window" in caplog.text
+
+    def test_the_default_wait_outlives_the_old_sixty_second_deadline(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """The 2026-09-13 rule: time never gates arming. A server that takes
+        ten minutes to appear is still waited for, and the wait says so."""
+        clock = FakeClock()
+        probe = _false_then_true(after=600)
+        with caplog.at_level(logging.WARNING):
+            assert (
+                wait_for_x_server(
+                    probe=lambda: next(probe),
+                    sleep=clock.sleep,
+                    monotonic=clock.now,
+                )
+                is True
+            )
+        assert len(clock.slept) == 600
+        assert "X server answered after 600s; arming now" in caplog.text
+
+    def test_a_long_wait_heartbeats_at_warning(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """A silent wait looks like a hung unit; it must re-state itself."""
+        clock = FakeClock()
+        probe = _false_then_true(after=int(X_WAIT_HEARTBEAT_S * 2) + 1)
+        with caplog.at_level(logging.WARNING):
+            wait_for_x_server(
+                probe=lambda: next(probe), sleep=clock.sleep, monotonic=clock.now
+            )
+        heartbeats = [r for r in caplog.records if "still waiting" in r.message]
+        assert len(heartbeats) == 2
+        assert all(r.levelno == logging.WARNING for r in heartbeats)
+        assert "nothing is locked" in heartbeats[0].message
+
+    def test_a_heartbeat_skips_ahead_after_a_long_sleep(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """One coarse sleep past several heartbeat slots logs once, not once
+        per missed slot, and the next heartbeat lands on a future slot."""
+        clock = FakeClock()
+        probe = _false_then_true(after=2)
+        with caplog.at_level(logging.WARNING):
+            wait_for_x_server(
+                probe=lambda: next(probe),
+                sleep=clock.sleep,
+                monotonic=clock.now,
+                interval_s=X_WAIT_HEARTBEAT_S * 3,
+            )
+        assert sum("still waiting" in r.message for r in caplog.records) == 1
 
     def test_zero_live_outputs_does_not_block_arming(self) -> None:
         """THE 2b rule: a connected X server with dark monitors still arms.

gatelock/pyproject.toml

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