feat(memory): MemoryProvider.health_check() + Honcho/Mem0 native probes

agent/memory_provider.py

@@ -57,6 +57,31 @@ def is_available(self) -> bool:
         Should not make network calls — just check config and installed deps.
         """
 
+
+    def health_check(self) -> tuple[bool, str]:
+        """Probe the backing service and return (healthy, reason).
+
+        Default delegates to is_available() -- providers that cannot probe
+        their backend get (False, "is_available() returned False") for failures.
+        Providers that CAN probe should override to make a lightweight network
+        call (GET /health, minimal API round-trip, etc.) within ~5 seconds.
+
+        Return (True, "") on success.
+        Return (False, "<prefix>: <detail>") on failure. Use prefixes:
+          "auth:"         -- bad key / expired token
+          "unreachable:"  -- connection refused / timeout / DNS failure
+          "not_found:"    -- endpoint 404
+          "rate_limited:" -- 429
+          "unavailable:"  -- is_available() returned False (default case)
+
+        Must never raise -- doctor invokes this in a loop.
+        """
+        try:
+            ok = self.is_available()
+            return (ok, "" if ok else "unavailable: is_available() returned False")
+        except Exception as exc:  # noqa: BLE001
+            return (False, f"unavailable: is_available() raised: {exc}")
+
     @abstractmethod
     def initialize(self, session_id: str, **kwargs) -> None:
         """Initialize for a session.

hermes_cli/doctor.py

@@ -1843,29 +1843,21 @@ def _gh_authenticated() -> bool:
             from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path
             hcfg = HonchoClientConfig.from_global_config()
             _honcho_cfg_path = resolve_config_path()
-
             if not _honcho_cfg_path.exists():
                 check_warn("Honcho config not found", "run: hermes memory setup")
             elif not hcfg.enabled:
                 check_info(f"Honcho disabled (set enabled: true in {_honcho_cfg_path} to activate)")
-            elif not (hcfg.api_key or hcfg.base_url):
-                _fail_and_issue(
-                    "Honcho API key or base URL not set",
-                    "run: hermes memory setup",
-                    "No Honcho API key — run 'hermes memory setup'",
-                    issues,
-                )
             else:
-                from plugins.memory.honcho.client import get_honcho_client, reset_honcho_client
-                reset_honcho_client()
-                try:
-                    get_honcho_client(hcfg)
-                    check_ok(
-                        "Honcho connected",
-                        f"workspace={hcfg.workspace_id} mode={hcfg.recall_mode} freq={hcfg.write_frequency}",
-                    )
-                except Exception as _e:
-                    _fail_and_issue("Honcho connection failed", str(_e), f"Honcho unreachable: {_e}", issues)
+                from plugins.memory.honcho import HonchoMemoryProvider
+                _provider = HonchoMemoryProvider()
+                _healthy, _reason = _provider.health_check()
+                if _healthy:
+                    check_ok("Honcho connected",
+                             f"workspace={hcfg.workspace_id} mode={hcfg.recall_mode} freq={hcfg.write_frequency}")
+                elif _reason.startswith("auth:"):
+                    _fail_and_issue("Honcho auth rejected", _reason, f"Honcho auth error: {_reason}", issues)
+                else:
+                    _fail_and_issue("Honcho connection failed", _reason, f"Honcho unreachable: {_reason}", issues)
         except ImportError:
             _fail_and_issue(
                 "honcho-ai not installed",
@@ -1875,41 +1867,23 @@ def _gh_authenticated() -> bool:
             )
         except Exception as _e:
             check_warn("Honcho check failed", str(_e))
-    elif _active_memory_provider == "mem0":
-        try:
-            from plugins.memory.mem0 import _load_config as _load_mem0_config
-            mem0_cfg = _load_mem0_config()
-            mem0_key = mem0_cfg.get("api_key", "")
-            if mem0_key:
-                check_ok("Mem0 API key configured")
-                check_info(f"user_id={mem0_cfg.get('user_id', '?')}  agent_id={mem0_cfg.get('agent_id', '?')}")
-            else:
-                _fail_and_issue(
-                    "Mem0 API key not set",
-                    "(set MEM0_API_KEY in .env or run hermes memory setup)",
-                    "Mem0 is set as memory provider but API key is missing",
-                    issues,
-                )
-        except ImportError:
-            _fail_and_issue(
-                "Mem0 plugin not loadable",
-                "pip install mem0ai",
-                "Mem0 is set as memory provider but mem0ai is not installed",
-                issues,
-            )
-        except Exception as _e:
-            check_warn("Mem0 check failed", str(_e))
     else:
