Skip to content

memory: holographic health_check + collapse doctor branches (closes #42) #52

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PowerCreek merged 1 commit into from
May 23, 2026
Merged

memory: holographic health_check + collapse doctor branches (closes #42) #52

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
219 changes: 91 additions & 128 deletions hermes_cli/doctor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2235,143 +2235,106 @@ def _gh_authenticated() -> bool:

if not _active_memory_provider:
check_ok("Built-in memory active", "(no external provider configured — this is fine)")
elif _active_memory_provider == "honcho":
else:
# Unified memory-provider probe (#42). Every shipped provider
# now implements `health_check() -> (bool, reason)` with the
# RFC #42 reason-prefix taxonomy. Doctor dispatches on the
# prefix to choose check_ok / _fail_and_issue / check_warn.
try:
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):
from plugins.memory import load_memory_provider
_provider = load_memory_provider(_active_memory_provider)
except Exception as _e:
_provider = None
check_warn(f"{_active_memory_provider} plugin not loadable",
str(_e))
_provider = None

if _provider is None and _active_memory_provider:
# load_memory_provider returned None — plugin not on disk.
# Skip the dispatch below (the except-branch above already
# surfaced a warn if the load itself raised).
pass
elif _provider is None:
check_warn(f"{_active_memory_provider} plugin not found",
"run: hermes memory setup")
else:
try:
_healthy, _reason = _provider.health_check()
except Exception as _e: # noqa: BLE001
# RFC #42 contract says health_check MUST NOT raise.
# A propagating exception is a provider bug — surface
# it as a warn and keep going.
_healthy, _reason = False, f"health_check_raised: {_e}"

_pname = _active_memory_provider
_prefix, _, _detail = _reason.partition(":")
_detail = _detail.strip()
if _healthy:
check_ok(f"{_pname} reachable")
elif _reason == "no_api_key" or _reason == "no_credentials":
_fail_and_issue(
"Honcho API key or base URL not set",
f"{_pname} not configured",
"run: hermes memory setup",
"No Honcho API key — run 'hermes memory setup'",
f"{_pname} is set as memory provider but credentials are missing",
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)
except ImportError:
_fail_and_issue(
"honcho-ai not installed",
"pip install honcho-ai",
"Honcho is set as memory provider but honcho-ai is not installed",
issues,
)
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", "")
mem0_user = mem0_cfg.get("user_id", "hermes-user")
mem0_agent = mem0_cfg.get("agent_id", "hermes")
if not mem0_key:
elif _reason == "no_url" or _reason == "no_endpoint":
_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",
f"{_pname} URL not set",
"set the provider's *_BASE_URL / *_API_URL / "
"*_ENDPOINT env var, or run: hermes memory setup",
f"{_pname} URL not configured",
issues,
)
elif _reason == "no_config":
check_warn(f"{_pname} config not found",
"run: hermes memory setup")
elif _reason == "disabled":
check_info(f"{_pname} disabled in config")
elif _reason == "sdk_missing" or _prefix == "sdk_missing":
_fail_and_issue(
f"{_pname} SDK not installed",
"see the provider's plugin docs for the correct "
"`pip install` package",
f"{_pname} SDK missing — see plugin docs",
issues,
)
elif _prefix == "auth":
_fail_and_issue(
f"{_pname} auth rejected",
_detail[:200],
f"{_pname} authentication rejected — verify the "
"configured API key / token",
issues,
)
elif _prefix == "not_found":
_fail_and_issue(
f"{_pname} endpoint not found",
_detail[:200],
f"{_pname} returned 404 — verify the configured base URL",
issues,
)
elif _prefix == "http":
check_warn(f"{_pname} unexpected HTTP status",
_detail[:200])
elif _prefix == "unreachable":
check_warn(f"{_pname} unreachable",
f"{_detail[:200]} "
"(network down, server restarting, or "
"transient — credentials may still be valid)")
elif _prefix == "config_error":
check_warn(f"{_pname} config error",
_detail[:200])
elif _prefix == "health_check_raised":
check_warn(f"{_pname} health_check raised",
f"{_detail[:200]} (RFC #42 contract: "
"health_check must not raise; "
"treat as provider bug)")
else:
# Probe with a minimal API call so a wrong / expired key
# surfaces here instead of failing at first-request time
# in production (see #34). Honcho already does this via
# get_honcho_client(); Mem0 didn't.
try:
from mem0 import MemoryClient
_mem0_client = MemoryClient(api_key=mem0_key)
_mem0_client.get_all(user_id=mem0_user, limit=1)
except ImportError:
_fail_and_issue(
"mem0ai not installed",
"pip install mem0ai",
"Mem0 is set as memory provider but mem0ai SDK is not installed",
issues,
)
except Exception as _mp_exc: # noqa: BLE001
msg = str(_mp_exc)
msg_lc = msg.lower()
is_auth = (
"401" in msg
or "403" in msg
or "unauthorized" in msg_lc
or "forbidden" in msg_lc
or "invalid api key" in msg_lc
or "authentication" in msg_lc
)
if is_auth:
_fail_and_issue(
"Mem0 auth rejected",
msg[:200],
"Mem0 API key rejected — verify MEM0_API_KEY "
"at https://app.mem0.ai",
issues,
)
else:
check_warn(
"Mem0 probe failed",
f"{msg[:200]} "
"(network down, SDK changed, or transient — "
"the key itself may still be valid)",
)
else:
check_ok(
"Mem0 connected",
f"user_id={mem0_user} agent_id={mem0_agent}",
)
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,
# supermemory, hindsight, byterover, retaindb, holographic).
# Each provider's `is_available()` only checks env-vars / CLI
# presence — it does NOT probe backend reachability (#36).
# Until a `health_check()` ABC method lands, the most we can
# truthfully say is "config detected" — not "active".
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 configured")
check_info(
"is_available() reports True (env vars / CLI "
"present) — backend reachability not probed by "
"doctor. Honcho + Mem0 do real probes; other "
"providers wait on a `health_check()` ABC method. "
"See TechDevGroup/hermes-agent#36.")
elif _provider:
check_warn(
f"{_active_memory_provider} configured but not available",
"run: hermes memory status")
else:
check_warn(
f"{_active_memory_provider} plugin not found",
"run: hermes memory setup")
except Exception as _e:
check_warn(
f"{_active_memory_provider} check failed", str(_e))
# Unknown reason — surface verbatim. Better than
# swallowing.
check_warn(f"{_pname} health check failed",
_reason[:200])

