From 059a2e180fd7911810ad1ffb6da0143508367f32 Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 13:35:46 -0700 Subject: [PATCH 1/5] feat(#663): MCP bind ledger + gateway epoch + mcp_probe tool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Primitive for detecting an identity-unbound resumed session (CC #60949/#9608: CC won't re-init an MCP transport after a 404, so a daemon restart that tears down the :8890 gateway leaves live agents' tools dead until a fresh relaunch). - shared_mcp: per-gateway-generation epoch (bumps on every app build) + an in-memory per-agent bind ledger written only from inside the tool handler (a tool executing proves the agent's real MCP client traversed the current gateway; request arrival does not, since a stale client can still 404). Epoch bump clears the ledger so a pre-restart bind never counts as current. - pinky-self: mcp_probe(nonce, launch_id) core tool — records a probe success and returns {agent, nonce, launch_id, gateway_epoch, observed_at}. Co-Authored-By: Claude Opus 4.8 --- src/pinky_daemon/shared_mcp.py | 88 ++++++++++++++++++++++++++++++++++ src/pinky_self/server.py | 30 +++++++++++- 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/src/pinky_daemon/shared_mcp.py b/src/pinky_daemon/shared_mcp.py index c15dbe01..0bc3b9c0 100644 --- a/src/pinky_daemon/shared_mcp.py +++ b/src/pinky_daemon/shared_mcp.py @@ -16,6 +16,7 @@ import sys import threading import time +import uuid from collections.abc import Callable from contextvars import ContextVar @@ -136,6 +137,88 @@ def __radd__(self, other: str) -> str: return other + self._resolve() +# ── MCP bind ledger (issue #663) ────────────────────────────── +# +# Tracks, per agent, the last successful round-trip through THIS gateway +# generation — used to detect a resumed CC session whose MCP transport came +# up unbound. Upstream CC bug (anthropics/claude-code #60949, #9608, both +# closed-as-not-planned): CC does not re-initialize an HTTP/SSE MCP transport +# after a 404 "session not found", so a daemon restart that tears down this +# gateway leaves live agents' tools dead until a fresh relaunch. +# +# A tool *executing* is positive proof the agent's real MCP client traversed +# the current gateway, so the ledger is written from inside the tool handler +# (record_probe_success) — never on mere request arrival, since a stale client +# can reach the gateway and still get a 404. +# +# gateway_epoch is reassigned every time the gateway app is (re)built — on a +# daemon restart AND on a gateway crash-restart — so a recorded probe can be +# matched to the gateway generation that actually served it. + +_gateway_epoch: str = "" +_probe_ledger: dict[str, dict] = {} +_ledger_lock = threading.Lock() + + +def bump_gateway_epoch() -> str: + """Assign a fresh gateway epoch (called once per gateway app build).""" + global _gateway_epoch + _gateway_epoch = uuid.uuid4().hex[:12] + # A new generation invalidates every prior bind — clear the ledger so a + # stale pre-restart entry can never be mistaken for a current-gen success. + with _ledger_lock: + _probe_ledger.clear() + return _gateway_epoch + + +def get_gateway_epoch() -> str: + """Current gateway generation id (empty until the app is first built).""" + return _gateway_epoch + + +def record_probe_success(agent_name: str, nonce: str, launch_id: str = "") -> dict: + """Record a successful mcp_probe round-trip for an agent. + + Called from inside the mcp_probe tool handler — i.e. only after the agent's + MCP client successfully reached this gateway generation. Returns the + recorded entry (with the live gateway_epoch and observed_at). + """ + if not agent_name: + return {} + entry = { + "nonce": nonce, + "launch_id": launch_id, + "gateway_epoch": _gateway_epoch, + "observed_at": time.time(), + } + with _ledger_lock: + _probe_ledger[agent_name] = entry + return dict(entry) + + +def get_probe_status(agent_name: str) -> dict: + """Return the last-success ledger entry for an agent plus the current epoch. + + `bound` (a success exists for the CURRENT gateway generation) is left for + the caller to compute as ``current_epoch and current_epoch == bound_epoch``. + """ + with _ledger_lock: + entry = dict(_probe_ledger.get(agent_name, {})) + observed = entry.get("observed_at") + current = _gateway_epoch + bound_epoch = entry.get("gateway_epoch", "") + return { + "agent": agent_name, + "current_epoch": current, + "bound_epoch": bound_epoch, + "bound": bool(current and current == bound_epoch), + "nonce": entry.get("nonce", ""), + "launch_id": entry.get("launch_id", ""), + "observed_at": observed, + "age_sec": (time.time() - observed) if observed else None, + } + + # ── ASGI Middleware ─────────────────────────────────────────── class AgentNameMiddleware: @@ -182,6 +265,11 @@ def create_shared_app( from starlette.applications import Starlette from starlette.routing import Mount + # New gateway generation: bump the epoch (and clear the bind ledger) so a + # resumed agent's pre-restart MCP binding is never counted as current. (#663) + epoch = bump_gateway_epoch() + _log(f"[shared-mcp] Gateway epoch: {epoch}") + routes = [] session_managers = [] # Track streamable HTTP session managers for lifespan diff --git a/src/pinky_self/server.py b/src/pinky_self/server.py index 814977b7..20e7c6d6 100644 --- a/src/pinky_self/server.py +++ b/src/pinky_self/server.py @@ -27,6 +27,7 @@ import json import os import sys +import time import urllib.error import urllib.parse import urllib.request @@ -36,7 +37,7 @@ from mcp.server.fastmcp import FastMCP from pinky_daemon.auth import build_internal_auth_headers, resolve_request_signing_secret -from pinky_daemon.shared_mcp import LazyAgentName, resolve_lazy +from pinky_daemon.shared_mcp import LazyAgentName, record_probe_success, resolve_lazy def _log(msg: str) -> None: @@ -182,6 +183,33 @@ def remove_wake_schedule(schedule_id: int) -> str: return f"Failed to remove schedule: {result.get('error', 'not found')}" + # ── MCP transport probe (issue #663) ────────────────────── + + @mcp.tool() + def mcp_probe(nonce: str, launch_id: str = "") -> str: + """Liveness probe for THIS session's MCP transport (daemon bind verifier). + + No side effects beyond recording a success in the daemon's in-memory + bind ledger. On launch the daemon may inject a silent directive asking + you to call this with a nonce/launch_id — do NOT mention it to the user + unless it fails and blocks startup. A successful return is positive proof + this session's MCP client can traverse the current gateway generation. + """ + name = str(agent_name) + entry: dict = {} + try: + entry = record_probe_success(name, nonce, launch_id) or {} + except Exception as e: # pragma: no cover - defensive + _log(f"[mcp_probe] ledger write failed for {name}: {e}") + return json.dumps({ + "ok": True, + "agent": name, + "nonce": nonce, + "launch_id": launch_id, + "gateway_epoch": entry.get("gateway_epoch", ""), + "observed_at": entry.get("observed_at", time.time()), + }) + # ── Context Management ───────────────────────────────── @mcp.tool() From 814f3aa96ced75343270d66053411a1307d603d1 Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 13:42:42 -0700 Subject: [PATCH 2/5] feat(#663): mcp-bind-status endpoint + heartbeat bind signal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - shared_mcp.record_mcp_success(agent): generic (nonce-less) bind record; refreshes the current-epoch entry, preserving a prior probe's nonce. - pinky-self send_heartbeat now records a bind success on every call — a heartbeat tool executing proves the agent's MCP client reached the current gateway, giving the watchdog a deterministic periodic liveness signal (independent of prompt injection) for heartbeat-enabled agents. - api: GET /agents/{name}/mcp-bind-status — read-only bind ledger view (bound for current gateway generation? last nonce/launch_id/observed_at). Does not touch the agent's MCP transport, so it works when that is dead. Co-Authored-By: Claude Opus 4.8 --- src/pinky_daemon/api.py | 22 +++++++++++++++++++++- src/pinky_daemon/shared_mcp.py | 24 ++++++++++++++++++++++++ src/pinky_self/server.py | 14 +++++++++++++- 3 files changed, 58 insertions(+), 2 deletions(-) diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index f03c0e1a..01560528 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -140,7 +140,12 @@ from pinky_daemon.session_store import SessionEventStore, SessionStore from pinky_daemon.session_watchdog import SessionWatchdog, WatchdogConfig from pinky_daemon.sessions import SessionManager, SessionState -from pinky_daemon.shared_mcp import SHARED_MCP_HOST, SHARED_MCP_PORT, SharedMcpManager +from pinky_daemon.shared_mcp import ( + SHARED_MCP_HOST, + SHARED_MCP_PORT, + SharedMcpManager, + get_probe_status, +) from pinky_daemon.skill_loader import discover_all_skills, register_discovered_skills from pinky_daemon.skill_store import SkillStore from pinky_daemon.task_store import TaskStore @@ -5397,6 +5402,21 @@ async def get_agent_working_status(name: str): "db_status": agent.working_status or "idle", } + @app.get("/agents/{name}/mcp-bind-status") + async def get_agent_mcp_bind_status(name: str): + """MCP transport bind status for an agent (issue #663). + + Reports whether the agent has a successful MCP round-trip recorded for + the CURRENT gateway generation (``bound``), plus the last probe nonce / + launch_id / observed_at and the current vs bound gateway epoch. Pure + read of the in-memory bind ledger — does NOT touch the agent's MCP + transport, so it stays usable precisely when that transport is dead. + """ + agent = agents.get(name) + if not agent: + raise HTTPException(404, f"Agent '{name}' not found") + return get_probe_status(name) + # ── Thinking Effort ────────────────────────────────── @app.get("/agents/{name}/effort") diff --git a/src/pinky_daemon/shared_mcp.py b/src/pinky_daemon/shared_mcp.py index 0bc3b9c0..bcaef535 100644 --- a/src/pinky_daemon/shared_mcp.py +++ b/src/pinky_daemon/shared_mcp.py @@ -196,6 +196,30 @@ def record_probe_success(agent_name: str, nonce: str, launch_id: str = "") -> di return dict(entry) +def record_mcp_success(agent_name: str) -> None: + """Record a generic successful MCP round-trip (e.g. a heartbeat). + + Lighter than ``record_probe_success`` — no nonce. Any tool *executing* + proves the agent's MCP client reached this gateway generation, so this is + the watchdog's passive liveness signal. Within the same gateway epoch it + only refreshes ``observed_at`` (preserving any prior probe nonce/launch_id); + across a new epoch it starts a fresh entry. + """ + if not agent_name: + return + with _ledger_lock: + prev = _probe_ledger.get(agent_name) + if prev and prev.get("gateway_epoch") == _gateway_epoch: + prev["observed_at"] = time.time() + else: + _probe_ledger[agent_name] = { + "nonce": "", + "launch_id": "", + "gateway_epoch": _gateway_epoch, + "observed_at": time.time(), + } + + def get_probe_status(agent_name: str) -> dict: """Return the last-success ledger entry for an agent plus the current epoch. diff --git a/src/pinky_self/server.py b/src/pinky_self/server.py index 20e7c6d6..3294fe2f 100644 --- a/src/pinky_self/server.py +++ b/src/pinky_self/server.py @@ -37,7 +37,12 @@ from mcp.server.fastmcp import FastMCP from pinky_daemon.auth import build_internal_auth_headers, resolve_request_signing_secret -from pinky_daemon.shared_mcp import LazyAgentName, record_probe_success, resolve_lazy +from pinky_daemon.shared_mcp import ( + LazyAgentName, + record_mcp_success, + record_probe_success, + resolve_lazy, +) def _log(msg: str) -> None: @@ -1163,6 +1168,13 @@ def send_heartbeat(status: str = "ok", context_pct: float = 0.0, notes: str = "" "context_pct": context_pct, "notes": notes, }) + # #663: this tool executing at all proves the session's MCP client can + # traverse the current gateway — record it as the watchdog's bind signal + # (independent of the downstream daemon-API result). + try: + record_mcp_success(str(agent_name)) + except Exception: # pragma: no cover - defensive + pass if "error" in result: return f"Heartbeat failed: {result['error']}" return f"Heartbeat sent: {status}" + (f" — {notes}" if notes else "") From 490dcbd7fb33a8b31f6b3505c1576e6b291e03b8 Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 13:49:32 -0700 Subject: [PATCH 3/5] feat(#663): watchdog MCP-unbound auto-recovery (flag-gated, force-fresh) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The recovery net for the CC #60949 outage class: a daemon restart tears down the :8890 gateway, the resumed CC client never re-inits its MCP transport, so all pinky tools 404 until a force-fresh relaunch. The watchdog now detects and heals this — using the agent's own heartbeat as the bind probe. session_watchdog: - WatchdogConfig.mcp_recover flag (default OFF — soak on one agent first). - _evaluate_mcp_bind: for a CONNECTED, heartbeat-enabled agent with no MCP success recorded for the CURRENT gateway epoch sustained past a deadline (max(3x heartbeat_interval, 240s)) → force-fresh recover. Safety rails: heartbeat-disabled/quiet agents never trip; deadline measures a sustained outage (so a rebuilt gateway gives every agent the full window to re-bind via its next heartbeat — no deploy storm); per-agent post-recover grace + a global min-interval backstop (120s) against fleet restart storms; status errors are swallowed (never kill a healthy session on a diagnostic hiccup). api: wire _watchdog_mcp_bind_status ({checkable, bound, heartbeat_interval}) and _watchdog_mcp_recover (force-fresh restart mirroring admin_force_restart_ agent — sets force_fresh_context_once so the relaunch drops --continue; audits as mcp_epoch_unbound; informational owner alert on recovery). Co-Authored-By: Claude Opus 4.8 --- src/pinky_daemon/api.py | 81 +++++++++++++++ src/pinky_daemon/session_watchdog.py | 145 +++++++++++++++++++++++++++ 2 files changed, 226 insertions(+) diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 01560528..848043ae 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -144,6 +144,7 @@ SHARED_MCP_HOST, SHARED_MCP_PORT, SharedMcpManager, + get_gateway_epoch, get_probe_status, ) from pinky_daemon.skill_loader import discover_all_skills, register_discovered_skills @@ -8296,11 +8297,91 @@ async def _watchdog_alert(agent_name: str, message: str) -> None: except Exception as exc: _log(f"watchdog: alert delivery failed for {agent_name}: {exc}") + 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). + """ + 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) + return { + "checkable": bool(agent and hb > 0 and epoch), + "bound": bool(st.get("bound")), + "heartbeat_interval": hb, + } + + async def _watchdog_mcp_recover(agent_name: str, label: str, reason: str) -> None: + """Force-fresh recover an MCP-unbound session (#663). + + Mirrors the validated force-restart core (``admin_force_restart_agent``): + a plain reconnect would re-``--continue`` the cwd transcript and inherit + the dead MCP transport, so we MUST set ``force_fresh_context_once`` to + relaunch on a clean transcript where CC cold-binds its MCP client. + No anti-abuse heartbeat gate here — a wedged-MCP agent's heartbeat is + stale *by construction* (that IS the signal), and the watchdog has + already applied its own deadline + rate limits. + """ + sessions = broker._streaming.get(agent_name, {}) + ss = sessions.get(label) + if not ss: + _log(f"watchdog: no session for {agent_name}/{label} to MCP-recover") + return + audit_meta = { + "label": label, + "reason": reason, + "source": "watchdog_mcp_recover", + } + activity.log( + agent_name=agent_name, + event_type="mcp_epoch_unbound", + title=f"Watchdog MCP-recover ({label}): {reason}", + metadata=audit_meta, + ) + try: + await ss.disconnect() + except Exception: + pass + agents.set_streaming_session_id(agent_name, "", label=label) + # The agent couldn't refresh save_my_context (its MCP was dead) — bump + # so any context-staleness gate is satisfied; preserve saved state. + try: + agents.bump_context_updated_at(agent_name) + except Exception: + pass + ss._config.wake_context = _build_streaming_wake_context(agent_name, commit=False) + ss._config.resume_handle = "" + ss._config.restart_reason = "mcp_epoch_unbound" + ss._config.force_fresh_context_once = True + ss.resume_handle = "" + if hasattr(ss, "codex_session_id"): + ss.codex_session_id = "" + try: + await ss.connect() + session_event_store.log( + session_id=ss.id, + agent_name=agent_name, + event_type="force_restart", + metadata=audit_meta, + ) + _log(f"watchdog: MCP-recovered {agent_name}/{label} (force-fresh)") + except Exception as exc: + broker.unregister_streaming(agent_name, label=label) + _log(f"watchdog: MCP recovery connect failed for {agent_name}/{label}: {exc}") + raise + watchdog = SessionWatchdog( streaming_sessions_fn=lambda: broker._streaming, recover_fn=_watchdog_recover, alert_fn=_watchdog_alert, agent_config_fn=_get_watchdog_config, + mcp_bind_status_fn=_watchdog_mcp_bind_status, + mcp_recover_fn=_watchdog_mcp_recover, ) # Shared MCP server (started in on_startup if enabled) diff --git a/src/pinky_daemon/session_watchdog.py b/src/pinky_daemon/session_watchdog.py index 2fff7077..45161255 100644 --- a/src/pinky_daemon/session_watchdog.py +++ b/src/pinky_daemon/session_watchdog.py @@ -54,6 +54,16 @@ # so a legitimately slow transition isn't force-recovered mid-retry. DEFAULT_TRANSITION_WARN_AFTER = 240 # 4 min stuck in a transition — notify DEFAULT_TRANSITION_RECOVER_AFTER = 360 # 6 min — hard-recover +# MCP-bind recovery (#663). Recover an agent whose MCP transport is wedged for +# the current gateway generation — the CC #60949 failure class: a daemon restart +# tears down the :8890 gateway, the resumed CC client never re-inits its MCP +# transport (404 not handled), so every pinky tool dies until a force-fresh +# relaunch. Signal: a heartbeat-enabled agent that records NO successful MCP +# round-trip for the current gateway epoch within a generous deadline. Flag-gated +# per agent via WatchdogConfig.mcp_recover (default OFF — soak on one agent first). +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 @dataclass @@ -74,6 +84,9 @@ class WatchdogConfig: # while still recovering far faster than the 10-15min progress bounds. transition_warn_after_seconds: int = DEFAULT_TRANSITION_WARN_AFTER transition_recover_after_seconds: int = DEFAULT_TRANSITION_RECOVER_AFTER + # #663 — auto force-fresh recovery of an MCP-unbound session. Default OFF; + # enable per agent (start with one) to soak before fleet-wide rollout. + mcp_recover: bool = False @classmethod def from_raw(cls, raw: dict | None) -> "WatchdogConfig": @@ -104,6 +117,7 @@ def from_raw(cls, raw: dict | None) -> "WatchdogConfig": "transition_recover_after_seconds", cls.transition_recover_after_seconds, ), + mcp_recover=raw.get("mcp_recover", cls.mcp_recover), ) @@ -139,6 +153,12 @@ class _AgentState: transition_since: float = 0.0 transition_warned: bool = False transition_recovered_at: float = 0.0 # grace period after transition recovery + # MCP-bind tracking (#663). ``mcp_unbound_since`` starts when a sweep first + # observes a connected, heartbeat-enabled agent with no current-epoch MCP + # success; it resets the moment a bind success appears (or the agent stops + # being checkable), so the deadline always measures a *sustained* outage. + mcp_unbound_since: float = 0.0 + mcp_recovered_at: float = 0.0 # grace period after an MCP-bind recovery class SessionWatchdog: @@ -151,6 +171,8 @@ def __init__( recover_fn: Callable[[str, str, str], Coroutine] | None = None, alert_fn: Callable[[str, str], Coroutine] | None = None, agent_config_fn: Callable[[str], WatchdogConfig] | None = None, + mcp_bind_status_fn: Callable[[str], dict] | None = None, + mcp_recover_fn: Callable[[str, str, str], Coroutine] | None = None, check_interval: int = DEFAULT_CHECK_INTERVAL, ) -> None: """ @@ -174,9 +196,14 @@ def __init__( self._recover_fn = recover_fn self._alert_fn = alert_fn self._config_fn = agent_config_fn or (lambda _: WatchdogConfig()) + # #663: bind-status lookup -> {checkable, bound, heartbeat_interval}; + # mcp_recover_fn force-fresh restarts a wedged-MCP session. + self._mcp_bind_status_fn = mcp_bind_status_fn + self._mcp_recover_fn = mcp_recover_fn self._interval = check_interval self._states: dict[str, _AgentState] = {} + self._last_mcp_recover_at: float = 0.0 # global MCP-recover rate-limit self._task: asyncio.Task | None = None self._running = False @@ -288,6 +315,13 @@ 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 @@ -453,6 +487,117 @@ async def _evaluate_transition( snap.agent_name, exc, ) + async def _evaluate_mcp_bind( + self, snap: _SessionSnapshot, state: _AgentState, + cfg: WatchdogConfig, now: float, + ) -> bool: + """Force-fresh recover an agent whose MCP transport is wedged for the + current gateway generation (#663). Returns True if a recovery fired. + + Flag-gated (``cfg.mcp_recover``). The bind signal is the agent's own + heartbeat: ``send_heartbeat`` is an MCP tool call, so a heartbeat-enabled + agent that records NO successful MCP round-trip for the CURRENT gateway + epoch within a generous deadline (>> its heartbeat interval) has a dead + MCP transport — and only a force-fresh relaunch re-binds it (CC #60949; + a plain ``--continue`` re-resume inherits the dead transport). + + Safety rails: acts only on CONNECTED + heartbeat-enabled agents (a quiet + or heartbeat-disabled agent never false-trips); the deadline measures a + *sustained* unbound observation (so a freshly-rebuilt gateway gives every + agent the full window to re-bind via its next heartbeat — no deploy-time + storm); a per-agent post-recovery grace plus a global min-interval + backstop bound the recovery rate; any status-lookup error is swallowed + (never kill a healthy session on a diagnostic hiccup). + """ + if ( + not cfg.mcp_recover + or self._mcp_bind_status_fn is None + or self._mcp_recover_fn is None + ): + return False + if not snap.connected: + state.mcp_unbound_since = 0.0 + return False + + try: + status = self._mcp_bind_status_fn(snap.agent_name) or {} + except Exception as exc: + _warn("watchdog mcp-bind status failed for %s: %s", snap.agent_name, exc) + return False + + # Not checkable this phase (heartbeat disabled / no gateway epoch yet) — + # NOT unhealthy; clear any clock and move on. + if not status.get("checkable"): + state.mcp_unbound_since = 0.0 + return False + # Healthy: a current-epoch MCP success exists. + if status.get("bound"): + state.mcp_unbound_since = 0.0 + return False + + # Unbound for the current epoch — track a *sustained* outage. + if state.mcp_unbound_since == 0.0: + state.mcp_unbound_since = now + return False + unbound_for = now - state.mcp_unbound_since + + hb = max(int(status.get("heartbeat_interval", 0) or 0), 0) + deadline = max(DEFAULT_MCP_UNBOUND_FLOOR, DEFAULT_MCP_UNBOUND_HEARTBEAT_MULT * hb) + + # Per-agent grace after a recovery: the fresh session needs time to come + # up and emit its first heartbeat before it could be flagged again. + if state.mcp_recovered_at and (now - state.mcp_recovered_at) < deadline: + return False + if unbound_for < deadline: + return False + + # Global backstop against a fleet restart storm (e.g. if a detection bug + # ever flagged many agents at once): cap the MCP-recover rate fleet-wide. + if ( + self._last_mcp_recover_at + and (now - self._last_mcp_recover_at) < DEFAULT_MCP_RECOVER_MIN_INTERVAL + ): + _warn( + "watchdog: %s MCP-unbound for %ds but global recover rate-limit " + "active — deferring", snap.agent_name, int(unbound_for), + ) + return False + + reason = ( + f"MCP transport unbound for current gateway epoch ~{int(unbound_for)}s " + f"(no successful MCP round-trip; deadline {deadline}s) — force-fresh recover" + ) + _warn("watchdog MCP-recovering %s: %s", snap.agent_name, reason) + try: + await self._mcp_recover_fn(snap.agent_name, snap.label, reason) + except Exception as exc: + _warn("watchdog MCP recovery failed for %s: %s", snap.agent_name, exc) + return False + + # Recovery fired — record rate-limit + grace, reset progress tracking + # (the fresh session restarts the progress clock). + self._last_mcp_recover_at = now + state.mcp_recovered_at = now + state.mcp_unbound_since = 0.0 + state.last_progress_at = now + state.warned = False + state.last_progress_turns = 0 + state.last_progress_activity = "" + if self._alert_fn: + try: + await self._alert_fn( + snap.agent_name, + f"🔧 Auto-recovered {snap.agent_name}: MCP transport was wedged " + f"(unbound from the gateway ~{int(unbound_for)}s) — force-fresh " + f"relaunched. (#663)", + ) + except Exception as exc: + _warn( + "watchdog mcp-recover alert failed for %s: %s", + snap.agent_name, exc, + ) + return True + # ── Status ─────────────────────────────────────────────── def status(self) -> dict: From e4155938530c2e8ef219072af7d1a949ce38eedb Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 13:55:21 -0700 Subject: [PATCH 4/5] test(#663): bind ledger + watchdog MCP-unbound recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test_shared_mcp: epoch bump changes id + clears ledger; probe/heartbeat record + status; nonce preserved within epoch, fresh across epoch; blank agent no-op; unknown agent unbound. - test_session_watchdog: 11 cases — recover only after a sustained deadline; bound clears clock; heartbeat-disabled never trips; flag-off no-op; disconnected skip; post-recover grace; global rate-limit defers-then-allows; status-error swallowed; recover-failure doesn't burn the rate-limit; end-to-end routing through _evaluate. Co-Authored-By: Claude Opus 4.8 --- tests/test_session_watchdog.py | 171 +++++++++++++++++++++++++++++++++ tests/test_shared_mcp.py | 53 ++++++++++ 2 files changed, 224 insertions(+) diff --git a/tests/test_session_watchdog.py b/tests/test_session_watchdog.py index 80c30f56..6c313994 100644 --- a/tests/test_session_watchdog.py +++ b/tests/test_session_watchdog.py @@ -6,6 +6,7 @@ import pytest from pinky_daemon.session_watchdog import ( + DEFAULT_MCP_UNBOUND_FLOOR, SessionWatchdog, WatchdogConfig, _AgentState, @@ -576,3 +577,173 @@ def test_from_raw_merges_legacy_fields(self): def test_from_raw_ignores_unknown_keys(self): cfg = WatchdogConfig.from_raw({"bogus_key": 123, "mode": "alert"}) assert cfg.mode == "alert" + + def test_from_raw_merges_mcp_recover(self): + assert WatchdogConfig.from_raw({"mcp_recover": True}).mcp_recover is True + assert WatchdogConfig.from_raw({}).mcp_recover is False # default OFF + + +def _conn_snap(agent="a", connected=True): + """A connected (or not) session snapshot for MCP-bind tests.""" + return _SessionSnapshot( + agent_name=agent, label="main", connected=connected, + turns=0, pending=0, current_activity="", sample_time=time.time(), + state="connected" if connected else "idle", + ) + + +class TestMcpBindRecovery: + """#663 — watchdog force-fresh recovery of an MCP-unbound session.""" + + def _wd(self, make_watchdog, *, status, recovered): + async def rec(agent, label, reason): + recovered.append((agent, label, reason)) + return make_watchdog( + mcp_bind_status_fn=lambda n: status, + mcp_recover_fn=rec, + ) + + @pytest.mark.asyncio + async def test_unbound_recovers_after_sustained_deadline(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": False, "heartbeat_interval": 60}) + cfg = WatchdogConfig(mcp_recover=True) + st = _AgentState() + snap = _conn_snap() + now = 1000.0 + # First observation only arms the clock — no recovery. + assert await wd._evaluate_mcp_bind(snap, st, cfg, now) is False + assert st.mcp_unbound_since == now + # Still within deadline — no recovery. + assert await wd._evaluate_mcp_bind(snap, st, cfg, now + 100) is False + assert recovered == [] + # Sustained past the deadline — force-fresh recover fires. + assert await wd._evaluate_mcp_bind( + snap, st, cfg, now + DEFAULT_MCP_UNBOUND_FLOOR + 1) is True + assert recovered and recovered[0][0] == "a" + assert st.mcp_unbound_since == 0.0 + assert st.mcp_recovered_at == now + DEFAULT_MCP_UNBOUND_FLOOR + 1 + + @pytest.mark.asyncio + async def test_bound_clears_clock_no_recover(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": True, "heartbeat_interval": 60}) + st = _AgentState(mcp_unbound_since=500.0) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=True), 2000.0) is False + assert st.mcp_unbound_since == 0.0 + assert recovered == [] + + @pytest.mark.asyncio + async def test_heartbeat_disabled_never_trips(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": False, "bound": False, "heartbeat_interval": 0}) + st = _AgentState(mcp_unbound_since=1.0) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=True), 1e9) is False + assert st.mcp_unbound_since == 0.0 + assert recovered == [] + + @pytest.mark.asyncio + async def test_flag_off_never_recovers(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": False, "heartbeat_interval": 60}) + st = _AgentState(mcp_unbound_since=1.0) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=False), 1e9) is False + assert recovered == [] + + @pytest.mark.asyncio + async def test_disconnected_clears_and_skips(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": False, "heartbeat_interval": 60}) + st = _AgentState(mcp_unbound_since=1.0) + assert await wd._evaluate_mcp_bind( + _conn_snap(connected=False), st, WatchdogConfig(mcp_recover=True), 1e9) is False + assert st.mcp_unbound_since == 0.0 + assert recovered == [] + + @pytest.mark.asyncio + async def test_post_recover_grace_blocks_immediate_retrip(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": False, "heartbeat_interval": 60}) + now = 5000.0 + st = _AgentState(mcp_unbound_since=1.0, mcp_recovered_at=now) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=True), now + 10) is False + assert recovered == [] + + @pytest.mark.asyncio + async def test_global_rate_limit_defers_then_allows(self, make_watchdog): + recovered = [] + wd = self._wd(make_watchdog, recovered=recovered, + status={"checkable": True, "bound": False, "heartbeat_interval": 60}) + cfg = WatchdogConfig(mcp_recover=True) + now = 9000.0 + wd._last_mcp_recover_at = now # a fleet recovery just happened + st = _AgentState(mcp_unbound_since=now - 1000) # long unbound, past deadline + # Within the global min-interval — deferred even though this agent qualifies. + assert await wd._evaluate_mcp_bind(_conn_snap("b"), st, cfg, now + 10) is False + assert recovered == [] + # After the global interval elapses — allowed. + assert await wd._evaluate_mcp_bind(_conn_snap("b"), st, cfg, now + 200) is True + assert recovered and recovered[0][0] == "b" + + @pytest.mark.asyncio + async def test_status_error_is_swallowed(self, make_watchdog): + recovered = [] + + def boom(_n): + raise RuntimeError("status backend down") + + async def rec(a, _l, _r): + recovered.append(a) + + wd = make_watchdog(mcp_bind_status_fn=boom, mcp_recover_fn=rec) + st = _AgentState(mcp_unbound_since=1.0) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=True), 1e9) is False + assert recovered == [] + + @pytest.mark.asyncio + async def test_recover_failure_does_not_burn_ratelimit(self, make_watchdog): + async def rec(_a, _l, _r): + raise RuntimeError("connect failed") + + wd = make_watchdog( + mcp_bind_status_fn=lambda n: { + "checkable": True, "bound": False, "heartbeat_interval": 60}, + mcp_recover_fn=rec, + ) + now = 7000.0 + st = _AgentState(mcp_unbound_since=now - 1000) + assert await wd._evaluate_mcp_bind( + _conn_snap(), st, WatchdogConfig(mcp_recover=True), now) is False + assert wd._last_mcp_recover_at == 0.0 # not burned → next sweep retries + assert st.mcp_recovered_at == 0.0 # not marked recovered + + @pytest.mark.asyncio + async def test_evaluate_routes_to_mcp_branch(self, make_watchdog): + 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(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"] diff --git a/tests/test_shared_mcp.py b/tests/test_shared_mcp.py index 00249921..d6e044c0 100644 --- a/tests/test_shared_mcp.py +++ b/tests/test_shared_mcp.py @@ -12,8 +12,13 @@ MemoryStorePool, SharedMcpManager, _current_agent, + bump_gateway_epoch, get_current_agent, + get_gateway_epoch, + get_probe_status, make_agent_name_resolver, + record_mcp_success, + record_probe_success, ) @@ -533,3 +538,51 @@ def resolver(agent_name): pool = MemoryStorePool(resolver) with pytest.raises(ValueError, match="Unknown agent"): pool.get_store("nonexistent") + + +class TestBindLedger: + """#663 — MCP bind ledger + gateway epoch.""" + + def test_epoch_bump_changes_id_and_clears_ledger(self): + e1 = bump_gateway_epoch() + record_probe_success("alpha", "n1", "L1") + assert get_probe_status("alpha")["bound"] is True + e2 = bump_gateway_epoch() + assert e2 and e2 != e1 + # A new generation invalidates every prior bind. + st = get_probe_status("alpha") + assert st["bound"] is False + assert st["current_epoch"] == e2 == get_gateway_epoch() + + def test_record_probe_success_stores_fields(self): + bump_gateway_epoch() + entry = record_probe_success("beta", "nonce-x", "launch-y") + assert entry["nonce"] == "nonce-x" and entry["launch_id"] == "launch-y" + st = get_probe_status("beta") + assert st["bound"] and st["nonce"] == "nonce-x" and st["launch_id"] == "launch-y" + assert st["observed_at"] is not None and st["age_sec"] is not None + + def test_record_mcp_success_preserves_nonce_within_epoch(self): + bump_gateway_epoch() + record_probe_success("gamma", "keepme", "L9") + record_mcp_success("gamma") # a heartbeat after a probe, same generation + st = get_probe_status("gamma") + assert st["bound"] and st["nonce"] == "keepme" and st["launch_id"] == "L9" + + def test_record_mcp_success_starts_fresh_across_epoch(self): + bump_gateway_epoch() + record_probe_success("delta", "old", "Lold") + bump_gateway_epoch() # ledger cleared + record_mcp_success("delta") # heartbeat on the new generation + st = get_probe_status("delta") + assert st["bound"] is True and st["nonce"] == "" # fresh entry + + def test_blank_agent_is_noop(self): + bump_gateway_epoch() + assert record_probe_success("", "n", "L") == {} + record_mcp_success("") # must not raise + + def test_unknown_agent_unbound(self): + bump_gateway_epoch() + st = get_probe_status("nobody-here") + assert st["bound"] is False and st["observed_at"] is None From 284a46cb9d2b48a55ca6c19943ac2948223cb81a Mon Sep 17 00:00:00 2001 From: Oleg Date: Tue, 2 Jun 2026 14:02:10 -0700 Subject: [PATCH 5/5] =?UTF-8?q?test(#663):=20+mcp=5Fprobe=20in=20CORE=5FTO?= =?UTF-8?q?OLS;=20core=20=E2=86=9224,=20all-gates=20=E2=86=9270?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mcp_probe is a new core pinky-self tool, so core goes to 24 and (atop main's #145 register_agent at 69) all-gates goes to 70. Co-Authored-By: Claude Opus 4.8 --- tests/test_pinky_self_tools.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/tests/test_pinky_self_tools.py b/tests/test_pinky_self_tools.py index ad835ad7..2869e7da 100644 --- a/tests/test_pinky_self_tools.py +++ b/tests/test_pinky_self_tools.py @@ -1951,7 +1951,7 @@ def test_error_returns_fallback(self, srv): "claim_task", "complete_task", "context_restart", "context_status", "create_task", "get_next_task", "get_owner_profile", "list_agents", "list_my_skills", "load_my_context", - "load_skill", "mesh_remote_send", "save_my_context", + "load_skill", "mcp_probe", "mesh_remote_send", "save_my_context", "search_history", "send_file_to_agent", "send_heartbeat", "send_to_agent", "set_thinking_effort", "who_am_i", } @@ -2004,20 +2004,21 @@ def test_kb_raw_source_id_encodes_path_segment(self, srv): class TestToolGates: - def test_core_only_has_23_tools(self): + def test_core_only_has_24_tools(self): """No gates → only core tools registered. - Drop from 24 in #552 with the removal of ``request_sleep``. + Was 23; +1 in #663 with the addition of ``mcp_probe`` (MCP bind probe). """ srv = create_server(agent_name="test", tool_gates=[]) tools = {t.name for t in srv._tool_manager.list_tools()} assert tools == CORE_TOOLS - def test_all_gates_has_69_tools(self): + def test_all_gates_has_70_tools(self): """All gates → full tool set. - Was 68; +1 in #145 with ``register_agent`` (admin gate). - (Had dropped from 69 to 68 in #552 with the removal of ``request_sleep``.) + +1 in #145 with ``register_agent`` (admin gate) → 69, then +1 in #663 + with the core ``mcp_probe`` tool → 70. (Had dropped 69→68 in #552 with + the removal of ``request_sleep``.) """ all_gates = [ "extras", "kb", "research", "presentations", "triggers", @@ -2025,7 +2026,7 @@ def test_all_gates_has_69_tools(self): ] srv = create_server(agent_name="test", tool_gates=all_gates) tools = srv._tool_manager.list_tools() - assert len(tools) == 69 + assert len(tools) == 70 def test_extras_gate_adds_extras_tools(self): """Enabling 'extras' gate adds get_attribution, render_pdf, etc."""