c13287c915
# Conflicts: # apps/desktop/electron/main.ts # hermes_cli/backup.py # hermes_cli/config.py # hermes_cli/plugin_catalog.py # hermes_cli/plugins_cmd.py # hermes_cli/plugins_cmd_catalog.py # hermes_cli/plugins_discovery.py # hermes_cli/profiles.py # hermes_cli/update_cmd_deps.py # pyproject.toml # tests/gateway/test_dm_topics.py # tests/hermes_cli/test_config.py # tests/hermes_cli/test_plugins_cmd.py # tests/hermes_cli/test_update_autostash.py # tests/tools/test_lazy_deps.py # tools/lazy_deps.py # tools/skill_ledger.py # utils.py # website/docs/user-guide/security.md
387 lines
14 KiB
Python
387 lines
14 KiB
Python
"""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
|
|
|
|
manifest, inputs = memory_provider_dependency_inputs(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.
|
|
"pip_dependencies": list(dict.fromkeys(
|
|
_string_list(manifest.get("pip_dependencies")) + _string_list(manifest.get("python_dependencies")))),
|
|
"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."
|
|
),
|
|
)
|