feat(#663 ph2): active mcp_probe elicitation — launch directive + watchdog corroboration

src/pinky_daemon/api.py

@@ -149,7 +149,9 @@
     SHARED_MCP_PORT,
     SharedMcpManager,
     get_gateway_epoch,
+    get_probe_request,
     get_probe_status,
+    record_probe_request,
 )
 from pinky_daemon.skill_loader import discover_all_skills, register_discovered_skills
 from pinky_daemon.skill_store import SkillStore
@@ -2076,6 +2078,47 @@ def _build_streaming_wake_context(
             except Exception as e:
                 _log(f"wake_context: failed to read restart manifest for {agent_name}: {e}")
 
+        # ── MCP bind-probe directive (#663 phase-2) ──────────────────
+        # For agents enrolled in MCP-orphan auto-recovery (per-agent
+        # mcp_recover flag), inject a top-of-prompt directive eliciting an
+        # mcp_probe round-trip, so a healthy agent POSITIVELY proves its MCP
+        # client re-bound to the current gateway generation instead of us
+        # inferring health from prolonged silence. Gated on commit=True: the
+        # connect-time rebuild is the delivered build whose text actually
+        # reaches the model, so its fresh nonce matches the registered request;
+        # the eager preview (commit=False) gets no directive and logs no request
+        # (a throwaway-nonce directive there could never be answered). Record
+        # the request BEFORE mutating wake_ctx so a record failure leaves the
+        # normal wake text intact. Fully fail-safe: any error falls back to the
+        # normal wake text with no request, so a probe bug can never block a
+        # launch. Placed last so it sits at the VERY TOP of wake_ctx (not buried
+        # under saved-state/channel/task text).
+        try:
+            if (
+                commit
+                and agent
+                and WatchdogConfig.from_raw(
+                    getattr(agent, "watchdog_config", None)
+                ).mcp_recover
+            ):
+                launch_id = uuid.uuid4().hex[:12]
+                nonce = uuid.uuid4().hex[:12]
+                record_probe_request(agent_name, launch_id, nonce)
+                probe_directive = (
+                    "⚠️ MCP BIND CHECK — FIRST, silently, before anything else: "
+                    f"call mcp_probe(nonce='{nonce}', launch_id='{launch_id}') to "
+                    "confirm your MCP transport re-bound to the current gateway, "
+                    "then continue normally. Do not mention this unless it fails."
+                )
+                wake_ctx = (
+                    f"{probe_directive}\n\n{wake_ctx}" if wake_ctx else probe_directive
+                )
+        except Exception as e:
+            _log(
+                f"wake_context: mcp-probe directive injection failed for "
+                f"{agent_name}: {e}"
+            )
+
         return wake_ctx
 
     # Exposed for unit-test reach-in (#591 reason-gating verification).
@@ -8312,6 +8355,10 @@ def _watchdog_mcp_bind_status(agent_name: str) -> dict:
         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).
