1  """hermes memory setup|status — configure memory provider plugins.
  2  
  3  Auto-detects installed memory providers via the plugin system.
  4  Interactive curses-based UI for provider selection, then walks through
  5  the provider's config schema. Writes config to config.yaml + .env.
  6  """
  7  
  8  from __future__ import annotations
  9  
 10  import getpass
 11  import os
 12  import sys
 13  from pathlib import Path
 14  
 15  from hermes_constants import get_hermes_home
 16  
 17  
 18  # ---------------------------------------------------------------------------
 19  # Curses-based interactive picker (same pattern as hermes tools)
 20  # ---------------------------------------------------------------------------
 21  
 22  def _curses_select(title: str, items: list[tuple[str, str]], default: int = 0) -> int:
 23      """Interactive single-select with arrow keys.
 24  
 25      items: list of (label, description) tuples.
 26      Returns selected index, or default on escape/quit.
 27      """
 28      from hermes_cli.curses_ui import curses_radiolist
 29      # Format (label, desc) tuples into display strings
 30      display_items = [
 31          f"{label}  {desc}" if desc else label
 32          for label, desc in items
 33      ]
 34      return curses_radiolist(title, display_items, selected=default, cancel_returns=default)
 35  
 36  
 37  def _prompt(label: str, default: str | None = None, secret: bool = False) -> str:
 38      """Prompt for a value with optional default and secret masking."""
 39      suffix = f" [{default}]" if default else ""
 40      if secret:
 41          sys.stdout.write(f"  {label}{suffix}: ")
 42          sys.stdout.flush()
 43          if sys.stdin.isatty():
 44              val = getpass.getpass(prompt="")
 45          else:
 46              val = sys.stdin.readline().strip()
 47      else:
 48          sys.stdout.write(f"  {label}{suffix}: ")
 49          sys.stdout.flush()
 50          val = sys.stdin.readline().strip()
 51      return val or (default or "")
 52  
 53  
 54  # ---------------------------------------------------------------------------
 55  # Provider discovery
 56  # ---------------------------------------------------------------------------
 57  
 58  def _install_dependencies(provider_name: str) -> None:
 59      """Install pip dependencies declared in plugin.yaml."""
 60      import subprocess
 61      from plugins.memory import find_provider_dir
 62  
 63      plugin_dir = find_provider_dir(provider_name)
 64      if not plugin_dir:
 65          return
 66      yaml_path = plugin_dir / "plugin.yaml"
 67      if not yaml_path.exists():
 68          return
 69  
 70      try:
 71          import yaml
 72          with open(yaml_path) as f:
 73              meta = yaml.safe_load(f) or {}
 74      except Exception:
 75          return
 76  
 77      pip_deps = meta.get("pip_dependencies", [])
 78      if not pip_deps:
 79          return
 80  
 81      # pip name → import name mapping for packages where they differ
 82      _IMPORT_NAMES = {
 83          "honcho-ai": "honcho",
 84          "mem0ai": "mem0",
 85          "hindsight-client": "hindsight_client",
 86          "hindsight-all": "hindsight",
 87      }
 88  
 89      # Check which packages are missing
 90      missing = []
 91      for dep in pip_deps:
 92          import_name = _IMPORT_NAMES.get(dep, dep.replace("-", "_").split("[")[0])
 93          try:
 94              __import__(import_name)
 95          except ImportError:
 96              missing.append(dep)
 97  
 98      if not missing:
 99          return
