test(deadline): pin review cases for poll_until callers and the sweep

From review: document that poll_until bounds only its sleeps, not time
spent in the predicate; pin that unlink_files_older_than keeps sweeping
after one unlink raises PermissionError; pin that
_wait_for_user_dbus_socket exports the systemd env exactly once when
the socket appears only on the final check, made at the deadline after
a cut last sleep. Both new tests fail against a mutated implementation
(the per-file suppress removed; main's loop shape).

agent/deadline.py

@@ -143,6 +143,8 @@ def poll_until(predicate: Callable[[], _T], timeout: float, interval: float) ->
 
     The first check runs at once and the last sleep is cut to the time left, so a condition that
     holds when the deadline arrives still counts. ``timeout <= 0`` means exactly one check.
+    Only the sleeps are bounded: time spent inside ``predicate`` is not, so a predicate that blocks
+    (socket probes, subprocess calls) can carry the final check past the deadline.
     """
     deadline = time.monotonic() + timeout
     while True:

tests/hermes_cli/test_gateway_service_dbus_wait.py

@@ -0,0 +1,23 @@
+"""``_wait_for_user_dbus_socket``: readiness found on poll_until's final check still exports the env."""
+
+import pytest
+
+pytest.importorskip("pwd")  # POSIX systemd helper, same gate as test_gateway_service
+
+import hermes_cli.gateway as gateway_cli
+
+
+def test_ready_only_on_final_check_exports_env_once(monkeypatch):
+    calls, slept = [], []
+    checks = iter([False, True])
+    # Deadline at 1.0. After the first miss the clock reads 0.9, so the last sleep is cut to 0.1
+    # and the second check is the final one, made at the deadline.
+    clock = iter([0.0, 0.9, 1.0])
+    monkeypatch.setattr(gateway_cli, "_ensure_user_systemd_env", lambda: calls.append("env"))
+    monkeypatch.setattr(gateway_cli, "_user_systemd_socket_ready", lambda: next(checks))
+    monkeypatch.setattr("agent.deadline.time.monotonic", lambda: next(clock))
+    monkeypatch.setattr("agent.deadline.time.sleep", slept.append)
+
+    assert gateway_cli._wait_for_user_dbus_socket(timeout=1.0) is True
+    assert slept == [pytest.approx(0.1)]
+    assert calls == ["env"]

tests/test_utils_unlink_files_older_than.py

@@ -2,6 +2,7 @@
 
 import os
 import time
+from pathlib import Path
 
 from utils import unlink_files_older_than
 
@@ -28,3 +29,21 @@ def test_removes_only_matching_files_older_than_the_age(tmp_path):
 
 def test_missing_directory_is_zero_not_an_error(tmp_path):
     assert unlink_files_older_than(tmp_path / "absent", "*", 0) == 0
+
+
+def test_unlink_failure_skips_that_file_and_keeps_sweeping(tmp_path, monkeypatch):
+    now = time.time()
+    stuck = _aged(tmp_path / "a.json", 120, now)
+    later = _aged(tmp_path / "b.json", 120, now)
+    real_unlink = Path.unlink
+
+    def unlink(self, *args, **kwargs):
+        if self.name == "a.json":
+            raise PermissionError("locked")
+        return real_unlink(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", unlink)
+    monkeypatch.setattr(Path, "glob", lambda self, pattern: iter([self / "a.json", self / "b.json"]))  # fail first
+
+    assert unlink_files_older_than(tmp_path, "*.json", 60, now=now) == 1
+    assert stuck.exists() and not later.exists()