ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(web_server): simplify config/dashboard/memory helper modules

Browse Source 
- web_server_config: _apply_model_assignment_sync (224 LOC) split into
  main/aux phase helpers (_resolve_assignment_credentials,
  _apply_nous_gateway_defaults, _register_custom_endpoint, _stale_aux_pins,
  _cron_model_impact); _infer_type -> table; schema option lookups unified
  via _schema_select_options; docstrings compacted (every WHY kept).
- web_server_dashboard: mount_spa asset-rewrite loops, shared no-store/
  headless constants; theme normaliser uses _dict_field/_nonempty_str/
  _css_ident predicates; plugin discovery split into search-dirs + entry
  builder; hub auth hint extracted; API-mount gate -> _plugin_api_mount_skip_reason.
- web_server_memory: status ladder -> _memory_provider_status, kind
  inference -> _schema_field_kind, small defensive-layer collapses.
Golden corpus (48 fixtures, old vs new) identical; route table identical.
This commit is contained in:
Teknium
2026-09-02 21:11:37 -07:00
parent 113f04616b
commit 7c113cd426
3 changed files with 680 additions and 1336 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_server_config.py
+307 -614
View File
File diff suppressed because it is too large Load Diff
hermes_cli/web_server_dashboard.py
+298 -588
View File
File diff suppressed because it is too large Load Diff
hermes_cli/web_server_memory.py
+75 -134
View File
@@ -1,8 +1,8 @@
"""Memory-provider dashboard helpers: manifest/schema loading, setup-env and dependency probes, configured-status discovery.
Split out of ``hermes_cli.web_server``; every externally used name is re-imported
there, so ``web_server.<name>`` keeps resolving (and monkeypatching) as before.
Helpers that tests patch on ``web_server`` are reached lazily through it.
Split out of ``hermes_cli.web_server``; every externally used name is re-imported there so
``web_server.<name>`` keeps resolving (and monkeypatching). Helpers that tests patch on
``web_server`` are reached lazily through it.
"""
import logging
@@ -19,12 +19,17 @@ from typing import Any, Dict, List, Optional
# Same logger the code used before extraction (record parity).
_log = logging.getLogger("hermes_cli.web_server")
_MEMORY_PROVIDER_IMPORT_NAMES = {
"honcho-ai": "honcho",
"mem0ai": "mem0",
"hindsight-client": "hindsight_client",
"hindsight-all": "hindsight",
}
def _normalize_memory_provider_name(name: Any) -> str:
provider = str(name or "").strip()
if provider.lower() in {"built-in", "builtin", "none"}:
return ""
return provider
return "" if provider.lower() in {"built-in", "builtin", "none"} else provider
def _load_memory_provider(name: str):
@@ -67,14 +72,9 @@ def _memory_provider_setup_manifest(name: str) -> Dict[str, Any]:
for raw in manifest.get("external_dependencies") or []:
if not isinstance(raw, dict):
continue
dep = {
"name": str(raw.get("name") or "").strip(),
"install": str(raw.get("install") or "").strip(),
"check": str(raw.get("check") or "").strip(),
}
if dep["name"] or dep["install"] or dep["check"]:
dep = {k: str(raw.get(k) or "").strip() for k in ("name", "install", "check")}
if any(dep.values()):
external_dependencies.append(dep)
return {
"pip_dependencies": _string_list(manifest.get("pip_dependencies")),
"external_dependencies": external_dependencies,
@@ -88,14 +88,6 @@ def _memory_provider_setup_info(name: str) -> Dict[str, Any]:
return setup
_MEMORY_PROVIDER_IMPORT_NAMES = {
"honcho-ai": "honcho",
"mem0ai": "mem0",
"hindsight-client": "hindsight_client",
"hindsight-all": "hindsight",
}
def _memory_provider_dependency_package(dep: str) -> str:
return re.split(r"[\[<>=!~;]", dep, maxsplit=1)[0].strip()
@@ -122,25 +114,15 @@ def _memory_provider_setup_env() -> Dict[str, str]:
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"),
]
existing_path = env.get("PATH", "")
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 + existing_path
env["PATH"] = prefix + os.pathsep + env.get("PATH", "")
return env
def _run_setup_command(
command: Any,
*,
display: str,
shell: bool = False,
timeout: int = 180,
command: Any, *, display: str, shell: bool = False, timeout: int = 180
) -> subprocess.CompletedProcess:
return subprocess.run(
command,
@@ -149,8 +131,7 @@ def _run_setup_command(
env=_memory_provider_setup_env(),
capture_output=True,
text=True,
# Lossy UTF-8 decode — setup tools emit UTF-8; never let a
# locale-mismatched byte raise in the reader thread (#52649).
# Lossy UTF-8 decode — setup tools emit UTF-8; a locale-mismatched byte must never raise.
encoding="utf-8",
errors="replace",
timeout=timeout,
@@ -160,35 +141,43 @@ def _run_setup_command(
def _memory_provider_dependencies_installed(setup: Dict[str, Any]) -> bool:
from hermes_cli.web_server import _dependency_importable
pip_dependencies = _string_list(setup.get("pip_dependencies"))
external_dependencies = setup.get("external_dependencies") or []
pip_ok = all(_dependency_importable(dep) for dep in pip_dependencies)
pip_ok = all(_dependency_importable(dep) for dep in _string_list(setup.get("pip_dependencies")))
external_ok = True
for dep in external_dependencies:
for dep in setup.get("external_dependencies") or []:
if not isinstance(dep, dict):
continue
check_cmd = str(dep.get("check") or "").strip()
install_cmd = str(dep.get("install") or "").strip()
if not check_cmd:
if install_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,
)
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 pip_ok and 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"):
@@ -204,42 +193,18 @@ def _normalize_memory_provider_schema(name: str, provider: Any) -> List[Dict[str
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 = []
explicit_kind = str(raw.get("kind") or raw.get("type") or "").strip().lower()
if raw.get("secret"):
kind = "secret"
elif choices:
kind = "select"
elif explicit_kind in {"bool", "boolean"} or isinstance(raw.get("default"), bool):
kind = "boolean"
elif explicit_kind in {"int", "integer"} or (
isinstance(raw.get("default"), int) and not isinstance(raw.get("default"), bool)
):
kind = "integer"
elif explicit_kind in {"float", "number"} or isinstance(raw.get("default"), float):
kind = "number"
else:
kind = "text"
options = []
for choice in choices:
value = str(choice)
options.append({"value": value, "label": value, "description": ""})
description = str(raw.get("description") or "")
fields.append({
"key": key,
"label": str(raw.get("label") or key.replace("_", " ").title()),
"kind": kind,
"description": description,
"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": options,
"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"),
@@ -247,7 +212,6 @@ def _normalize_memory_provider_schema(name: str, provider: Any) -> List[Dict[str
"step": raw.get("step"),
"_env_key": str(raw.get("env_var") or "") or None,
})
return fields
@@ -268,20 +232,17 @@ def _read_memory_provider_existing_values(name: str) -> Dict[str, Any]:
hermes_home = get_hermes_home()
values: Dict[str, Any] = {}
# Common native provider stores.
for path in (
hermes_home / f"{name}.json",
hermes_home / name / "config.json",
):
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(cfg, dict) else {}
memory_cfg = cfg.get("memory")
if isinstance(memory_cfg, dict):
provider_cfg = memory_cfg.get(name)
if isinstance(provider_cfg, dict):
@@ -291,12 +252,11 @@ def _read_memory_provider_existing_values(name: str) -> Dict[str, Any]:
values = {**legacy_cfg, **values}
# Holographic stores under plugins.hermes-memory-store.
plugins_cfg = cfg.get("plugins") if isinstance(cfg, dict) else {}
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
@@ -304,8 +264,7 @@ def _env_lookup(env_key: Optional[str]) -> str:
from hermes_cli.web_server import load_env
if not env_key:
return ""
env_on_disk = load_env()
return str(env_on_disk.get(env_key) or os.environ.get(env_key) or "")
return str(load_env().get(env_key) or os.environ.get(env_key) or "")
def _coerce_bool(value: Any, *, default: bool = False) -> bool:
@@ -333,7 +292,6 @@ def _field_default(field: Dict[str, Any]) -> Any:
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"))
@@ -352,27 +310,20 @@ def _field_value(field: Dict[str, Any], data: Dict[str, Any]) -> Any:
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"]))
value = _field_value(field, data)
return value not in (None, "")
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,
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,
"key": str(dep_key), "kind": "text", "default": "", "_env_key": None
}
actual = _field_value(dep_field, data)
if str(actual) != str(expected):
if str(_field_value(dep_field, data)) != str(expected):
return False
return True
@@ -381,13 +332,21 @@ 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}
visible_fields = [
field for field in fields if _field_visible(field, data, fields_by_key)
]
required_fields = [field for field in visible_fields if field.get("required")]
if not required_fields:
return True
return all(_field_is_set(field, data) for field in required_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 row["available"] and 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]]:
@@ -406,11 +365,8 @@ def _discover_memory_provider_statuses() -> List[Dict[str, Any]]:
except Exception:
_log.exception("discover_memory_providers failed")
cfg = load_config()
active = ""
mem = cfg.get("memory")
if isinstance(mem, dict):
active = _normalize_memory_provider_name(mem.get("provider"))
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,
@@ -422,28 +378,17 @@ def _discover_memory_provider_statuses() -> List[Dict[str, Any]]:
providers: List[Dict[str, Any]] = []
for name in sorted(discovered):
row = discovered[name]
provider = None if row["missing"] else _load_memory_provider(name)
missing = row["missing"]
provider = None if missing else _load_memory_provider(name)
setup = _memory_provider_setup_info(name)
configured = False if row["missing"] else _memory_provider_is_configured(name, provider)
schema_fields = [] if row["missing"] else _normalize_memory_provider_schema(name, provider)
if row["missing"]:
status = "missing"
elif not row["available"] and not setup.get("dependencies_installed", True):
status = "unavailable"
elif not configured:
status = "needs_config"
elif not row["available"] and schema_fields:
status = "needs_config"
elif not row["available"]:
status = "unavailable"
else:
status = "ready"
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": status,
"status": _memory_provider_status(row, setup, configured, schema_fields),
"setup": setup,
})
return providers
@@ -453,13 +398,9 @@ def _require_memory_provider_ready(name: str) -> None:
from hermes_cli.web_server import _discover_memory_provider_statuses
if not name:
return
statuses = {row["name"]: row for row in _discover_memory_provider_statuses()}
row = statuses.get(name)
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}'.",
)
raise HTTPException(status_code=400, detail=f"Unknown memory provider '{name}'.")
if row["status"] != "ready":
raise HTTPException(
status_code=400,