fix(agent): one blocked periodic callback no longer stalls lease refresh and liveness timers (#102574; salvage #105716, same mechanism as #102458)

agent/periodic_scheduler.py

@@ -3,15 +3,20 @@
 Replaces the per-child ``while not stop.wait(interval): body()`` daemon
 threads (delegate heartbeat, durable turn-lease refresher, turn-liveness
 watchdog).  With ~130 in-process subagents those added 2-3 sleeping OS
-threads per child; this module runs every periodic body on ONE daemon
-thread ordered by a heap of due times.
+threads per child.  This module keeps ONE daemon thread that only orders due
+times; every due body runs on its own short-lived daemon worker.  A blocked
+callback therefore cannot delay unrelated lease/liveness timers, while
+steady-state thread use stays near zero (workers exist only while a body is
+actually running, never one per scheduled handle).
 
 Semantics match the loop they replace: the first call happens ``interval``
 seconds after :func:`schedule`, and each following call ``interval`` seconds
 after the previous body *returned* (drift-free wrt. body duration was never
 a property of the old loops either).  A body that returns ``False`` stops
 itself; a body that raises is logged at debug and rescheduled — one bad
-callback must never kill the shared thread.
+callback must never kill the shared thread.  A handle never overlaps itself:
+it is re-queued only once its in-flight run has returned.  A worker-start
+failure never retires the handle: it is re-queued and logged at warning.
 """
 
 from __future__ import annotations
@@ -26,18 +31,20 @@
 logger = logging.getLogger(__name__)
 
 _THREAD_NAME = "hermes-periodic-scheduler"
+_CALLBACK_THREAD_PREFIX = "hermes-periodic-callback"
 
 
 class ScheduledHandle:
     """Cancel token for one scheduled periodic callback."""
 
-    __slots__ = ("_fn", "_interval", "_cancelled", "_scheduler")
+    __slots__ = ("_fn", "_interval", "_cancelled", "_scheduler", "_runner")
 
     def __init__(self, scheduler: "PeriodicScheduler", fn: Callable[[], object], interval: float):
         self._scheduler = scheduler
         self._fn = fn
         self._interval = interval
         self._cancelled = False
+        self._runner: Optional[threading.Thread] = None
 
     @property
     def cancelled(self) -> bool:
@@ -56,12 +63,11 @@ def __init__(self) -> None:
         self._heap: list = []  # (due, seq, handle)
         self._seq = itertools.count()
         self._thread: Optional[threading.Thread] = None
-        self._running: Optional[ScheduledHandle] = None
 
     def schedule(self, fn: Callable[[], object], interval: float) -> ScheduledHandle:
         handle = ScheduledHandle(self, fn, float(interval))
         with self._cond:
-            heapq.heappush(self._heap, (time.monotonic() + handle._interval, next(self._seq), handle))
+            self._requeue(handle)
             if self._thread is None or not self._thread.is_alive():
                 self._thread = threading.Thread(target=self._run, name=_THREAD_NAME, daemon=True)
                 self._thread.start()
@@ -72,8 +78,52 @@ def _cancel(self, handle: ScheduledHandle, wait: Optional[float]) -> None:
         with self._cond:
             handle._cancelled = True
             self._cond.notify()
-            if wait and self._running is handle and threading.current_thread() is not self._thread:
-                self._cond.wait_for(lambda: self._running is not handle, timeout=wait)
+            runner = handle._runner
+        if wait and runner is not None and threading.current_thread() is not runner:
+            runner.join(wait)
+
+    def _dispatch(self, handle: ScheduledHandle) -> None:
+        """Start ``handle``'s body on its own worker.  Called with ``_cond`` held
+        so ``cancel`` can never observe a half-set runner."""
+        runner = threading.Thread(
+            target=self._run_callback,
+            args=(handle,),
+            name=f"{_CALLBACK_THREAD_PREFIX}-{id(handle):x}",
+            daemon=True,
+        )
+        handle._runner = runner
+        try:
+            runner.start()
+        except Exception:
+            handle._runner = None
+            logger.warning(
+                "failed to start periodic callback worker %r; retrying in %s s",
+                handle._fn,
+                handle._interval,
+                exc_info=True,
+            )
+            if not handle._cancelled:
+                self._requeue(handle)
+                self._cond.notify()
+
+    def _requeue(self, handle: ScheduledHandle) -> None:
+        """Push ``handle``'s next due time (``_cond`` held)."""
+        heapq.heappush(self._heap, (time.monotonic() + handle._interval, next(self._seq), handle))
+
+    def _run_callback(self, handle: ScheduledHandle) -> None:
+        stop = False
+        try:
+            stop = handle._fn() is False
+        except Exception:
+            logger.debug("periodic callback %r raised", handle._fn, exc_info=True)
+        finally:
+            with self._cond:
+                handle._runner = None
+                if stop:
+                    handle._cancelled = True
+                elif not handle._cancelled:
+                    self._requeue(handle)
+                self._cond.notify()
 
     def _run(self) -> None:
         while True:
@@ -91,28 +141,13 @@ def _run(self) -> None:
                         self._cond.wait(delay)
                         continue
                     heapq.heappop(self._heap)
-                    self._running = handle
+                    self._dispatch(handle)
                     break
-            stop = False
-            try:
-                stop = handle._fn() is False
-            except Exception:
-                logger.debug("periodic callback %r raised", handle._fn, exc_info=True)
-            with self._cond:
-                self._running = None
-                if stop:
-                    handle._cancelled = True
-                elif not handle._cancelled:
-                    heapq.heappush(
-                        self._heap,
-                        (time.monotonic() + handle._interval, next(self._seq), handle),
-                    )
-                self._cond.notify_all()
 
 
 _DEFAULT = PeriodicScheduler()
 
 
 def schedule(fn: Callable[[], object], interval: float) -> ScheduledHandle:
-    """Run ``fn()`` every ``interval`` seconds on the shared scheduler thread."""
+    """Run ``fn()`` every ``interval`` seconds via the shared scheduler."""
     return _DEFAULT.schedule(fn, interval)

agent/turn_facade_lease.py

@@ -4,7 +4,8 @@
 resume, gateway, background delivery). ``admit_durable_turn_lease`` acquires the row lease (or
 returns the early result the façade must hand back); ``DurableTurnLease`` owns the periodic
 refresher, the turn-liveness watchdog wiring, and the lease-loss / stall interrupt plumbing. Both
