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."""