fix: the planned-stop consumer takes the pause lock only for a pause on disk and drops settled receipts

consume_planned_stop_marker_for_self took .hermes-update-paused-gateways.lock
on every platform for every stop (up to 2 s when busy), creating the lock
file in the root home, and on a busy lock fell back to keeping the request
and writing a .accepted receipt that nothing ever removed. The lock only
protects the checkpoint into a Windows update pause record; with no record
or claim of this checkout on disk (every non-Windows host, or a pause
recovery/the updater already retired) there is nothing to checkpoint.

The consumer now holds the lock only when a pause is on disk. The
checkpoint hook re-checks after the request was read, so a pause that
appeared after the first look (an update writes its record before its
request) is still checkpointed under the lock. With no pause on disk, the
consumer removes the .accepted receipt and an expired stamped request an
earlier drain left: their debt is settled. While any pause is on disk both
stay (never earlier).

test_a_busy_pause_mutex_never_turns_a_planned_stop_into_an_unplanned_one
now puts a pause record on disk first: without one, keeping the request on
a busy lock was the leftover this fixes; with one, its assertions (planned,
request retained, no block) are unchanged.

gateway/status.py

@@ -1682,7 +1682,7 @@ def _read_live_pid_marker(path: Path, ttl_s: int) -> Optional[tuple[dict[str, An
         return None
     target_pid = _pid_from_record(record, "target_pid")
     if target_pid is None or _marker_is_stale(record.get("written_at") or "", ttl_s):
-        if record.get("accepted") is not True:
+        if record.get("accepted") is not True or not _update_pause_on_disk():
             _unlink_quietly(path)
         return None
     return record, target_pid, record.get("target_start_time")
@@ -1961,17 +1961,45 @@ def write_planned_stop_marker(target_pid: int) -> bool:
 _PLANNED_STOP_MUTEX_WAIT_S = 2.0
 
 
+def _update_pause_on_disk() -> bool:
+    """A Windows update pause of this checkout is on disk (its record or a recovery's claim), or the
+    record directory cannot be listed. Only then does a consume need the pause lock and a checkpoint,
+    and only then can accepted-stop evidence beside a request still be owed: with none, every pause
+    was settled by recovery or the updater (or never made: every non-Windows host)."""
+    from hermes_cli import update_pause_record
+    record = update_pause_record.record_path()
+    try:
+        with os.scandir(record.parent) as entries:
+            return any(e.name == record.name or (e.name.startswith(record.name + ".") and e.name.endswith(".claim"))
+                       for e in entries)
+    except FileNotFoundError:
+        return False
+    except OSError:
+        return True
+
+
+def _checkpoint_planned_stop(path: Path, record: dict[str, Any]) -> None:
+    """``on_consume``, decided after the request was read: an update writes its record before its
+    request, so a pause that appeared since the consumer first looked is checkpointed too."""
+    from hermes_cli import update_pause_record
+    if _update_pause_on_disk():
+        with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S):
+            update_pause_record.mark_stop_consumed(path, record)
+
+
 def consume_planned_stop_marker_for_self() -> bool:
     """Return True when the current process is being intentionally stopped."""
     from hermes_cli import update_pause_record
-    # Recovery must see either the request or its checkpoint, never the gap
-    # between validating/consuming the marker and scheduling asynchronous stop.
+    path = _get_planned_stop_marker_path()
     try:
-        with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S):
-            return _consume_pid_marker_for_self(
-                _get_planned_stop_marker_path(), ttl_s=_PLANNED_STOP_MARKER_TTL_S,
-                on_consume=update_pause_record.mark_stop_consumed,
-            )
+        paused = _update_pause_on_disk()
+        if not paused:  # an earlier drain's receipt: its debt is settled
+            _unlink_quietly(update_pause_record._accepted_path(path))
+        # With a pause on disk, recovery must see either the request or its checkpoint, never the
+        # gap between validating/consuming the marker and scheduling asynchronous stop.
+        with update_pause_record._mutex(_PLANNED_STOP_MUTEX_WAIT_S) if paused else contextlib.nullcontext():
+            return _consume_pid_marker_for_self(path, ttl_s=_PLANNED_STOP_MARKER_TTL_S,
+                                                on_consume=_checkpoint_planned_stop)
     except OSError as exc:
         # The pause bookkeeping (a busy mutex, a checkpoint that cannot be written) never decides
         # whether this stop was planned: classify without consuming, and leave a receipt beside the

tests/gateway/test_planned_stop_accepted_debt.py

@@ -10,6 +10,8 @@ from __future__ import annotations
 import json
 import os
 import sys
+import threading
+import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
@@ -80,3 +82,48 @@ def test_a_stamped_accepted_request_outlives_its_ttl_while_the_pause_is_owed(tmp
         assert status.planned_stop_marker_targets_self() is False
     assert marker.exists(), f"the {second_look} deleted the stamped accepted request"
     assert _owed(saved["pause_id"]) == {"p": os.getpid()}, "an accepted stop lost its restart debt"
+
+
+@pytest.mark.parametrize("home_writable", [True, False], ids=["receipt", "stamped_request"])
+def test_accepted_stop_evidence_is_removed_once_its_pause_is_settled_and_never_before(
+        tmp_path, monkeypatch, home_writable):
+    """``.accepted`` receipts and stamped requests are cleaned up by the next consume once no pause of
+    this checkout is on disk (recovery or the updater retired it), and kept while one still is."""
+    marker, saved = _accepted_unrecorded(tmp_path, monkeypatch, home_writable=home_writable)
+    evidence = pause_record._accepted_path(marker) if home_writable else marker
+    status.consume_planned_stop_marker_for_self()
+    assert evidence.exists(), "accepted-stop evidence removed while its pause is still owed"
+    pause_record.discharge(saved)
+    assert not pause_record.record_path().exists(), "premise: the pause is settled"
+    status.consume_planned_stop_marker_for_self()
+    assert not pause_record._accepted_path(marker).exists(), "a settled receipt was left behind"
+    assert not marker.exists(), "a settled stamped request was left behind"
+
+
+def test_with_no_pause_on_disk_the_consumer_never_waits_on_the_pause_lock(tmp_path, monkeypatch):
+    """No pause record (every non-Windows host, or a settled pause): a plain ``hermes gateway stop`` is
+    consumed without the pause lock, so another holder costs it no wait and leaves no request or
+    ``.accepted`` receipt behind."""
+    _home(tmp_path, monkeypatch)
+    held, release = threading.Event(), threading.Event()
+
+    def holder():
+        with pause_record._mutex():
+            held.set()
+            release.wait(10)
+
+    thread = threading.Thread(target=holder)
+    thread.start()
+    try:
+        assert held.wait(5), "premise: another thread holds the pause lock"
+        marker = status._get_planned_stop_marker_path()
+        assert status.write_planned_stop_marker(os.getpid())
+        started = time.monotonic()
+        assert status.consume_planned_stop_marker_for_self() is True
+        elapsed = time.monotonic() - started
+    finally:
+        release.set()
+        thread.join()
+    assert elapsed < 1.0, f"the consumer waited {elapsed:.2f}s on a pause lock no pause needed"
+    assert not marker.exists(), "the consumed request was left on disk"
+    assert not pause_record._accepted_path(marker).exists(), "an .accepted receipt was written with no pause"

tests/hermes_cli/test_update_pause_boundaries.py

@@ -169,6 +169,7 @@ def test_failed_consumer_checkpoint_keeps_request_and_user_stop_is_not_update_de
 def test_a_busy_pause_mutex_never_turns_a_planned_stop_into_an_unplanned_one(cell):
     base, _, launch = cell
     holder = launch('''
+        r.write({"pause_id": "busy", "resume_needed": True})  # a pause on disk: the consume needs the mutex
         with r._mutex():
             Path(sys.argv[2]).touch()
             while not Path(sys.argv[3]).exists(): time.sleep(.05)