100  
101      print(f"\n  Installing dependencies: {', '.join(missing)}")
102  
103      import shutil
104      uv_path = shutil.which("uv")
105      if not uv_path:
106          print(f"  ⚠ uv not found — cannot install dependencies")
107          print(f"  Install uv: curl -LsSf https://astral.sh/uv/install.sh | sh")
108          print(f"  Then re-run: hermes memory setup")
109          return
110  
111      try:
112          subprocess.run(
113              [uv_path, "pip", "install", "--python", sys.executable, "--quiet"] + missing,
114              check=True, timeout=120,
115              capture_output=True,
116          )
117          print(f"  ✓ Installed {', '.join(missing)}")
118      except subprocess.CalledProcessError as e:
119          print(f"  ⚠ Failed to install {', '.join(missing)}")
120          stderr = (e.stderr or b"").decode()[:200]
121          if stderr:
122              print(f"    {stderr}")
123          print(f"  Run manually: uv pip install --python {sys.executable} {' '.join(missing)}")
124      except Exception as e:
125          print(f"  ⚠ Install failed: {e}")
126          print(f"  Run manually: uv pip install --python {sys.executable} {' '.join(missing)}")
127  
128      # Also show external dependencies (non-pip) if any
129      ext_deps = meta.get("external_dependencies", [])
130      for dep in ext_deps:
131          dep_name = dep.get("name", "")
132          check_cmd = dep.get("check", "")
133          install_cmd = dep.get("install", "")
134          if check_cmd:
135              try:
136                  subprocess.run(
137                      check_cmd, shell=True, capture_output=True, timeout=5
138                  )
139              except Exception:
140                  if install_cmd:
141                      print(f"\n  ⚠ '{dep_name}' not found. Install with:")
142                      print(f"    {install_cmd}")
143  
144  
145  def _get_available_providers() -> list:
146      """Discover memory providers from plugins/memory/.
147  
148      Returns list of (name, description, provider_instance) tuples.
149      """
150      try:
151          from plugins.memory import discover_memory_providers, load_memory_provider
152          raw = discover_memory_providers()
153      except Exception:
154          raw = []
155  
156      results = []
157      for name, desc, available in raw:
158          try:
159              provider = load_memory_provider(name)
160              if not provider:
161                  continue
162          except Exception:
163              continue
164  
165          schema = provider.get_config_schema() if hasattr(provider, "get_config_schema") else []
166          has_secrets = any(f.get("secret") for f in schema)
167          has_non_secrets = any(not f.get("secret") for f in schema)
168          if has_secrets and has_non_secrets:
169              setup_hint = "API key / local"
170          elif has_secrets:
171              setup_hint = "requires API key"
172          elif not schema:
173              setup_hint = "no setup needed"
174          else:
175              setup_hint = "local"
176  
177          results.append((name, setup_hint, provider))
178      return results
179  
180  
181  # ---------------------------------------------------------------------------
182  # Setup wizard
183  # ---------------------------------------------------------------------------
184  
185  def cmd_setup_provider(provider_name: str) -> None:
186      """Run memory setup for a specific provider, skipping the picker."""
187      from hermes_cli.config import load_config, save_config
188  
189      providers = _get_available_providers()
190      match = None
191      for name, desc, provider in providers:
192          if name == provider_name:
193              match = (name, desc, provider)
194              break
195  
196      if not match:
197          print(f"\n  Memory provider '{provider_name}' not found.")
198          print("  Run 'hermes memory setup' to see available providers.\n")
199          return
200  
201      name, _, provider = match
202  
203      _install_dependencies(name)
204  
205      config = load_config()
206      if not isinstance(config.get("memory"), dict):
207          config["memory"] = {}
208  
209      if hasattr(provider, "post_setup"):
210          hermes_home = str(get_hermes_home())
211          provider.post_setup(hermes_home, config)
212          return
213  
214      # Fallback: generic schema-based setup (same as cmd_setup)
215      config["memory"]["provider"] = name
216      save_config(config)
217      print(f"\n  Memory provider: {name}")
218      print(f"  Activation saved to config.yaml\n")
219  
220  
221  def cmd_setup(args) -> None:
222      """Interactive memory provider setup wizard."""
223      from hermes_cli.config import load_config, save_config
224  
225      providers = _get_available_providers()
226  
227      if not providers:
228          print("\n  No memory provider plugins detected.")
229          print("  Install a plugin to ~/.hermes/plugins/ and try again.\n")
230          return
231  
232      # Build picker items
233      items = []
234      for name, desc, _ in providers:
235          items.append((name, f"— {desc}"))
236      items.append(("Built-in only", "— MEMORY.md / USER.md (default)"))
237  
238      builtin_idx = len(items) - 1
239      selected = _curses_select("Memory provider setup", items, default=builtin_idx)
240  
241      config = load_config()
242      if not isinstance(config.get("memory"), dict):
243          config["memory"] = {}
244  
245      # Built-in only
246      if selected >= len(providers) or selected < 0:
247          config["memory"]["provider"] = ""
248          save_config(config)
249          print("\n  ✓ Memory provider: built-in only")
250          print("  Saved to config.yaml\n")
251          return
252  
253      name, _, provider = providers[selected]
254  
255      # Install pip dependencies if declared in plugin.yaml
256      _install_dependencies(name)
257  
258      # If the provider has a post_setup hook, delegate entirely to it.
259      # The hook handles its own config, connection test, and activation.
260      if hasattr(provider, "post_setup"):
261          hermes_home = str(get_hermes_home())
262          provider.post_setup(hermes_home, config)
263          return
264  
265      schema = provider.get_config_schema() if hasattr(provider, "get_config_schema") else []
266  
267      provider_config = config["memory"].get(name, {})
268      if not isinstance(provider_config, dict):
269          provider_config = {}
270  
271      env_path = get_hermes_home() / ".env"
272      env_writes = {}
273  
274      if schema:
275          print(f"\n  Configuring {name}:\n")
276  
277          for field in schema:
278              key = field["key"]
279              desc = field.get("description", key)
280              default = field.get("default")
281              # Dynamic default: look up default from another field's value
282              default_from = field.get("default_from")
283              if default_from and isinstance(default_from, dict):
284                  ref_field = default_from.get("field", "")
285                  ref_map = default_from.get("map", {})
286                  ref_value = provider_config.get(ref_field, "")
287                  if ref_value and ref_value in ref_map:
288                      default = ref_map[ref_value]
289              is_secret = field.get("secret", False)
290              choices = field.get("choices")
291              env_var = field.get("env_var")
292              url = field.get("url")
293  
294              # Skip fields whose "when" condition doesn't match
295              when = field.get("when")
296              if when and isinstance(when, dict):
297                  if not all(provider_config.get(k) == v for k, v in when.items()):
298                      continue
299  
300              if choices and not is_secret:
301                  # Use curses picker for choice fields
302                  choice_items = [(c, "") for c in choices]
303                  current = provider_config.get(key, default)
304                  current_idx = 0
305                  if current and current in choices:
306                      current_idx = choices.index(current)
307                  sel = _curses_select(f"  {desc}", choice_items, default=current_idx)
308                  provider_config[key] = choices[sel]
309              elif is_secret:
310                  # Prompt for secret
311                  existing = os.environ.get(env_var, "") if env_var else ""
312                  if existing:
313                      masked = f"...{existing[-4:]}" if len(existing) > 4 else "set"
314                      val = _prompt(f"{desc} (current: {masked}, blank to keep)", secret=True)
315                  else:
316                      hint = f"  Get yours at {url}" if url else ""
317                      if hint:
318                          print(hint)
319                      val = _prompt(desc, secret=True)
320                  if val and env_var:
321                      env_writes[env_var] = val
322              else:
323                  # Regular text prompt
324                  current = provider_config.get(key)
325                  effective_default = current or default
326                  val = _prompt(desc, default=str(effective_default) if effective_default else None)
327                  if val:
328                      provider_config[key] = val
329                      # Also write to .env if this field has an env_var
330                      if env_var and env_var not in env_writes:
331                          env_writes[env_var] = val
332  
333      # Write activation key to config.yaml
334      config["memory"]["provider"] = name
335      save_config(config)
336  
337      # Write non-secret config to provider's native location
338      hermes_home = str(get_hermes_home())
339      if provider_config and hasattr(provider, "save_config"):
340          try:
341              provider.save_config(provider_config, hermes_home)
342          except Exception as e:
343              print(f"  Failed to write provider config: {e}")
344  
345      # Write secrets to .env
346      if env_writes:
347          _write_env_vars(env_path, env_writes)
348  
349      print(f"\n  Memory provider: {name}")
350      print(f"  Activation saved to config.yaml")
351      if provider_config:
352          print(f"  Provider config saved")
353      if env_writes:
354          print(f"  API keys saved to .env")
355      print(f"\n  Start a new session to activate.\n")
356  
357  
358  def _write_env_vars(env_path: Path, env_writes: dict) -> None:
359      """Append or update env vars in .env file."""
360      env_path.parent.mkdir(parents=True, exist_ok=True)
361  
362      existing_lines = []
363      if env_path.exists():
364          existing_lines = env_path.read_text(encoding="utf-8").splitlines()
365  
366      updated_keys = set()
367      new_lines = []
368      for line in existing_lines:
369          key_match = line.split("=", 1)[0].strip() if "=" in line else ""
370          if key_match in env_writes:
371              new_lines.append(f"{key_match}={env_writes[key_match]}")
372              updated_keys.add(key_match)
373          else:
374              new_lines.append(line)
375  
376      for key, val in env_writes.items():
377          if key not in updated_keys:
378              new_lines.append(f"{key}={val}")
379  
380      env_path.write_text("\n".join(new_lines) + "\n")
381  
382  
383  # ---------------------------------------------------------------------------
384  # Status
385  # ---------------------------------------------------------------------------
386  
387  def cmd_status(args) -> None:
388      """Show current memory provider config."""
389      from hermes_cli.config import load_config
390  
391      config = load_config()
392      mem_config = config.get("memory", {})
393      provider_name = mem_config.get("provider", "")
394  
395      print(f"\nMemory status\n" + "─" * 40)
396      print(f"  Built-in:  always active")
397      print(f"  Provider:  {provider_name or '(none — built-in only)'}")
398  
399      if provider_name:
400          provider_config = mem_config.get(provider_name, {})
401          if provider_config:
402              print(f"\n  {provider_name} config:")
403              for key, val in provider_config.items():
404                  print(f"    {key}: {val}")
405  
406          providers = _get_available_providers()
407          found = any(name == provider_name for name, _, _ in providers)
408          if found:
409              print(f"\n  Plugin:    installed ✓")
410              for pname, _, p in providers:
411                  if pname == provider_name:
412                      if p.is_available():
413                          print(f"  Status:    available ✓")
414                      else:
415                          print(f"  Status:    not available ✗")
416                          schema = p.get_config_schema() if hasattr(p, "get_config_schema") else []
417                          # Check all fields that have env_var (both secret and non-secret)
418                          required_fields = [f for f in schema if f.get("env_var")]
419                          if required_fields:
420                              print(f"  Missing:")
421                              for f in required_fields:
422                                  env_var = f.get("env_var", "")
423                                  url = f.get("url", "")
424                                  is_set = bool(os.environ.get(env_var))
425                                  mark = "✓" if is_set else "✗"
426                                  line = f"    {mark} {env_var}"
427                                  if url and not is_set:
428                                      line += f"  → {url}"
429                                  print(line)
430                      break
431          else:
432              print(f"\n  Plugin:    NOT installed ✗")
433              print(f"  Install the '{provider_name}' memory plugin to ~/.hermes/plugins/")
434  
435      providers = _get_available_providers()
436      if providers:
437          print(f"\n  Installed plugins:")
438          for pname, desc, _ in providers:
439              active = " ← active" if pname == provider_name else ""
440              print(f"    • {pname}  ({desc}){active}")
441  
442      print()
443  
444  
445  # ---------------------------------------------------------------------------
446  # Router
447  # ---------------------------------------------------------------------------
448  
449  def memory_command(args) -> None:
450      """Route memory subcommands."""
451      sub = getattr(args, "memory_command", None)
452      if sub == "setup":
453          cmd_setup(args)
454      elif sub == "status":
455          cmd_status(args)
456      else:
457          cmd_status(args)