From 305aa0c1bf20630974207b429ec64f0689d21ee9 Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 14:58:04 -0700 Subject: [PATCH 1/2] feat(#663): activate MCP-recover for low-cadence agents (R1 enable-gate + R2 checkable) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two refinements that let `mcp_recover` be enabled on an agent whose progress/transition watchdog is OFF (barsik's shape) — the activation blockers found in the #664 post-merge review. R1 — watchdog enable-gate (session_watchdog._evaluate): `mcp_recover` is its own opt-in, independent of the master `cfg.enabled` switch. The MCP-bind branch now runs before the `enabled` gate (return early only when BOTH `enabled` and `mcp_recover` are off); the progress/transition watchdog stays gated on `enabled`. During a BOOTING/RECONNECTING transition the branch is a cheap no-op (connected is False), so the transition branch still handles that wedge. R2 — checkable signal (api._watchdog_mcp_bind_status): `checkable` was `heartbeat_interval > 0`, which wrongly excludes wake-driven sidekicks that carry interval==0 yet heartbeat in practice (barsik; Murzik Finding #2). Now also true when the agent emitted a recent AGENT-ORIGIN heartbeat (get_latest_agent_heartbeat — excludes synthetic server_presence rows, so a wedged agent can't look checkable-and-bound). Extracted as the pure, unit-tested helper `compute_mcp_checkable`. Still flag-gated (mcp_recover default OFF). Tests: R1 enable-gate routing + progress-gated-off, R2 checkable policy matrix. 52 pass. Co-Authored-By: Claude Opus 4.8 --- src/pinky_daemon/api.py | 34 ++++++-- src/pinky_daemon/session_watchdog.py | 79 +++++++++++++++-- tests/test_session_watchdog.py | 122 +++++++++++++++++++++++++++ 3 files changed, 220 insertions(+), 15 deletions(-) diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 848043ae..f50d96b3 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -138,7 +138,11 @@ from pinky_daemon.research_store import ResearchStore from pinky_daemon.scheduler import AgentScheduler from pinky_daemon.session_store import SessionEventStore, SessionStore -from pinky_daemon.session_watchdog import SessionWatchdog, WatchdogConfig +from pinky_daemon.session_watchdog import ( + SessionWatchdog, + WatchdogConfig, + compute_mcp_checkable, +) from pinky_daemon.sessions import SessionManager, SessionState from pinky_daemon.shared_mcp import ( SHARED_MCP_HOST, @@ -8300,18 +8304,34 @@ async def _watchdog_alert(agent_name: str, message: str) -> None: def _watchdog_mcp_bind_status(agent_name: str) -> dict: """Bind-status for the watchdog's #663 MCP-recover check. - ``checkable`` requires the agent to be heartbeat-enabled (so a missing - current-epoch MCP success is a real wedge signal, not just a quiet - agent) AND a gateway generation to be established. ``bound`` is true - when a successful MCP round-trip has been recorded for the CURRENT - gateway epoch (the heartbeat bind signal). + ``checkable`` means a missing current-epoch MCP success is a real wedge + signal (not just a quiet / heartbeat-disabled agent) AND a gateway + generation is established. It is true when the agent heartbeats on a + cadence (``heartbeat_interval > 0``) OR has emitted a recent AGENT-ORIGIN + heartbeat — the latter covers wake-driven sidekicks (e.g. barsik) that + carry ``heartbeat_interval == 0`` but heartbeat in practice (#663 Murzik + Finding #2 / R2). ``bound`` is true when a successful MCP round-trip has + been recorded for the CURRENT gateway epoch (the heartbeat bind signal). """ agent = agents.get(agent_name) hb = int(getattr(agent, "heartbeat_interval", 0) or 0) if agent else 0 epoch = get_gateway_epoch() st = get_probe_status(agent_name) + # Agent-origin heartbeat recency — NOT get_latest_heartbeat, so synthetic + # server_presence rows can't make a wedged agent look checkable-and-bound. + latest_hb_ts: float | None = None + if agent: + latest_hb = agents.get_latest_agent_heartbeat(agent_name) + if latest_hb and latest_hb.timestamp: + latest_hb_ts = float(latest_hb.timestamp) return { - "checkable": bool(agent and hb > 0 and epoch), + "checkable": compute_mcp_checkable( + agent_exists=bool(agent), + epoch=epoch, + heartbeat_interval=hb, + latest_agent_heartbeat_ts=latest_hb_ts, + now=time.time(), + ), "bound": bool(st.get("bound")), "heartbeat_interval": hb, } diff --git a/src/pinky_daemon/session_watchdog.py b/src/pinky_daemon/session_watchdog.py index 45161255..b3e30729 100644 --- a/src/pinky_daemon/session_watchdog.py +++ b/src/pinky_daemon/session_watchdog.py @@ -64,6 +64,54 @@ DEFAULT_MCP_UNBOUND_FLOOR = 240 # min deadline regardless of heartbeat interval DEFAULT_MCP_UNBOUND_HEARTBEAT_MULT = 3 # deadline >= this * heartbeat_interval DEFAULT_MCP_RECOVER_MIN_INTERVAL = 120 # global min secs between any two MCP recoveries +# checkable-via-history window (#663 / R2). An agent with heartbeat_interval==0 +# (no scheduler-driven cadence) is still "checkable" if it emitted an +# AGENT-ORIGIN heartbeat within this window — i.e. it is an actively-heartbeating +# agent, so a sustained current-epoch unbound is a real wedge signal. Generous +# enough (>> the unbound deadline) that a freshly-orphaned low-cadence agent +# stays checkable through detection, but bounded so a long-quiet agent stops +# being a recovery target (and can't be falsely force-restarted). +DEFAULT_MCP_CHECKABLE_HEARTBEAT_MAX_AGE = 1800 # 30 min + + +def compute_mcp_checkable( + *, + agent_exists: bool, + epoch: str, + heartbeat_interval: int, + latest_agent_heartbeat_ts: float | None, + now: float, + max_heartbeat_age: float = DEFAULT_MCP_CHECKABLE_HEARTBEAT_MAX_AGE, +) -> bool: + """Decide whether an agent's MCP bind status is *checkable* (#663 / R2). + + "Checkable" means: this agent is one we expect to be making MCP round-trips, + so a missing current-epoch bind is a genuine wedge signal — not just a quiet + or heartbeat-disabled agent. True iff a gateway epoch is established AND + EITHER: + + * ``heartbeat_interval > 0`` — the agent is configured to heartbeat on a + cadence (the classic scheduler-monitored agent); or + * the agent emitted an AGENT-ORIGIN heartbeat within ``max_heartbeat_age`` + — covers active agents that carry ``heartbeat_interval == 0`` (wake-driven + sidekicks like barsik) yet heartbeat in practice. The timestamp MUST come + from ``get_latest_agent_heartbeat`` (NOT ``get_latest_heartbeat``) so the + scheduler's synthetic ``server_presence`` rows can't make a wedged agent + look checkable-and-healthy — #663 Murzik Finding #2. + + A pure function (no I/O) so the policy is unit-testable in isolation; the API + layer threads in the live ``agents``/ledger reads. + """ + if not (agent_exists and epoch): + return False + if heartbeat_interval > 0: + return True + if ( + latest_agent_heartbeat_ts is not None + and (now - latest_agent_heartbeat_ts) <= max_heartbeat_age + ): + return True + return False @dataclass @@ -289,7 +337,12 @@ def _take_snapshot( async def _evaluate(self, snap: _SessionSnapshot, now: float) -> None: cfg = self._config_fn(snap.agent_name) - if not cfg.enabled: + # The MCP-bind recovery branch (#663) is gated on its OWN flag + # (cfg.mcp_recover), independent of the master cfg.enabled switch: an + # agent can run with the progress/transition watchdog disabled yet still + # opt into MCP-orphan auto-recovery. So return early only when BOTH are + # off; everything below the MCP-bind branch stays gated on cfg.enabled. + if not cfg.enabled and not cfg.mcp_recover: return state_key = (snap.agent_name, snap.label) @@ -299,6 +352,23 @@ async def _evaluate(self, snap: _SessionSnapshot, now: float) -> None: last_progress_at=now, )) + # ── MCP-bind recovery branch (#663) ────────────────────────── + # Self-gated on cfg.mcp_recover (a no-op when off), and evaluated BEFORE + # the cfg.enabled gate so it works for agents whose progress/transition + # watchdog is disabled. Also independent of the progress logic below: a + # wedged-MCP session can still "make progress" on non-MCP turns, so + # progress must not mask a dead transport. Returns True (and we stop) on + # recovery. During a BOOTING/RECONNECTING transition snap.connected is + # False, so this is a cheap no-op (clears the unbound clock) and the + # transition branch below still handles the wedge. + if await self._evaluate_mcp_bind(snap, state, cfg, now): + return + + # Everything below — the lifecycle-transition and progress/backlog + # watchdog — is gated on the master enable flag. + if not cfg.enabled: + return + # ── Lifecycle transition-age branch (#109) ────────────────── # Handled BEFORE the progress/backlog logic: a session wedged in # BOOTING/RECONNECTING reports connected=False and would be dropped @@ -315,13 +385,6 @@ async def _evaluate(self, snap: _SessionSnapshot, now: float) -> None: state.transition_warned = False state.transition_recovered_at = 0.0 - # ── MCP-bind recovery branch (#663) ────────────────────────── - # Runs independent of the progress/backlog logic below: a wedged-MCP - # session can still "make progress" on non-MCP turns, so progress must - # not mask a dead transport. Returns True (and we stop) on recovery. - if await self._evaluate_mcp_bind(snap, state, cfg, now): - return - # Detect progress: turn count increased or activity changed made_progress = ( snap.turns > state.last_progress_turns diff --git a/tests/test_session_watchdog.py b/tests/test_session_watchdog.py index 6c313994..124ad502 100644 --- a/tests/test_session_watchdog.py +++ b/tests/test_session_watchdog.py @@ -6,11 +6,13 @@ import pytest from pinky_daemon.session_watchdog import ( + DEFAULT_MCP_CHECKABLE_HEARTBEAT_MAX_AGE, DEFAULT_MCP_UNBOUND_FLOOR, SessionWatchdog, WatchdogConfig, _AgentState, _SessionSnapshot, + compute_mcp_checkable, ) @@ -747,3 +749,123 @@ async def rec(a, _l, _r): assert recovered == [] await wd._evaluate(snap, now + DEFAULT_MCP_UNBOUND_FLOOR + 5) # fires assert recovered == ["a"] + + @pytest.mark.asyncio + async def test_evaluate_mcp_branch_runs_when_watchdog_disabled(self, make_watchdog): + """R1 (#663): mcp_recover is its own opt-in — the MCP-bind branch fires + even when the master watchdog (cfg.enabled) is OFF.""" + recovered = [] + + async def rec(a, _l, _r): + recovered.append(a) + + wd = make_watchdog( + mcp_bind_status_fn=lambda n: { + "checkable": True, "bound": False, "heartbeat_interval": 0}, + mcp_recover_fn=rec, + # The exact barsik shape: progress watchdog disabled, mcp_recover on. + agent_config_fn=lambda n: WatchdogConfig(enabled=False, mcp_recover=True), + ) + snap = _conn_snap() + now = 3000.0 + await wd._evaluate(snap, now) # arms the unbound clock + assert recovered == [] + await wd._evaluate(snap, now + DEFAULT_MCP_UNBOUND_FLOOR + 5) # fires + assert recovered == ["a"] + + @pytest.mark.asyncio + async def test_evaluate_disabled_without_mcp_recover_is_noop(self, make_watchdog): + """A fully-disabled agent (enabled=False, mcp_recover=False) returns + before any work — no MCP-bind lookup, no tracked state entry.""" + calls = [] + wd = make_watchdog( + mcp_bind_status_fn=lambda n: calls.append(n) or { + "checkable": True, "bound": False, "heartbeat_interval": 0}, + mcp_recover_fn=lambda *a: None, + agent_config_fn=lambda n: WatchdogConfig(enabled=False, mcp_recover=False), + ) + await wd._evaluate(_conn_snap(), 1000.0) + assert calls == [] # status fn never consulted + assert ("a", "main") not in wd._states # no state created for a no-op agent + + @pytest.mark.asyncio + async def test_disabled_watchdog_skips_progress_logic(self, make_watchdog): + """With enabled=False + mcp_recover=True, the progress/backlog watchdog + stays gated OFF — a maximally-stuck session never warns or recovers via + the generic path; only the MCP-bind branch is live.""" + recovered, alerts = [], [] + + async def rec(a, _l, _r): + recovered.append(a) + + async def alert(a, _m): + alerts.append(a) + + wd = make_watchdog( + recover_fn=rec, + alert_fn=alert, + # bound=True → MCP branch is a no-op, isolating the progress gate. + mcp_bind_status_fn=lambda n: { + "checkable": True, "bound": True, "heartbeat_interval": 0}, + mcp_recover_fn=lambda *a: None, + agent_config_fn=lambda n: WatchdogConfig(enabled=False, mcp_recover=True), + ) + # A connected session with backlog that makes no progress across sweeps. + stuck = _SessionSnapshot( + agent_name="a", label="main", connected=True, turns=0, pending=3, + current_activity="thinking", sample_time=time.time(), state="connected", + ) + now = 1000.0 + await wd._evaluate(stuck, now) # seed state + # Jump far past warn_after (600) and recover_after (900). + await wd._evaluate(stuck, now + 10_000) + assert recovered == [] # progress-recover gated off + assert alerts == [] # progress-warn gated off + + +class TestComputeMcpCheckable: + """R2 (#663): checkable policy — recent agent-origin heartbeat covers + heartbeat_interval==0 sidekicks (Murzik Finding #2).""" + + NOW = 10_000.0 + + def test_no_agent_never_checkable(self): + assert compute_mcp_checkable( + agent_exists=False, epoch="e1", heartbeat_interval=60, + latest_agent_heartbeat_ts=self.NOW, now=self.NOW) is False + + def test_no_epoch_never_checkable(self): + # No gateway generation established yet — nothing to bind against. + assert compute_mcp_checkable( + agent_exists=True, epoch="", heartbeat_interval=60, + latest_agent_heartbeat_ts=self.NOW, now=self.NOW) is False + + def test_cadence_agent_checkable_regardless_of_history(self): + # heartbeat_interval>0 is sufficient — even with no heartbeat history. + assert compute_mcp_checkable( + agent_exists=True, epoch="e1", heartbeat_interval=60, + latest_agent_heartbeat_ts=None, now=self.NOW) is True + + def test_zero_interval_recent_heartbeat_checkable(self): + # barsik's shape: heartbeat_interval==0 but a recent agent-origin beat. + assert compute_mcp_checkable( + agent_exists=True, epoch="e1", heartbeat_interval=0, + latest_agent_heartbeat_ts=self.NOW - 60, now=self.NOW) is True + + def test_zero_interval_stale_heartbeat_not_checkable(self): + assert compute_mcp_checkable( + agent_exists=True, epoch="e1", heartbeat_interval=0, + latest_agent_heartbeat_ts=self.NOW - DEFAULT_MCP_CHECKABLE_HEARTBEAT_MAX_AGE - 1, + now=self.NOW) is False + + def test_zero_interval_no_history_not_checkable(self): + assert compute_mcp_checkable( + agent_exists=True, epoch="e1", heartbeat_interval=0, + latest_agent_heartbeat_ts=None, now=self.NOW) is False + + def test_window_boundary_inclusive(self): + # Exactly at the window edge counts as fresh (<=). + assert compute_mcp_checkable( + agent_exists=True, epoch="e1", heartbeat_interval=0, + latest_agent_heartbeat_ts=self.NOW - DEFAULT_MCP_CHECKABLE_HEARTBEAT_MAX_AGE, + now=self.NOW) is True From dbc57ee6d2d7faafd8d945816b78525dc49594b0 Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 15:05:01 -0700 Subject: [PATCH 2/2] test(#663): add transition-snap + mcp_recover guard test (Murzik nit) Proves the R1 _evaluate reorder doesn't steal lifecycle transitions: with enabled=True + mcp_recover=True and a RECONNECTING (connected=False) snap, the MCP-bind branch no-ops and the transition branch still owns the wedge. Co-Authored-By: Claude Opus 4.8 --- tests/test_session_watchdog.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/test_session_watchdog.py b/tests/test_session_watchdog.py index 124ad502..d9385c3f 100644 --- a/tests/test_session_watchdog.py +++ b/tests/test_session_watchdog.py @@ -14,6 +14,7 @@ _SessionSnapshot, compute_mcp_checkable, ) +from pinky_daemon.transport_state import SessionState class FakeSession: @@ -822,6 +823,37 @@ async def alert(a, _m): assert recovered == [] # progress-recover gated off assert alerts == [] # progress-warn gated off + @pytest.mark.asyncio + async def test_transition_snap_with_mcp_recover_defers_to_transition_branch( + self, make_watchdog + ): + """With enabled=True + mcp_recover=True and a session in a transition + (BOOTING/RECONNECTING → connected=False), the MCP-bind branch is a no-op + (it owns CONNECTED sessions) and the lifecycle-transition branch still + takes the wedge — proves the R1 reorder didn't steal transitions.""" + recovered = [] + + async def rec(a, _l, _r): + recovered.append(a) + + wd = make_watchdog( + mcp_bind_status_fn=lambda n: { + "checkable": True, "bound": False, "heartbeat_interval": 60}, + mcp_recover_fn=rec, + agent_config_fn=lambda n: WatchdogConfig(enabled=True, mcp_recover=True), + ) + transitioning = _SessionSnapshot( + agent_name="a", label="main", connected=False, turns=0, pending=0, + current_activity="", sample_time=time.time(), + state=SessionState.RECONNECTING.value, + ) + await wd._evaluate(transitioning, 1000.0) + assert recovered == [] # MCP-bind branch did not fire on a non-connected snap + # Transition branch ran: it seeded its sampled timer for this state. + st = wd._states[("a", "main")] + assert st.transition_state == SessionState.RECONNECTING.value + assert st.mcp_unbound_since == 0.0 # branch cleared the unbound clock + class TestComputeMcpCheckable: """R2 (#663): checkable policy — recent agent-origin heartbeat covers