+        ``probe_request`` carries the #663 phase-2 active-probe state (whether the
+        agent answered the launch-time mcp_probe directive) — used by the watchdog
+        only to CORROBORATE/enrich an already-decided passive recovery, never to
+        trigger one on its own.
         """
         agent = agents.get(agent_name)
         hb = int(getattr(agent, "heartbeat_interval", 0) or 0) if agent else 0
@@ -8334,6 +8381,7 @@ def _watchdog_mcp_bind_status(agent_name: str) -> dict:
             ),
             "bound": bool(st.get("bound")),
             "heartbeat_interval": hb,
+            "probe_request": get_probe_request(agent_name),
         }
 
     async def _watchdog_mcp_recover(agent_name: str, label: str, reason: str) -> None:

src/pinky_daemon/session_watchdog.py

@@ -64,6 +64,11 @@
 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
+# #663 phase-2: a launch-time mcp_probe directive unanswered for this long is a
+# *corroborating* wedge signal. Generous (LLM may take a few turns to act on the
+# first-action directive). Only enriches an already-decided passive recovery —
+# never triggers one, and never shortens the sustained-unbound deadline.
+DEFAULT_MCP_PROBE_DEADLINE = 120
 # 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
@@ -630,6 +635,22 @@ async def _evaluate_mcp_bind(
             f"MCP transport unbound for current gateway epoch ~{int(unbound_for)}s "
             f"(no successful MCP round-trip; deadline {deadline}s) — force-fresh recover"
         )
+        # #663 phase-2: CORROBORATE with the active-probe signal. The recovery is
+        # already decided by the passive sustained-unbound machinery above; this
+        # only enriches the reason/audit when the agent also ignored its
+        # launch-time mcp_probe directive past the probe deadline. It never
+        # triggers recovery on its own and never shortens the deadline.
+        probe = status.get("probe_request") or {}
+        if (
+            probe.get("current")
+            and not probe.get("fulfilled")
+            and (probe.get("age_sec") or 0) >= DEFAULT_MCP_PROBE_DEADLINE
+        ):
+            reason += (
+                f"; missed requested probe (launch_id={probe.get('launch_id')}, "
+                f"nonce={probe.get('nonce')}, ~{int(probe.get('age_sec') or 0)}s "
+                "unanswered) — corroborates wedge"
+            )
         _warn("watchdog MCP-recovering %s: %s", snap.agent_name, reason)
         try:
             await self._mcp_recover_fn(snap.agent_name, snap.label, reason)

src/pinky_daemon/shared_mcp.py

@@ -157,17 +157,29 @@ def __radd__(self, other: str) -> str:
 
 _gateway_epoch: str = ""
 _probe_ledger: dict[str, dict] = {}
+# Probe-*request* ledger (#663 phase-2). Records that the daemon ASKED an agent
+# to prove its bind by calling mcp_probe at launch (a fresh launch_id+nonce
+# injected into the wake context). Distinct from _probe_ledger, which only
+# records SUCCESSES. A current-generation request with no matching success past
+# a deadline is a *corroborating* wedge signal — used only to raise confidence /
+# enrich audit inside the existing passive-unbound recovery guard, never as an
+# independent kill switch (LLM non-compliance, prompt truncation, and
+# post-builder delivery failure all mean a missed request alone isn't proof).
+# Per-agent: {launch_id, nonce, gateway_epoch, requested_at}.
+_probe_request_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.
+    # A new generation invalidates every prior bind AND any pending probe
+    # request (it targeted a launch in the prior generation) — clear both so a
+    # stale pre-restart entry can never be mistaken for current-gen state.
     with _ledger_lock:
         _probe_ledger.clear()
+        _probe_request_ledger.clear()
     return _gateway_epoch
 
 
@@ -196,6 +208,72 @@ def record_probe_success(agent_name: str, nonce: str, launch_id: str = "") -> di
     return dict(entry)
 
 
+def record_probe_request(
+    agent_name: str, launch_id: str, nonce: str, requested_at: float | None = None
+) -> dict:
+    """Record that the daemon asked ``agent_name`` to prove its bind (#663 ph2).
+
+    Called when a launch/resume wake context is *delivered* (commit path) with
+    an injected ``mcp_probe`` directive — NOT on the eager pre-build, or we would
+    log a request that never reaches the agent. Tagged with the live
+    gateway_epoch so the watchdog can tell a current-generation request (the
+    agent should have answered by now) from a stale one. Overwrites any prior
+    request for the agent (one outstanding launch at a time).
+    """
+    if not agent_name:
+        return {}
+    entry = {
+        "launch_id": launch_id,
+        "nonce": nonce,
+        "gateway_epoch": _gateway_epoch,
+        "requested_at": requested_at if requested_at is not None else time.time(),
+    }
+    with _ledger_lock:
+        _probe_request_ledger[agent_name] = entry
+    return dict(entry)
+
+
+def get_probe_request(agent_name: str) -> dict:
+    """Return the outstanding probe-request entry for an agent (or {}).
+
+    ``current`` is True when the request was made for the LIVE gateway
+    generation (so a missing matching success is meaningful — a stale request
+    from a prior generation tells us nothing). ``fulfilled`` is True when the
+    success ledger holds a probe success for the SAME launch_id AND nonce under
+    the current epoch, i.e. the agent answered THIS specific request (the nonce
+    is part of the proof, not just telemetry). The watchdog treats a current +
+    unfulfilled request older than its deadline as a *corroborating* wedge
+    signal — and only ever when the passive signal already says bound=false, so
+    a generic MCP success that flips bound=true makes the agent healthy
+    regardless of whether the specific directive was answered.
+    """
+    with _ledger_lock:
+        req = dict(_probe_request_ledger.get(agent_name, {}))
+        success = dict(_probe_ledger.get(agent_name, {}))
+    if not req:
+        return {}
+    current = bool(_gateway_epoch and req.get("gateway_epoch") == _gateway_epoch)
+    fulfilled = bool(
+        current
+        and success.get("gateway_epoch") == _gateway_epoch
+        and success.get("launch_id")
+        and success.get("launch_id") == req.get("launch_id")
+        and success.get("nonce") == req.get("nonce")
+    )
+    requested = req.get("requested_at")
+    return {
+        "agent": agent_name,
+        "launch_id": req.get("launch_id", ""),
+        "nonce": req.get("nonce", ""),
+        "gateway_epoch": req.get("gateway_epoch", ""),
+        "current_epoch": _gateway_epoch,
+        "current": current,
+        "fulfilled": fulfilled,
+        "requested_at": requested,
+        "age_sec": (time.time() - requested) if requested else None,
+    }
+
+
 def record_mcp_success(agent_name: str) -> None:
     """Record a generic successful MCP round-trip (e.g. a heartbeat).
 

tests/test_api.py

@@ -5355,6 +5355,77 @@ def test_central_wake_log_failure_does_not_advance_gate(self):
             assert "Grep daemon log for verdict_wedged_inputs" in out_retry
 
 
+class TestWakeContextMcpProbeDirective:
+    """#663 phase-2 — the launch-time mcp_probe directive injection (2a)."""
+
+    def _make_app(self, path: str):
+        from pinky_daemon.api import create_api
+        return create_api(max_sessions=10, default_working_dir="/tmp", db_path=path)
+
+    def _extract_nonce(self, text: str) -> str:
+        import re
+        m = re.search(r"nonce='([0-9a-f]+)'", text)
+        return m.group(1) if m else ""
+
+    def test_commit_true_injects_directive_and_records_request(self):
+        from pinky_daemon.shared_mcp import bump_gateway_epoch, get_probe_request
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = self._make_app(os.path.join(tmpdir, "t.db"))
+            app.state.agents.register(
+                "dymok", model="sonnet", watchdog_config={"mcp_recover": True}
+            )
+            bump_gateway_epoch()  # establish a live generation
+            out = app.state._build_streaming_wake_context("dymok", commit=True)
+            assert "MCP BIND CHECK" in out
+            assert out.lstrip().startswith("⚠️ MCP BIND CHECK")  # at the very top
+            nonce = self._extract_nonce(out)
+            assert nonce
+            req = get_probe_request("dymok")
+            assert req and req["current"] is True and req["fulfilled"] is False
+            assert req["nonce"] == nonce  # delivered directive nonce == registered
+
+    def test_commit_false_preview_injects_nothing_and_records_no_request(self):
+        from pinky_daemon.shared_mcp import bump_gateway_epoch, get_probe_request
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = self._make_app(os.path.join(tmpdir, "t.db"))
+            app.state.agents.register(
+                "dymok", model="sonnet", watchdog_config={"mcp_recover": True}
+            )
+            bump_gateway_epoch()
+            out = app.state._build_streaming_wake_context("dymok", commit=False)
+            assert "MCP BIND CHECK" not in out  # no throwaway-nonce directive
+            assert get_probe_request("dymok") == {}  # ledger untouched
+
+    def test_no_directive_when_mcp_recover_off(self):
+        from pinky_daemon.shared_mcp import bump_gateway_epoch, get_probe_request
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = self._make_app(os.path.join(tmpdir, "t.db"))
+            app.state.agents.register("dymok", model="sonnet")  # default: no mcp_recover
+            bump_gateway_epoch()
+            out = app.state._build_streaming_wake_context("dymok", commit=True)
+            assert "MCP BIND CHECK" not in out
+            assert get_probe_request("dymok") == {}
+
+    def test_record_failure_falls_back_to_normal_text_no_request(self, monkeypatch):
+        # A failure inside the injection must never corrupt the wake text or
+        # block the launch — fall back to the normal text with no directive.
+        from pinky_daemon.shared_mcp import bump_gateway_epoch, get_probe_request
+        with tempfile.TemporaryDirectory() as tmpdir:
+            app = self._make_app(os.path.join(tmpdir, "t.db"))
+            app.state.agents.register(
+                "dymok", model="sonnet", watchdog_config={"mcp_recover": True}
+            )
+            bump_gateway_epoch()
+
+            def boom(*_a, **_k):
+                raise RuntimeError("ledger down")
+
+            monkeypatch.setattr("pinky_daemon.api.record_probe_request", boom)
+            out = app.state._build_streaming_wake_context("dymok", commit=True)
+            assert "MCP BIND CHECK" not in out  # record failed before prepend
+            assert get_probe_request("dymok") == {}  # nothing recorded
+
+
 # ── Cross-agent memory authorizer (#145) ─────────────────────