mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-10-20 05:15:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

feat(desktop): schema-driven memory-provider config surface

Browse source 
Make the desktop memory settings dynamic instead of hardcoded per
provider. The dropdown is now populated from discover_memory_providers()
(bundled + user-installed + pip) rather than a static enum, and each
provider's config panel is derived from its own get_config_schema() —
the same declaration `hermes memory setup` uses — so adding or porting a
provider is pure declaration with no bespoke UI, conditional, or
endpoint.

- memory_providers.py: reworked from a hand-written Hindsight registry
  into a pure adapter (describe_provider + coerce_value) that normalizes
  a provider's raw schema into typed fields — secret(+env), select,
  boolean, typed text — carrying `when` conditionals, url, and required.
- MemoryProvider ABC: add optional read_current_config() (default {}),
  the read-back mirror of save_config(). Ported mem0, hindsight, honcho,
  holographic; the rest fall back to schema defaults safely.
- web_server GET/PUT /api/memory/providers/{name}/config now load the
  live provider, derive its schema, write non-secrets via the provider's
  own save_config() (each keeps its native storage), persist secrets to
  the env store, and `when`-gate validation so hidden fields aren't
  required or written. Secrets stay write-only (is_set only).
- ProviderConfigPanel: `when`-conditional visibility (handles Hindsight's
  mode-gated duplicate keys), boolean toggle, and credential url links.
  Dropdown driven by getMemoryStatus(); hardcoded enum removed.

Tests assert the mapping contract and endpoint behavior (schema
derivation, save-via-save_config, secret-never-returned, when-gating,
select rejection) against real bundled providers rather than a snapshot
of a hardcoded list.
This commit is contained in:
Brooklyn Nicholson 2026-06-18 17:16:13 -05:00
parent 03d9a95a74
commit e48d18c8e2
16 changed files with 770 additions and 291 deletions
Download patch file Download diff file Expand all files Collapse all files

285
hermes_cli/memory_providers.py
View file