_check_devagentic_graph()
_check_cron_scheduler()
Expand Down
42 changes: 42 additions & 0 deletions plugins/memory/holographic/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,48 @@ def name(self) -> str:
def is_available(self) -> bool:
return True # SQLite is always available, numpy is optional

def health_check(self) -> tuple[bool, str]:
"""Probe holographic by verifying the db_path's parent
directory is writable. There's no remote service to ping,
but a read-only HERMES_HOME (RO mount, wrong perms) would
still break the provider at runtime — so the probe is
meaningful.

Returns (RFC #42 conventions):
(True, "") — parent writable.
(False, "unreachable: <msg>") — mkdir failed or not writable.
(False, "config_error: <msg>") — config load itself raised.

MUST NOT raise.
"""
from pathlib import Path as _Path
import os as _os
try:
from hermes_constants import get_hermes_home
hermes_home = str(get_hermes_home())
except Exception as exc: # noqa: BLE001
return (False, f"config_error: {exc}")

try:
db_path = str(self._config.get(
"db_path", f"{hermes_home}/memory_store.db"))
db_path = db_path.replace("$HERMES_HOME", hermes_home)
db_path = db_path.replace("${HERMES_HOME}", hermes_home)
parent = _Path(db_path).parent
try:
parent.mkdir(parents=True, exist_ok=True)
except OSError as exc:
return (False,
f"unreachable: db_path parent {parent} "
f"unwritable ({exc})")
if not _os.access(parent, _os.W_OK):
return (False,
f"unreachable: db_path parent {parent} "
"not writable (check filesystem permissions)")
except Exception as exc: # noqa: BLE001
return (False, f"config_error: {exc}")
return (True, "")