-        # Generic check for other memory providers (openviking, hindsight, etc.)
         try:
             from plugins.memory import load_memory_provider
             _provider = load_memory_provider(_active_memory_provider)
-            if _provider and _provider.is_available():
-                check_ok(f"{_active_memory_provider} provider active")
-            elif _provider:
-                check_warn(f"{_active_memory_provider} configured but not available", "run: hermes memory status")
-            else:
+            if not _provider:
                 check_warn(f"{_active_memory_provider} plugin not found", "run: hermes memory setup")
+            else:
+                _healthy, _reason = _provider.health_check()
+                if _healthy:
+                    check_ok(f"{_active_memory_provider} reachable")
+                elif _reason.startswith("auth:"):
+                    _fail_and_issue(f"{_active_memory_provider} auth rejected", _reason,
+                                    f"{_active_memory_provider} auth error: {_reason}", issues)
+                elif _reason.startswith("unreachable:"):
+                    check_warn(f"{_active_memory_provider} unreachable", _reason)
+                else:
+                    check_warn(f"{_active_memory_provider} health check failed", _reason)
         except Exception as _e:
             check_warn(f"{_active_memory_provider} check failed", str(_e))
 

plugins/memory/honcho/__init__.py

@@ -246,6 +246,29 @@ def is_available(self) -> bool:
         except Exception:
             return False
 
+
+    def health_check(self) -> tuple[bool, str]:
+        """Probe Honcho backend via get_honcho_client (validates auth + connectivity)."""
+        try:
+            from plugins.memory.honcho.client import HonchoClientConfig, get_honcho_client, reset_honcho_client
+            hcfg = HonchoClientConfig.from_global_config()
+            if not hcfg.enabled:
+                return (False, "unavailable: Honcho disabled in config (set enabled: true)")
+            if not (hcfg.api_key or hcfg.base_url):
+                return (False, "auth: no API key or base URL configured")
+            reset_honcho_client()
+            get_honcho_client(hcfg)
+            return (True, "")
+        except ImportError:
+            return (False, "unavailable: honcho-ai not installed")
+        except Exception as exc:
+            msg = str(exc).lower()
+            if "401" in msg or "403" in msg or "unauthorized" in msg or "forbidden" in msg or ("invalid" in msg and "key" in msg):
+                return (False, f"auth: {exc}")
+            if "connect" in msg or "timeout" in msg or "refused" in msg or "unreachable" in msg:
+                return (False, f"unreachable: {exc}")
+            return (False, f"unreachable: {exc}")
+
     def save_config(self, values, hermes_home):
         """Write config to $HERMES_HOME/honcho.json (Honcho SDK native format)."""
         import json

plugins/memory/mem0/__init__.py

@@ -143,6 +143,29 @@ def is_available(self) -> bool:
         cfg = _load_config()
         return bool(cfg.get("api_key"))
 
+
+    def health_check(self) -> tuple[bool, str]:
+        """Probe Mem0 API with a minimal search to verify key + connectivity."""
+        try:
+            from mem0 import MemoryClient
+            cfg = self._config if self._config else _load_config()
+            api_key = cfg.get("api_key", "")
+            if not api_key:
+                return (False, "auth: MEM0_API_KEY not configured")
+            client = MemoryClient(api_key=api_key)
+            user_id = cfg.get("user_id", "health-check-probe")
+            client.get_all(user_id=user_id, limit=1)
+            return (True, "")
+        except ImportError:
+            return (False, "unavailable: mem0ai not installed")
+        except Exception as exc:
+            msg = str(exc).lower()
+            if "401" in msg or "403" in msg or "unauthorized" in msg or ("invalid" in msg and "api" in msg):
+                return (False, f"auth: {exc}")
+            if "connect" in msg or "timeout" in msg or "refused" in msg:
+                return (False, f"unreachable: {exc}")
+            return (False, f"unreachable: {exc}")
+
     def save_config(self, values, hermes_home):
         """Write config to $HERMES_HOME/mem0.json."""
         import json