@ -1,25 +1,42 @@
"""Declarative configuration schema for desktop memory providers.
"""Schema-driven configuration surface for desktop memory providers.
Each memory provider *declares* its configurable surface here — the fields, their
types, which values are secrets, and (for selects) the allowed options. A single
generic renderer in the desktop UI and a single generic ``GET/PUT
/api/memory/providers/{name}/config`` endpoint pair drive the whole experience,
so adding a new provider (mem0, honcho, ...) is pure declaration with zero
bespoke UI components or endpoints.
Memory providers already declare their configurable fields via
``MemoryProvider.get_config_schema()`` (the same declaration ``hermes memory
setup`` walks). This module is the *adapter* that normalizes those raw
declarations into a stable, JSON-serializable shape the desktop config panel
renders generically — no per-provider UI, no hand-maintained registry.
This module is intentionally pure data: it imports nothing from the config/env
layer. ``web_server`` owns the generic read/write logic that interprets these
declarations against config.yaml, the provider config file, and the env store.
Combined with ``discover_memory_providers()`` driving the dropdown, adding or
porting a provider is pure declaration: implement ``get_config_schema()`` (and
optionally ``read_current_config()`` / ``save_config()``) and the provider
shows up, configured, in the desktop UI with zero bespoke code.
This module is intentionally pure transformation: it imports nothing from the
config/env layer and does no I/O. ``web_server`` owns loading the live
provider, reading current values, writing via the provider's ``save_config()``,
and persisting secrets to the env store. Keeping the mapping here (and the I/O
there) is what lets the same normalized schema drive both the HTTP payload and
its validation.
"""
from __future__ import annotations
from dataclasses import dataclass, field as dataclass_field
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
# Field kinds understood by the generic renderer.
KIND_TEXT = "text"
KIND_SELECT = "select"
KIND_SECRET = "secret"
KIND_BOOLEAN = "boolean"
# Native value types, derived from a field's declared default. Submitted form
# values arrive as strings; ``coerce_value`` casts them back to these so a
# provider's config file keeps booleans/numbers rather than stringified ones.
VALUE_STR = "str"
VALUE_BOOL = "bool"
VALUE_INT = "int"
VALUE_FLOAT = "float"
@dataclass(frozen=True)
@ -33,29 +50,33 @@ class ProviderFieldOption:
@dataclass(frozen=True)
class ProviderField:
"""One configurable field on a memory provider.
"""One configurable field, normalized from a provider's schema entry.
A field is stored in exactly one place, decided by ``kind``:
Storage is decided by ``kind``:
* ``text`` / ``select`` — persisted to the provider's JSON config file
(``<hermes_home>/<provider>/config.json``) under ``key``.
* ``text`` / ``select`` / ``boolean`` — persisted by the provider's own
``save_config()`` to its native location, keyed by ``key``.
* ``secret`` — persisted to the env store under ``env_key`` and never read
back out over the API (only an ``is_set`` flag is surfaced).
back over the API (only an ``is_set`` flag is surfaced).
``aliases`` and ``env_fallbacks`` let a field read legacy values written by
earlier CLI/env setup without re-introducing per-provider code.
``when`` carries a provider's conditional-visibility clause (e.g. Hindsight
only shows ``api_url`` ``when`` ``mode == cloud``); the renderer hides
fields whose clause doesn't match the current values, and the same clause
gates server-side validation so hidden fields aren't required.
"""
key: str
label: str
kind: str = KIND_TEXT
value_type: str = VALUE_STR
default: str = ""
description: str = ""
placeholder: str = ""
options: tuple[ProviderFieldOption, ...] = ()
env_key: str | None = None
aliases: tuple[str, ...] = ()
env_fallbacks: tuple[str, ...] = ()
required: bool = False
url: str = ""
options: Tuple[ProviderFieldOption, ...] = ()
env_key: Optional[str] = None
when: Tuple[Tuple[str, str], ...] = ()
@property
def is_secret(self) -> bool:
@ -64,86 +85,166 @@ class ProviderField:
def allowed_values(self) -> set[str]:
return {opt.value for opt in self.options}
def when_matches(self, values: Dict[str, Any]) -> bool:
"""Whether this field is active given the current field values."""
return all(str(values.get(k, "")) == v for k, v in self.when)
@dataclass(frozen=True)
class MemoryProvider:
"""A declared memory provider and its configurable fields."""
"""A declared memory provider and its normalized configurable fields."""
name: str
label: str
fields: tuple[ProviderField, ...] = dataclass_field(default_factory=tuple)
fields: Tuple[ProviderField, ...] = ()
def field(self, key: str) -> Optional[ProviderField]:
for f in self.fields:
if f.key == key:
return f
return None
HINDSIGHT = MemoryProvider(
name="hindsight",
label="Hindsight",
fields=(
ProviderField(
key="mode",
label="Mode",
kind=KIND_SELECT,
default="cloud",
description="How Hermes connects to Hindsight.",
options=(
ProviderFieldOption(
"cloud",
"Cloud",
"Hindsight Cloud API (lightweight, just needs an API key)",
),
ProviderFieldOption(
"local_external",
"Local External",
"Connect to an existing Hindsight instance",
),
),
),
ProviderField(
key="api_key",
label="API key",
kind=KIND_SECRET,
env_key="HINDSIGHT_API_KEY",
description="Used to authenticate with the Hindsight API.",
placeholder="Enter Hindsight API key",
),
ProviderField(
key="api_url",
label="API URL",
kind=KIND_TEXT,
default="https://api.hindsight.vectorize.io",
aliases=("apiUrl",),
env_fallbacks=("HINDSIGHT_API_URL",),
),
ProviderField(
key="bank_id",
label="Bank ID",
kind=KIND_TEXT,
default="hermes",
aliases=("bankId",),
),
ProviderField(
key="recall_budget",
label="Recall budget",
kind=KIND_SELECT,
default="mid",
aliases=("budget",),
options=(
ProviderFieldOption("low", "low"),
ProviderFieldOption("mid", "mid"),
ProviderFieldOption("high", "high"),
),
),
),
)
def _label_from_key(raw: str) -> str:
"""Humanize a snake/kebab key into a field/provider label."""
cleaned = raw.replace("_", " ").replace("-", " ").strip()
if not cleaned:
return raw
# Title-case but leave acronym-ish all-caps tokens intact.
return " ".join(w if w.isupper() else w.capitalize() for w in cleaned.split())
# Registry of providers that expose a desktop config surface. Providers without
# an entry here (e.g. ``builtin``) simply render no config panel.
MEMORY_PROVIDERS: dict[str, MemoryProvider] = {
HINDSIGHT.name: HINDSIGHT,
}
def _value_type_of(default: Any) -> str:
# bool first — bool is a subclass of int.
if isinstance(default, bool):
return VALUE_BOOL
if isinstance(default, int):
return VALUE_INT
if isinstance(default, float):
return VALUE_FLOAT
return VALUE_STR
def get_memory_provider(name: str) -> MemoryProvider | None:
"""Return the declared provider for ``name``, or ``None`` if undeclared."""
def _default_to_str(default: Any) -> str:
if default is None:
return ""
if isinstance(default, bool):
return "true" if default else "false"
return str(default)
return MEMORY_PROVIDERS.get(name)
def _field_from_schema(raw: Dict[str, Any]) -> Optional[ProviderField]:
"""Normalize one ``get_config_schema()`` entry into a ``ProviderField``."""
key = str(raw.get("key") or "").strip()
if not key:
return None
default = raw.get("default")
secret = bool(raw.get("secret"))
choices = raw.get("choices")
env_var = raw.get("env_var")
when_raw = raw.get("when")
when: Tuple[Tuple[str, str], ...] = ()
if isinstance(when_raw, dict):
when = tuple((str(k), str(v)) for k, v in when_raw.items())
if secret:
kind, value_type, options = KIND_SECRET, VALUE_STR, ()
elif choices:
kind = KIND_SELECT
value_type = _value_type_of(default)
options = tuple(
ProviderFieldOption(value=str(c), label=str(c)) for c in choices
)
elif isinstance(default, bool):
kind, value_type, options = KIND_BOOLEAN, VALUE_BOOL, ()
else:
kind, value_type, options = KIND_TEXT, _value_type_of(default), ()
return ProviderField(
key=key,
label=_label_from_key(key),
kind=kind,
value_type=value_type,
default=_default_to_str(default),
description=str(raw.get("description") or ""),
placeholder=str(raw.get("placeholder") or ""),
required=bool(raw.get("required")),
url=str(raw.get("url") or ""),
options=options,
env_key=str(env_var) if env_var else None,
when=when,
)
def describe_provider(
name: str,
schema: List[Dict[str, Any]],
label: Optional[str] = None,
) -> MemoryProvider:
"""Adapt a provider's raw ``get_config_schema()`` into a normalized descriptor.
``schema`` is the list returned by the live provider; ``label`` overrides
the humanized name (e.g. a provider's display name from ``plugin.yaml``).
Unparseable entries are skipped rather than raising, so one malformed field
can't blank out a provider's whole config surface.
"""
fields: List[ProviderField] = []
for raw in schema or []:
if not isinstance(raw, dict):
continue
field = _field_from_schema(raw)
if field is not None:
fields.append(field)
return MemoryProvider(
name=name,
label=label or _label_from_key(name),
fields=tuple(fields),
)
def coerce_value(field: ProviderField, raw: str) -> Any:
"""Validate + cast a submitted string back to the field's native type.
Raises ``ValueError`` for a select value outside its options. Empty input
falls back to the field's declared default. Booleans accept the usual
truthy/falsy spellings; numbers parse leniently and fall back to the
default on a bad parse rather than corrupting the config.
"""
value = (raw or "").strip()
if field.kind == KIND_SELECT:
if not value:
value = field.default
if field.options and value not in field.allowed_values():
raise ValueError(f"Invalid value for '{field.key}'")
return value
if field.value_type == VALUE_BOOL:
if value == "":
value = field.default
return value.strip().lower() in {"1", "true", "yes", "on"}
if field.value_type in (VALUE_INT, VALUE_FLOAT):
if value == "":
value = field.default
try:
return int(value) if field.value_type == VALUE_INT else float(value)
except (TypeError, ValueError):
try:
return (
int(field.default)
if field.value_type == VALUE_INT
else float(field.default)
)
except (TypeError, ValueError):
return 0 if field.value_type == VALUE_INT else 0.0
return value or field.default