-timers run on the shared scheduler thread (``agent/periodic_scheduler.py``), not per-turn threads.
+timers run via the shared scheduler (``agent/periodic_scheduler.py``; timer thread orders,
+bodies run on per-handle workers), not per-turn threads.
 """
 import logging
 import os
@@ -172,7 +173,7 @@ def clear_interrupt(self) -> None:
                 _set_interrupt(False, agent._execution_thread_id)
 
     def refresh_tick(self):
-        """One periodic renewal (every ``refresh_interval`` on the shared scheduler); a miss or
+        """One periodic renewal (every ``refresh_interval`` via the shared scheduler); a miss or
         error interrupts the turn. Returning False stops the timer.
 
         The holder-qualified UPDATE fences a late refresher from a successor lease. The façade's

agent/turn_liveness.py

@@ -86,8 +86,8 @@ def resolve_turn_liveness_settings(
 
 
 class TurnLivenessWatchdog:
-    """Sampled-idle watchdog bound to one conversation turn (polls on the
-    shared periodic scheduler thread).
+    """Sampled-idle watchdog bound to one conversation turn (via the shared
+    periodic scheduler; timer thread orders, body runs on its own worker).
 
     ``activity_lock`` must be the SAME lock ``AIAgent._touch_activity`` stamps
     the activity clock with; run_agent owns the lease state and callbacks.
@@ -110,7 +110,7 @@ def __init__(
         self._deactivate_turn = deactivate_turn
 
     def schedule(self):
-        """Start polling on the shared periodic scheduler thread; returns the cancel handle.
+        """Start polling via the shared periodic scheduler; returns the cancel handle.
         Scheduled at turn entry, after the turn-active flag and activity clock are stamped."""
         from agent.periodic_scheduler import schedule
 

