"""Memory-provider dashboard helpers: manifest/schema loading, setup-env and dependency probes, configured-status discovery. """ import logging import json import os import shlex import subprocess from fastapi import HTTPException from pathlib import Path from typing import Any, Dict, List, Optional # Same logger the code used before extraction (record parity). _log = logging.getLogger("hermes_cli.web_server") def _normalize_memory_provider_name(name: Any) -> str: from agent.memory_provider import is_core_memory_provider provider = str(name or "").strip() return "" if is_core_memory_provider(provider) else provider def _load_memory_provider(name: str): try: from plugins.memory import load_memory_provider return load_memory_provider(name) except Exception: _log.debug("Failed to load memory provider %s", name, exc_info=True) return None def _memory_provider_manifest(name: str) -> Dict[str, Any]: try: from hermes_cli.memory_setup import memory_provider_dependency_inputs manifest, _inputs = memory_provider_dependency_inputs(name) return manifest except Exception: _log.debug("Failed to read memory provider manifest for %s", name, exc_info=True) return {} def _string_list(value: Any) -> List[str]: if not isinstance(value, list): return [] return [str(item).strip() for item in value if str(item).strip()] def _memory_provider_setup_manifest(name: str) -> tuple[dict, dict]: from hermes_cli.memory_setup import memory_provider_dependency_inputs from plugins.memory import find_provider_dir from pm.plugin_declarations import read_python_declaration manifest, inputs = memory_provider_dependency_inputs(name) plugin_dir = find_provider_dir(name) external_dependencies: List[Dict[str, str]] = [] for raw in manifest.get("external_dependencies") or []: if not isinstance(raw, dict): continue dep = {k: str(raw.get(k) or "").strip() for k in ("name", "install", "check")} if any(dep.values()): external_dependencies.append(dep) return { # Display only; neither import names nor these labels determine readiness. Shown from the # same declaration PM installs (an authored pyproject.toml wins over the manifest lists). "pip_dependencies": list(read_python_declaration(plugin_dir).install_requirements) if plugin_dir else [], "python_dependencies_declared": bool(inputs), "external_dependencies": external_dependencies, "required_env": _string_list(manifest.get("requires_env")), }, inputs def _memory_provider_setup_info(name: str) -> Dict[str, Any]: import pm try: setup, inputs = _memory_provider_setup_manifest(name) except Exception: _log.debug("Invalid dependency declaration for %s", name, exc_info=True) return {"pip_dependencies": [], "python_dependencies_declared": True, "external_dependencies": [], "required_env": [], "dependencies_installed": False} try: python_ready = not inputs or pm.venv_is_current(**inputs) except Exception: _log.debug("Could not read dependency state for %s", name, exc_info=True) python_ready = False setup["dependencies_installed"] = python_ready and _memory_provider_external_dependencies_installed(setup) return setup def _memory_provider_setup_env() -> Dict[str, str]: # External package-manager child (npm/uv/pip): exact env preservation — # scrubbing or HOME rewriting could break user tool auth/config. from tools.environments.local import build_subprocess_env env = build_subprocess_env(scrub_secrets=False, inherit_profile_home=False) home = Path.home() extra_bins = [home / ".brv-cli" / "bin", home / ".local" / "bin", home / ".npm-global" / "bin", Path("/usr/local/bin")] prefix = os.pathsep.join(str(path) for path in extra_bins if path.exists()) if prefix: env["PATH"] = prefix + os.pathsep + env.get("PATH", "") return env def _run_setup_command( command: Any, *, display: str, shell: bool = False, timeout: int = 180 ) -> subprocess.CompletedProcess: return subprocess.run( command, shell=shell, executable="/bin/bash" if shell else None, env=_memory_provider_setup_env(), capture_output=True, text=True, # Lossy UTF-8 decode — setup tools emit UTF-8; a locale-mismatched byte must never raise. # Force UTF-8 with lossy decoding so child output containing bytes that are invalid in the system # locale (e.g. GBK on Chinese Windows) can't raise UnicodeDecodeError inside the drain threads and # crash the gateway. See #53137. encoding="utf-8", errors="replace", timeout=timeout, check=False, ) def _memory_provider_external_dependencies_installed(setup: Dict[str, Any]) -> bool: external_ok = True for dep in setup.get("external_dependencies") or []: if not isinstance(dep, dict): continue check_cmd = str(dep.get("check") or "").strip() if not check_cmd: if str(dep.get("install") or "").strip(): external_ok = False continue try: completed = _run_setup_command(shlex.split(check_cmd), display=check_cmd, timeout=20) except Exception: external_ok = False continue if completed.returncode != 0: external_ok = False return external_ok def _schema_field_kind(raw: Dict[str, Any], choices: list) -> str: """Field kind from explicit ``kind``/``type`` hints, else inferred from ``default``.""" explicit_kind = str(raw.get("kind") or raw.get("type") or "").strip().lower() default = raw.get("default") if raw.get("secret"): return "secret" if choices: return "select" if explicit_kind in {"bool", "boolean"} or isinstance(default, bool): return "boolean" if explicit_kind in {"int", "integer"} or (isinstance(default, int) and not isinstance(default, bool)): return "integer" if explicit_kind in {"float", "number"} or isinstance(default, float): return "number" return "text" def _normalize_memory_provider_schema(name: str, provider: Any) -> List[Dict[str, Any]]: raw_schema: List[Dict[str, Any]] = [] if provider is not None and hasattr(provider, "get_config_schema"): try: raw = provider.get_config_schema() if isinstance(raw, list): raw_schema = [field for field in raw if isinstance(field, dict)] except Exception: _log.warning("Failed to read memory provider schema for %s", name, exc_info=True) fields: List[Dict[str, Any]] = [] for raw in raw_schema: key = str(raw.get("key") or "").strip() if not key: continue choices = raw.get("choices") or raw.get("options") or [] if not isinstance(choices, list): choices = [] fields.append({ "key": key, "label": str(raw.get("label") or key.replace("_", " ").title()), "kind": _schema_field_kind(raw, choices), "description": str(raw.get("description") or ""), "placeholder": str(raw.get("placeholder") or ""), "required": bool(raw.get("required", False)), "default": raw.get("default", ""), "options": [{"value": str(c), "label": str(c), "description": ""} for c in choices], "url": str(raw.get("url") or ""), "when": raw.get("when") if isinstance(raw.get("when"), dict) else None, "minimum": raw.get("minimum"), "maximum": raw.get("maximum"), "step": raw.get("step"), "_env_key": str(raw.get("env_var") or "") or None, }) return fields def _read_json_file(path: Path) -> Dict[str, Any]: try: data = json.loads(path.read_text(encoding="utf-8-sig")) if path.exists() else {} except Exception: _log.debug("Failed to read JSON config from %s", path, exc_info=True) return {} return data if isinstance(data, dict) else {} def _read_memory_provider_existing_values(name: str) -> Dict[str, Any]: """Best-effort read of existing provider config across legacy/native stores.""" from hermes_cli.config import get_hermes_home, load_config hermes_home = get_hermes_home() values: Dict[str, Any] = {} for path in (hermes_home / f"{name}.json", hermes_home / name / "config.json"): values.update(_read_json_file(path)) try: cfg = load_config() except Exception: cfg = {} if not isinstance(cfg, dict): cfg = {} memory_cfg = cfg.get("memory") if isinstance(memory_cfg, dict): provider_cfg = memory_cfg.get(name) if isinstance(provider_cfg, dict): values.update(provider_cfg) legacy_cfg = memory_cfg.get("provider_config") if isinstance(legacy_cfg, dict): values = {**legacy_cfg, **values} # Holographic stores under plugins.hermes-memory-store. plugins_cfg = cfg.get("plugins") if name == "holographic" and isinstance(plugins_cfg, dict): holographic_cfg = plugins_cfg.get("hermes-memory-store") if isinstance(holographic_cfg, dict): values.update(holographic_cfg) return values def _env_lookup(env_key: Optional[str]) -> str: from hermes_cli.config import load_env if not env_key: return "" return str(load_env().get(env_key) or os.environ.get(env_key) or "") def _coerce_bool(value: Any, *, default: bool = False) -> bool: if isinstance(value, bool): return value if value is None or value == "": return default if isinstance(value, (int, float)): return bool(value) text = str(value).strip().lower() if text in {"1", "true", "yes", "on"}: return True if text in {"0", "false", "no", "off"}: return False raise ValueError(f"Invalid boolean value: {value}") def _field_default(field: Dict[str, Any]) -> Any: default = field.get("default", "") if field["kind"] == "boolean": return _coerce_bool(default, default=False) return default def _field_value(field: Dict[str, Any], data: Dict[str, Any]) -> Any: if field["kind"] == "secret": return "" value = data.get(field["key"]) if value in (None, ""): value = _env_lookup(field.get("_env_key")) if value in (None, ""): value = _field_default(field) if field["kind"] == "select": allowed = {opt["value"] for opt in field.get("options", [])} value = str(value) return value if value in allowed else str(_field_default(field)) if field["kind"] == "boolean": return _coerce_bool(value, default=_coerce_bool(_field_default(field), default=False)) return str(value) def _field_is_set(field: Dict[str, Any], data: Dict[str, Any]) -> bool: if field["kind"] == "secret": return bool(_env_lookup(field.get("_env_key")) or data.get(field["key"])) return _field_value(field, data) not in (None, "") def _field_visible( field: Dict[str, Any], data: Dict[str, Any], fields_by_key: Optional[Dict[str, Dict[str, Any]]] = None ) -> bool: when = field.get("when") if not isinstance(when, dict) or not when: return True for dep_key, expected in when.items(): dep_field = (fields_by_key or {}).get(str(dep_key)) or { "key": str(dep_key), "kind": "text", "default": "", "_env_key": None } if str(_field_value(dep_field, data)) != str(expected): return False return True def _memory_provider_is_configured(name: str, provider: Any) -> bool: data = _read_memory_provider_existing_values(name) fields = _normalize_memory_provider_schema(name, provider) fields_by_key = {field["key"]: field for field in fields} return all( _field_is_set(field, data) for field in fields if field.get("required") and _field_visible(field, data, fields_by_key) ) def _memory_provider_status(row: Dict[str, Any], setup: Dict[str, Any], configured: bool, schema_fields: list) -> str: if row["missing"]: return "missing" if not setup.get("dependencies_installed", True): return "unavailable" if not configured or (not row["available"] and schema_fields): return "needs_config" return "ready" if row["available"] else "unavailable" def _discover_memory_provider_statuses() -> List[Dict[str, Any]]: from hermes_cli.config import load_config discovered: Dict[str, Dict[str, Any]] = {} try: from plugins.memory import discover_memory_providers for name, description, available in discover_memory_providers(): discovered[str(name)] = { "name": str(name), "description": str(description or ""), "available": bool(available), "missing": False, } except Exception: _log.exception("discover_memory_providers failed") mem = load_config().get("memory") active = _normalize_memory_provider_name(mem.get("provider")) if isinstance(mem, dict) else "" if active and active not in discovered: discovered[active] = { "name": active, "description": "Configured provider was not found.", "available": False, "missing": True, } providers: List[Dict[str, Any]] = [] for name in sorted(discovered): row = discovered[name] missing = row["missing"] provider = None if missing else _load_memory_provider(name) setup = _memory_provider_setup_info(name) configured = False if missing else _memory_provider_is_configured(name, provider) schema_fields = [] if missing else _normalize_memory_provider_schema(name, provider) providers.append({ "name": name, "description": row["description"], "available": row["available"], "configured": configured, "status": _memory_provider_status(row, setup, configured, schema_fields), "setup": setup, }) return providers def _require_memory_provider_ready(name: str) -> None: if not name: return row = next((r for r in _discover_memory_provider_statuses() if r["name"] == name), None) if row is None: raise HTTPException(status_code=400, detail=f"Unknown memory provider '{name}'.") if row["status"] != "ready": raise HTTPException( status_code=400, detail=( f"Memory provider '{name}' is not ready " f"({row['status'].replace('_', ' ')}). Configure it in the dashboard first." ), )