196
hermes_cli/web_server.py
View file

@ -65,7 +65,8 @@ from hermes_cli.config import (
from hermes_cli.memory_providers import (
MemoryProvider,
ProviderField,
get_memory_provider,
coerce_value,
describe_provider,
)
from gateway.status import (
get_running_pid,
@ -3172,60 +3173,86 @@ def _normalize_config_for_web(config: Dict[str, Any]) -> Dict[str, Any]:
return config
def _memory_provider_config_path(provider: MemoryProvider) -> Path:
return get_hermes_home() / provider.name / "config.json"
def _describe_memory_provider(name: str) -> Optional[MemoryProvider]:
"""Load a live memory provider and normalize its declared config schema.
Returns ``None`` for providers that aren't discoverable or declare no
configurable fields (e.g. built-in), so the generic panel renders nothing.
"""
def _read_memory_provider_file(provider: MemoryProvider) -> Dict[str, Any]:
path = _memory_provider_config_path(provider)
if not path.exists():
return {}
try:
data = json.loads(path.read_text(encoding="utf-8"))
from plugins.memory import (
discover_memory_providers,
load_memory_provider,
)
except Exception:
_log.warning("Failed to read memory provider config from %s", path, exc_info=True)
return None
try:
label = next(
(
(desc or _label).split(".")[0].strip() or _label
for _label, desc, _avail in [
(n, d, a) for n, d, a in discover_memory_providers() if n == name
]
),
None,
)
except Exception:
label = None
try:
provider = load_memory_provider(name)
except Exception:
provider = None
if provider is None:
return None
try:
schema = provider.get_config_schema() or []
except Exception:
_log.warning("get_config_schema() failed for memory provider %s", name, exc_info=True)
schema = []
descriptor = describe_provider(name, schema, label=label)
if not descriptor.fields:
return None
return descriptor
def _read_provider_current(name: str) -> Dict[str, Any]:
"""Best-effort read of a provider's persisted (non-secret) config values."""
try:
from plugins.memory import load_memory_provider
provider = load_memory_provider(name)
if provider is None:
return {}
data = provider.read_current_config()
return data if isinstance(data, dict) else {}
except Exception:
_log.debug("read_current_config() failed for memory provider %s", name, exc_info=True)
return {}
return data if isinstance(data, dict) else {}
def _read_field_value(field: ProviderField, data: Dict[str, Any]) -> str:
"""Resolve the stored value for a non-secret field, honoring legacy reads."""
for source_key in (field.key, *field.aliases):
value = data.get(source_key)
if value:
return str(value)
def _memory_provider_payload(descriptor: MemoryProvider) -> Dict[str, Any]:
current = _read_provider_current(descriptor.name)
env_on_disk = load_env()
for env_key in field.env_fallbacks:
value = env_on_disk.get(env_key)
if value:
return str(value)
return field.default
def _field_is_set(field: ProviderField, data: Dict[str, Any]) -> bool:
"""Whether a secret field has a value anywhere it may have been written."""
env_on_disk = load_env()
for env_key in (field.env_key, *field.env_fallbacks):
if env_key and env_on_disk.get(env_key):
return True
return any(data.get(source_key) for source_key in (field.key, *field.aliases))
def _memory_provider_payload(provider: MemoryProvider) -> Dict[str, Any]:
data = _read_memory_provider_file(provider)
fields: List[Dict[str, Any]] = []
for field in provider.fields:
for field in descriptor.fields:
entry: Dict[str, Any] = {
"key": field.key,
"label": field.label,
"kind": field.kind,
"value_type": field.value_type,
"default": field.default,
"description": field.description,
"placeholder": field.placeholder,
"required": field.required,
"url": field.url,
"when": [{"key": k, "value": v} for k, v in field.when],
"options": [
{"value": opt.value, "label": opt.label, "description": opt.description}
for opt in field.options
@ -3235,83 +3262,96 @@ def _memory_provider_payload(provider: MemoryProvider) -> Dict[str, Any]:
if field.is_secret:
# Secrets are write-only over the API; only expose whether one is set.
entry["value"] = ""
entry["is_set"] = _field_is_set(field, data)
entry["is_set"] = bool(field.env_key and env_on_disk.get(field.env_key))
elif field.key in current and current[field.key] not in (None, ""):
entry["value"] = _default_to_display(current[field.key], field)
entry["is_set"] = True
else:
value = _read_field_value(field, data)
if field.kind == "select" and value not in field.allowed_values():
value = field.default
entry["value"] = value
entry["is_set"] = bool(value)
entry["value"] = field.default
entry["is_set"] = False
fields.append(entry)
return {"name": provider.name, "label": provider.label, "fields": fields}
return {"name": descriptor.name, "label": descriptor.label, "fields": fields}
def _coerce_field_value(field: ProviderField, raw: str) -> str:
"""Validate and normalize a submitted non-secret value, or raise ValueError."""
def _default_to_display(value: Any, field: ProviderField) -> str:
"""Render a stored native value back to the string form the panel edits."""
value = (raw or "").strip()
if field.kind == "select":
if not value:
value = field.default
if value not in field.allowed_values():
raise ValueError(f"Invalid value for '{field.key}'")
return value
return value or field.default
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
@app.get("/api/memory/providers/{name}/config")
async def get_memory_provider_config(name: str):
provider = get_memory_provider(name)
if provider is None:
# Undeclared providers (e.g. builtin) have no config surface. Return an
# empty schema so the generic panel simply renders nothing.
descriptor = _describe_memory_provider(name)
if descriptor is None:
# Undeclared/built-in providers have no config surface. Return an empty
# schema so the generic panel simply renders nothing.
return {"name": name, "label": name, "fields": []}
return _memory_provider_payload(provider)
return _memory_provider_payload(descriptor)
@app.put("/api/memory/providers/{name}/config")
async def update_memory_provider_config(name: str, body: MemoryProviderConfigUpdate):
provider = get_memory_provider(name)
if provider is None:
descriptor = _describe_memory_provider(name)
if descriptor is None:
raise HTTPException(status_code=404, detail=f"Unknown memory provider: {name}")
values = body.values or {}
try:
existing = _read_memory_provider_file(provider)
json_values: Dict[str, Any] = {}
# Resolve the effective value of every field first so ``when`` clauses
# are evaluated against what the user actually submitted (e.g. the
# selected mode), not stale state.
resolved: Dict[str, str] = {}
for field in descriptor.fields:
if field.key in values:
resolved[field.key] = values[field.key]
elif not field.is_secret:
resolved[field.key] = field.default
config_values: Dict[str, Any] = {}
secrets: Dict[str, str] = {}
for field in provider.fields:
for field in descriptor.fields:
# Skip fields gated off by an unmet ``when`` clause — they aren't
# shown to the user, so don't validate or persist them.
if not field.when_matches(resolved):
continue
if field.is_secret:
submitted = (values.get(field.key) or "").strip()
if submitted and field.env_key:
secrets[field.env_key] = submitted
continue
raw = (
values[field.key]
if field.key in values
else str(existing.get(field.key, field.default))
)
json_values[field.key] = _coerce_field_value(field, raw)
raw = values.get(field.key, field.default)
config_values[field.key] = coerce_value(field, raw)
config = load_config()
memory_config = config.get("memory")
if not isinstance(memory_config, dict):
memory_config = {}
config["memory"] = memory_config
memory_config["provider"] = provider.name
memory_config["provider"] = descriptor.name
save_config(config)
path = _memory_provider_config_path(provider)
path.parent.mkdir(parents=True, exist_ok=True)
existing.update(json_values)
from utils import atomic_json_write
# Persist non-secret values through the provider's own save_config() so
# each provider keeps ownership of its native storage layout (this is
# the same path `hermes memory setup` uses).
if config_values:
try:
from plugins.memory import load_memory_provider
atomic_json_write(path, existing, mode=0o600)
provider = load_memory_provider(descriptor.name)
if provider is not None and hasattr(provider, "save_config"):
provider.save_config(config_values, str(get_hermes_home()))
except Exception:
_log.exception(
"save_config() failed for memory provider %s", descriptor.name
)
for env_key, secret in secrets.items():
save_env_value(env_key, secret)