From da8d65617b0ea46be057a671df0d70a38cef66cc Mon Sep 17 00:00:00 2001 From: olegbrok Date: Mon, 11 May 2026 14:01:40 -0700 Subject: [PATCH] fix(scheduler): skip resurrection eval for idle-sleeping agents (#446) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The session watchdog has been emitting "resurrection attempt 5/5 for dead agent X" every 30s for every idle-sleeping agent on the Pi fleet since PR #348/#349 landed. Those PRs added the idle_sleeping check, but only at the API callback site — by the time it fires, the scheduler has already consumed a rate-limit budget slot and logged the attempt. With 7 agents idle-sleeping most of the time, the journal fills with noise and Brad reports it "feels too eager." Root cause: _maybe_resurrect's only gate was the rate limit. The API callback would refuse the actual reconnect, but the scheduler had no way to know that ahead of time, so it kept burning budget + log slots. Fix: add is_resurrectable_fn precondition to AgentScheduler. Called before budget/log/callback. If it returns False, return silently — no budget consumed, no log emitted, no callback fired. api.py wires in _is_resurrectable which returns False when the streaming session is idle_sleeping (or missing entirely). Fails open on precondition exceptions so a buggy resolver can't silently disable resurrection for the whole fleet. Tests: - precondition False → callback never fires, budget stays at 0 - precondition True → original behavior preserved - precondition raises → falls through (fail-open) Pi was hot-patched in parallel (early-return in _maybe_resurrect) to quiet the noise immediately; this PR replaces that with the proper fix, so the next deploy cleanly overwrites the hot-patch. Co-authored-by: Oleg Co-authored-by: Claude Opus 4.7 --- src/pinky_daemon/api.py | 16 +++++++++ src/pinky_daemon/scheduler.py | 21 +++++++++++ tests/test_scheduler.py | 68 +++++++++++++++++++++++++++++++++++ 3 files changed, 105 insertions(+) diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index c4684843..82e22cfd 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -6309,10 +6309,26 @@ async def _heartbeat_resurrect(agent_name: str, _session_id: str) -> None: except Exception as e: _log(f"api: resurrection failed for {agent_name}: {e}") + def _is_resurrectable(agent_name: str) -> bool: + """Scheduler precondition: True iff resurrection should be attempted. + + Returns False for agents that don't currently want resurrection — most + commonly idle-sleeping sessions (which the API callback would refuse + anyway, but at the cost of a budget slot and a noisy log line every + scheduler tick). See #348/#349 for the original API-layer skip. + """ + ss = broker._get_streaming_session(agent_name) + if not ss: + return False # nothing to resurrect + if getattr(ss, "is_idle_sleeping", False): + return False # deliberately disconnected — leave it alone + return True + scheduler = AgentScheduler( agents, wake_callback=_wake_callback, heartbeat_callback=_heartbeat_resurrect, + is_resurrectable_fn=_is_resurrectable, direct_send_callback=broker.send_callback, dream_callback=_dream_callback, librarian_callback=_librarian_callback, diff --git a/src/pinky_daemon/scheduler.py b/src/pinky_daemon/scheduler.py index e5603fb1..1306ce70 100644 --- a/src/pinky_daemon/scheduler.py +++ b/src/pinky_daemon/scheduler.py @@ -183,6 +183,7 @@ def __init__( dream_callback=None, librarian_callback=None, streaming_sessions_fn=None, + is_resurrectable_fn=None, comms_cleanup_fn=None, trigger_store=None, activity=None, @@ -197,6 +198,11 @@ def __init__( self._librarian_callback = librarian_callback # async fn(agent_name, agent_config) self._last_librarian_check: dict[str, tuple] = {} # dedup key self._streaming_sessions_fn = streaming_sessions_fn # fn() -> dict[name, StreamingSession] + # Precondition check for resurrection. If supplied, must return True iff + # the named agent is currently in a state where resurrection is desired. + # Used to skip eval for idle-sleeping agents (which the API callback + # would refuse anyway, but at the cost of a budget slot and a log line). + self._is_resurrectable_fn = is_resurrectable_fn # fn(agent_name) -> bool self._comms_cleanup_fn = comms_cleanup_fn # fn() -> int (expired inbox cleanup) self._trigger_store = trigger_store # TriggerStore | None self._activity = activity # ActivityStore | None @@ -391,6 +397,21 @@ async def _maybe_resurrect( if not self._heartbeat_callback: return + # Precondition: skip agents that don't want resurrection at all + # (e.g. idle-sleeping). This avoids consuming the rate-limit budget + # and emitting "attempt N/N" log spam every tick for sleeping agents + # the API callback would refuse anyway. See #348/#349 — that fix + # landed at the API layer; this is the matching scheduler-level skip. + if self._is_resurrectable_fn is not None: + try: + if not self._is_resurrectable_fn(agent_name): + return + except Exception as e: + # Fail-open: if the precondition check itself errors, fall + # through to the existing path so we don't silently disable + # resurrection for everyone. + _log(f"scheduler: is_resurrectable_fn raised for {agent_name}: {e}") + # Trim attempts outside the window window_start = now - self.RESURRECTION_WINDOW_SECONDS attempts = [t for t in self._resurrection_attempts.get(agent_name, []) if t >= window_start] diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 011be662..346f900e 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -495,6 +495,74 @@ async def cb(agent_name, session_id): await scheduler._maybe_resurrect("ivan", "sid", future) assert len(called) == scheduler.RESURRECTION_MAX_ATTEMPTS + 1 + @pytest.mark.asyncio + async def test_resurrection_skipped_when_precondition_false(self, registry): + """is_resurrectable_fn returning False short-circuits before budget/log. + + Regression for the "watchdog spams 5/5 every 30s on idle-sleeping + agents" bug: the API callback used to be the only place idle-sleep + was checked, so the scheduler still consumed a budget slot and + emitted a log line for each tick. With is_resurrectable_fn, the + scheduler skips entirely. + """ + called = [] + + async def cb(agent_name, session_id): + called.append(agent_name) + + scheduler = AgentScheduler( + registry, + heartbeat_callback=cb, + is_resurrectable_fn=lambda name: False, # never resurrectable + ) + now = time.time() + for _ in range(scheduler.RESURRECTION_MAX_ATTEMPTS + 3): + await scheduler._maybe_resurrect("ivan", "sid", now) + + # Callback never fired + assert called == [] + # And budget was never consumed — the attempt list stays empty so a + # later state-change (agent stops idle-sleeping) gets the full quota. + assert scheduler._resurrection_attempts.get("ivan", []) == [] + + @pytest.mark.asyncio + async def test_resurrection_runs_when_precondition_true(self, registry): + """is_resurrectable_fn returning True preserves old behavior.""" + called = [] + + async def cb(agent_name, session_id): + called.append(agent_name) + + scheduler = AgentScheduler( + registry, + heartbeat_callback=cb, + is_resurrectable_fn=lambda name: True, + ) + now = time.time() + await scheduler._maybe_resurrect("ivan", "sid", now) + assert called == ["ivan"] + + @pytest.mark.asyncio + async def test_resurrection_precondition_failure_fails_open(self, registry): + """If is_resurrectable_fn raises, fall through (don't silently disable).""" + called = [] + + async def cb(agent_name, session_id): + called.append(agent_name) + + def bad_precondition(name): + raise RuntimeError("oops") + + scheduler = AgentScheduler( + registry, + heartbeat_callback=cb, + is_resurrectable_fn=bad_precondition, + ) + now = time.time() + await scheduler._maybe_resurrect("ivan", "sid", now) + # Fail-open: resurrection proceeds despite precondition exception + assert called == ["ivan"] + @pytest.mark.asyncio async def test_no_callback_means_no_crash(self, registry): """Dead agents are still legal even when no resurrection wiring exists."""