def save_config(self, values, hermes_home):
"""Write config to config.yaml under plugins.hermes-memory-store."""
from pathlib import Path
Expand Down
62 changes: 50 additions & 12 deletions tests/hermes_cli/test_doctor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -418,31 +418,69 @@ def test_mem0_provider_not_installed_shows_fail(self, monkeypatch, tmp_path):
assert "Memory Provider" in out
assert "Built-in memory active" not in out

def test_generic_provider_wording_no_longer_implies_reachability(
def test_generic_provider_dispatches_to_health_check(
self, monkeypatch, tmp_path):
"""#36: generic providers (e.g. openviking) only check env-var
presence in is_available(). Doctor must not say "active"
(implies reachable); must say "configured" + flag the gap."""
# Stub load_memory_provider to return a SimpleNamespace whose
# is_available() returns True without any network call —
# mirrors the production openviking/supermemory/etc. shape.
"""#42 step 3: doctor's unified memory-provider dispatch
now calls health_check() on every provider — no per-provider
elif blocks. A provider returning (True, "") shows up as
"<name> reachable" in the output."""
from types import SimpleNamespace

fake = SimpleNamespace(
is_available=lambda: True,
health_check=lambda: (True, ""),
name=lambda: "openviking",
)
import plugins.memory as _mem_pkg
monkeypatch.setattr(_mem_pkg, "load_memory_provider",
lambda name: fake)
out = self._run_doctor_and_capture(monkeypatch, tmp_path,
provider="openviking")
assert "openviking provider configured" in out
assert "openviking reachable" in out
# The pre-#42-step-3 wording is gone.
assert "openviking provider active" not in out
# The info row must point at the open issue so operators can
# follow the design discussion if they care.
assert "#36" in out
assert "backend reachability not probed" in out
assert "openviking provider configured" not in out
assert "backend reachability not probed" not in out

def test_generic_provider_dispatches_auth_failure_as_fail(
self, monkeypatch, tmp_path):
"""#42 step 3: when a provider returns a "auth:" reason,
doctor maps it to _fail_and_issue (red row + issue list
entry), not check_warn."""
from types import SimpleNamespace

fake = SimpleNamespace(
is_available=lambda: True,
health_check=lambda: (False, "auth: 401 invalid key"),
name=lambda: "mem0",
)
import plugins.memory as _mem_pkg
monkeypatch.setattr(_mem_pkg, "load_memory_provider",
lambda name: fake)
out = self._run_doctor_and_capture(monkeypatch, tmp_path,
provider="mem0")
assert "mem0 auth rejected" in out
assert "401 invalid key" in out

def test_generic_provider_dispatches_unreachable_as_warn(
self, monkeypatch, tmp_path):
"""unreachable: prefix → check_warn (transient, key may
still be valid)."""
from types import SimpleNamespace

fake = SimpleNamespace(
is_available=lambda: True,
health_check=lambda: (False, "unreachable: connection refused"),
name=lambda: "honcho",
)
import plugins.memory as _mem_pkg
monkeypatch.setattr(_mem_pkg, "load_memory_provider",
lambda name: fake)
out = self._run_doctor_and_capture(monkeypatch, tmp_path,
provider="honcho")
assert "honcho unreachable" in out
assert "connection refused" in out
assert "credentials may still be valid" in out


def test_run_doctor_termux_treats_docker_and_browser_warnings_as_expected(monkeypatch, tmp_path):
Expand Down
Loading