fix(scheduler): skip resurrection for idle-sleeping agents (cherry-pick of #446 → main)

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,

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]

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