tests/agent/test_periodic_scheduler.py

@@ -1,4 +1,4 @@
-"""agent/periodic_scheduler: one shared thread runs every periodic timer."""
+"""agent/periodic_scheduler: one timer thread dispatches isolated callbacks."""
 
 import threading
 import time
@@ -24,7 +24,7 @@ def test_two_intervals_fire_proportionally_and_cancel_stops_one():
 
     assert _wait_until(lambda: len(slow) >= 3)
     assert len(fast) > len(slow)  # 5x interval ratio -> clearly more fast ticks
-    # Both ran on this scheduler's single thread, not on new threads.
+    # Scheduling another timer creates no persistent per-handle thread.
     before = threading.active_count()
     sched.schedule(lambda: None, 0.01).cancel()
     assert threading.active_count() == before
@@ -90,3 +90,64 @@ def test_module_level_schedule_uses_shared_default():
     assert threading.active_count() == before
     for handle in handles:
         handle.cancel()
+
+
+def test_blocked_callback_does_not_stall_due_sibling(monkeypatch):
+    scheduler = PeriodicScheduler()
+    monkeypatch.setattr(periodic_scheduler, "_DEFAULT", scheduler)
+    blocker_entered = threading.Event()
+    release_blocker = threading.Event()
+    sibling_ran = threading.Event()
+
+    def blocker():
+        blocker_entered.set()
+        release_blocker.wait(5.0)
+        return False
+
+    def sibling():
+        sibling_ran.set()
+        return False
+
+    blocker_handle = schedule(blocker, 0.01)
+    assert blocker_entered.wait(2.0)
+    sibling_handle = schedule(sibling, 0.01)
+    try:
+        # Ordering, not a wall-clock bound: the sibling must fire WHILE the blocker still holds
+        # its worker. On main the sibling only runs after the blocker's 5 s wait expires.
+        assert sibling_ran.wait(2.0) and not release_blocker.is_set(), (
+            "a blocked periodic callback stalled an unrelated due callback"
+        )
+    finally:
+        release_blocker.set()
+        blocker_handle.cancel(wait=1.0)
+        sibling_handle.cancel(wait=1.0)
+
+
+def test_worker_start_failure_keeps_timer(monkeypatch):
+    sched = PeriodicScheduler()
+    fired: list = []
+    real_thread = threading.Thread
+    attempts = {"n": 0}
+
+    def flaky(*args, **kwargs):
+        # Only this scheduler's own callback worker fails, once; a leaked handle on the shared
+        # _DEFAULT scheduler must not be the one that consumes the single Boom.
+        if kwargs.get("target") is sched._run_callback and attempts["n"] == 0:
+            attempts["n"] += 1
+
+            class Boom:
+                def start(self):
+                    raise RuntimeError("no threads")
+
+            return Boom()
+        return real_thread(*args, **kwargs)
+
+    monkeypatch.setattr(periodic_scheduler.threading, "Thread", flaky)
+    handle = sched.schedule(lambda: fired.append(1), 0.01)
+    try:
+        assert _wait_until(lambda: bool(fired), timeout=3.0), (
+            "worker-start failure silently retired the timer"
+        )
+        assert not handle.cancelled
+    finally:
+        handle.cancel()

tools/delegate_tool_child_run.py

@@ -203,7 +203,7 @@ def _attr_line(attr):
 # ── Per-run helpers ──────────────────────────────────────────────────────────
 
 class _Heartbeat:
-    """One child's parent-activity heartbeat on the shared periodic scheduler thread
+    """One child's parent-activity heartbeat via the shared periodic scheduler
     (``agent.periodic_scheduler``) — not one daemon thread per child. NOT started at construction:
     the caller calls ``start()`` inside its ``try`` so a failed schedule (OS thread exhaustion on
     first use) leaves ``handle`` None and ``stop()`` is a no-op."""