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

refactor(memory): restructure plugins, add CLI, clean gateway, migration notice

Browse Source 
Plugin restructure:
- Move all memory plugins from plugins/<name>-memory/ to plugins/memory/<name>/
  (byterover, hindsight, holographic, honcho, mem0, openviking, retaindb)
- New plugins/memory/__init__.py discovery module that scans the directory
  directly, loading providers by name without the general plugin system
- run_agent.py uses load_memory_provider() instead of get_plugin_memory_providers()

CLI wiring:
- hermes memory setup — interactive curses picker + config wizard
- hermes memory status — show active provider, config, availability
- hermes memory off — disable external provider (built-in only)
- hermes honcho — now shows migration notice pointing to hermes memory setup

Gateway cleanup:
- Remove _get_or_create_gateway_honcho (already removed in prev commit)
- Remove _shutdown_gateway_honcho and _shutdown_all_gateway_honcho methods
- Remove all calls to shutdown methods (4 call sites)
- Remove _honcho_managers/_honcho_configs dict references

Dead code removal:
- Delete tools/honcho_tools.py (279 lines, import was already commented out)
- Delete tests/gateway/test_honcho_lifecycle.py (131 lines, tested removed methods)
- Remove if False placeholder from run_agent.py

Migration:
- Honcho migration notice on startup: detects existing honcho.json or
  ~/.honcho/config.json, prints guidance to run hermes memory setup.
  Only fires when memory.provider is not set and not in quiet mode.

Full suite: 7203 passed, 4 pre-existing failures. Zero regressions.
This commit is contained in:
Teknium
2026-03-30 23:46:51 -07:00
parent 52673c9cad
commit 55107a196e
25 changed files with 294 additions and 490 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/run.py
-60
View File
@@ -473,8 +473,6 @@ class GatewayRunner:
# Persistent Honcho managers keyed by gateway session key.
# This preserves write_frequency="session" semantics across short-lived
# per-message AIAgent instances.
self._honcho_managers: Dict[str, Any] = {}
self._honcho_configs: Dict[str, Any] = {}
# Rate-limit compression warning messages sent to users.
# Keyed by chat_id — value is the timestamp of the last warning sent.
@@ -512,61 +510,9 @@ class GatewayRunner:
# Track background tasks to prevent garbage collection mid-execution
self._background_tasks: set = set()
def _get_or_create_gateway_honcho(self, session_key: str):
"""Return a persistent Honcho manager/config pair for this gateway session."""
if not hasattr(self, "_honcho_managers"):
self._honcho_managers = {}
if not hasattr(self, "_honcho_configs"):
self._honcho_configs = {}
if session_key in self._honcho_managers:
return self._honcho_managers[session_key], self._honcho_configs.get(session_key)
try:
from honcho_integration.client import HonchoClientConfig, get_honcho_client
from honcho_integration.session import HonchoSessionManager
hcfg = HonchoClientConfig.from_global_config()
if not hcfg.enabled or not (hcfg.api_key or hcfg.base_url):
return None, hcfg
client = get_honcho_client(hcfg)
manager = HonchoSessionManager(
honcho=client,
config=hcfg,
context_tokens=hcfg.context_tokens,
)
self._honcho_managers[session_key] = manager
self._honcho_configs[session_key] = hcfg
return manager, hcfg
except Exception as e:
logger.debug("Gateway Honcho init failed for %s: %s", session_key, e)
return None, None
def _shutdown_gateway_honcho(self, session_key: str) -> None:
"""Flush and close the persistent Honcho manager for a gateway session."""
managers = getattr(self, "_honcho_managers", None)
configs = getattr(self, "_honcho_configs", None)
if managers is None or configs is None:
return
manager = managers.pop(session_key, None)
configs.pop(session_key, None)
if not manager:
return
try:
manager.shutdown()
except Exception as e:
logger.debug("Gateway Honcho shutdown failed for %s: %s", session_key, e)
def _shutdown_all_gateway_honcho(self) -> None:
"""Flush and close all persistent Honcho managers."""
managers = getattr(self, "_honcho_managers", None)
if not managers:
return
for session_key in list(managers.keys()):
self._shutdown_gateway_honcho(session_key)
# -- Setup skill availability ----------------------------------------
def _has_setup_skill(self) -> bool:
@@ -1294,7 +1240,6 @@ class GatewayRunner:
)
try:
await self._async_flush_memories(entry.session_id, key)
self._shutdown_gateway_honcho(key)
self.session_store._pre_flushed_sessions.add(entry.session_id)
except Exception as e:
logger.debug("Proactive memory flush failed for %s: %s", entry.session_id, e)
@@ -1441,7 +1386,6 @@ class GatewayRunner:
self._running_agents.clear()
self._pending_messages.clear()
self._pending_approvals.clear()
self._shutdown_all_gateway_honcho()
self._shutdown_event.set()
from gateway.status import remove_pid_file, write_runtime_status
@@ -3042,8 +2986,6 @@ class GatewayRunner:
_flush_task.add_done_callback(self._background_tasks.discard)
except Exception as e:
logger.debug("Gateway memory flush on reset failed: %s", e)
self._shutdown_gateway_honcho(session_key)
self._evict_cached_agent(session_key)
# Reset the session
@@ -4413,8 +4355,6 @@ class GatewayRunner:
except Exception as e:
logger.debug("Memory flush on resume failed: %s", e)
self._shutdown_gateway_honcho(session_key)
# Clear any running agent for this session key
if session_key in self._running_agents:
del self._running_agents[session_key]
hermes_cli/main.py
+42 -2
View File
@@ -4294,11 +4294,51 @@ For more help on a command:
)
def cmd_honcho(args):
from honcho_integration.cli import honcho_command
honcho_command(args)
print("\n ⚠️ The 'hermes honcho' command has been replaced.")
print(" Honcho is now a memory provider plugin.\n")
print(" To set up Honcho as your memory provider:")
print(" hermes memory setup\n")
print(" Your existing Honcho configuration and data are preserved.")
print(" Just select 'honcho' in the memory setup wizard.\n")
honcho_parser.set_defaults(func=cmd_honcho)
# =========================================================================
# memory command
# =========================================================================
memory_parser = subparsers.add_parser(
"memory",
help="Configure external memory provider",
description=(
"Set up and manage external memory provider plugins.\n\n"
"Available providers: honcho, openviking, mem0, hindsight,\n"
"holographic, retaindb, byterover.\n\n"
"Only one external provider can be active at a time.\n"
"Built-in memory (MEMORY.md/USER.md) is always active."
),
)
memory_sub = memory_parser.add_subparsers(dest="memory_command")
memory_sub.add_parser("setup", help="Interactive provider selection and configuration")
memory_sub.add_parser("status", help="Show current memory provider config")
memory_off_p = memory_sub.add_parser("off", help="Disable external provider (built-in only)")
def cmd_memory(args):
sub = getattr(args, "memory_command", None)
if sub == "off":
from hermes_cli.config import load_config, save_config
config = load_config()
if not isinstance(config.get("memory"), dict):
config["memory"] = {}
config["memory"]["provider"] = ""
save_config(config)
print("\n ✓ Memory provider: built-in only")
print(" Saved to config.yaml\n")
else:
from hermes_cli.memory_setup import memory_command
memory_command(args)
memory_parser.set_defaults(func=cmd_memory)
# =========================================================================
# tools command
# =========================================================================
hermes_cli/memory_setup.py
+17 -11
View File
@@ -126,28 +126,34 @@ def _prompt(label: str, default: str | None = None, secret: bool = False) -> str
# ---------------------------------------------------------------------------
def _get_available_providers() -> list:
"""Discover memory providers from installed plugins.
"""Discover memory providers from plugins/memory/.
Returns list of (name, description, provider_instance) tuples.
"""
try:
from hermes_cli.plugins import get_plugin_memory_providers
providers = get_plugin_memory_providers()
from plugins.memory import discover_memory_providers, load_memory_provider
raw = discover_memory_providers()
except Exception:
providers = []
raw = []
results = []
for p in providers:
name = getattr(p, "name", "unknown")
schema = p.get_config_schema() if hasattr(p, "get_config_schema") else []
for name, desc, available in raw:
try:
provider = load_memory_provider(name)
if not provider:
continue
except Exception:
continue
# Override description with setup hint
schema = provider.get_config_schema() if hasattr(provider, "get_config_schema") else []
has_secrets = any(f.get("secret") for f in schema)
if has_secrets:
desc = "requires API key"
setup_hint = "requires API key"
elif not schema:
desc = "no setup needed"
setup_hint = "no setup needed"
else:
desc = "local"
results.append((name, desc, p))
setup_hint = "local"
results.append((name, setup_hint, provider))
return results
plugins/__init__.py
+1
View File
@@ -0,0 +1 @@
# Hermes plugins package
plugins/memory/__init__.py
+213
View File
@@ -0,0 +1,213 @@
"""Memory provider plugin discovery.
Scans ``plugins/memory/<name>/`` directories for memory provider plugins.
Each subdirectory must contain ``__init__.py`` with a class implementing
the MemoryProvider ABC.
Memory providers are separate from the general plugin system — they live
in the repo and are always available without user installation. Only ONE
can be active at a time, selected via ``memory.provider`` in config.yaml.
Usage:
from plugins.memory import discover_memory_providers, load_memory_provider
available = discover_memory_providers() # [(name, desc, available), ...]
provider = load_memory_provider("openviking") # MemoryProvider instance
"""
from __future__ import annotations
import importlib
import importlib.util
import logging
import sys
from pathlib import Path
from typing import List, Optional, Tuple
logger = logging.getLogger(__name__)
_MEMORY_PLUGINS_DIR = Path(__file__).parent
def discover_memory_providers() -> List[Tuple[str, str, bool]]:
"""Scan plugins/memory/ for available providers.
Returns list of (name, description, is_available) tuples.
Does NOT import the providers — just reads plugin.yaml for metadata
and does a lightweight availability check.
"""
results = []
if not _MEMORY_PLUGINS_DIR.is_dir():
return results
for child in sorted(_MEMORY_PLUGINS_DIR.iterdir()):
if not child.is_dir() or child.name.startswith(("_", ".")):
continue
init_file = child / "__init__.py"
if not init_file.exists():
continue
# Read description from plugin.yaml if available
desc = ""
yaml_file = child / "plugin.yaml"
if yaml_file.exists():
try:
import yaml
with open(yaml_file) as f:
meta = yaml.safe_load(f) or {}
desc = meta.get("description", "")
except Exception:
pass
# Quick availability check — try loading and calling is_available()
available = True
try:
provider = _load_provider_from_dir(child)
if provider:
available = provider.is_available()
else:
available = False
except Exception:
available = False
results.append((child.name, desc, available))
return results
def load_memory_provider(name: str) -> Optional["MemoryProvider"]:
"""Load and return a MemoryProvider instance by name.
Returns None if the provider is not found or fails to load.
"""
provider_dir = _MEMORY_PLUGINS_DIR / name
if not provider_dir.is_dir():
logger.debug("Memory provider '%s' not found in %s", name, _MEMORY_PLUGINS_DIR)
return None
try:
provider = _load_provider_from_dir(provider_dir)
if provider:
return provider
logger.warning("Memory provider '%s' loaded but no provider instance found", name)
return None
except Exception as e:
logger.warning("Failed to load memory provider '%s': %s", name, e)
return None
def _load_provider_from_dir(provider_dir: Path) -> Optional["MemoryProvider"]:
"""Import a provider module and extract the MemoryProvider instance.
The module must have either:
- A register(ctx) function (plugin-style) — we simulate a ctx
- A top-level class that extends MemoryProvider — we instantiate it
"""
name = provider_dir.name
module_name = f"plugins.memory.{name}"
init_file = provider_dir / "__init__.py"
if not init_file.exists():
return None
# Check if already loaded
if module_name in sys.modules:
mod = sys.modules[module_name]
else:
# Handle relative imports within the plugin
# First ensure the parent packages are registered
for parent in ("plugins", "plugins.memory"):
if parent not in sys.modules:
parent_path = Path(__file__).parent
if parent == "plugins":
parent_path = parent_path.parent
parent_init = parent_path / "__init__.py"
if parent_init.exists():
spec = importlib.util.spec_from_file_location(
parent, str(parent_init),
submodule_search_locations=[str(parent_path)]
)
if spec:
parent_mod = importlib.util.module_from_spec(spec)
sys.modules[parent] = parent_mod
try:
spec.loader.exec_module(parent_mod)
except Exception:
pass
# Now load the provider module
spec = importlib.util.spec_from_file_location(
module_name, str(init_file),
submodule_search_locations=[str(provider_dir)]
)
if not spec:
return None
mod = importlib.util.module_from_spec(spec)
sys.modules[module_name] = mod
# Register submodules so relative imports work
# e.g., "from .store import MemoryStore" in holographic plugin
for sub_file in provider_dir.glob("*.py"):
if sub_file.name == "__init__.py":
continue
sub_name = sub_file.stem
full_sub_name = f"{module_name}.{sub_name}"
if full_sub_name not in sys.modules:
sub_spec = importlib.util.spec_from_file_location(
full_sub_name, str(sub_file)
)
if sub_spec:
sub_mod = importlib.util.module_from_spec(sub_spec)
sys.modules[full_sub_name] = sub_mod
try:
sub_spec.loader.exec_module(sub_mod)
except Exception as e:
logger.debug("Failed to load submodule %s: %s", full_sub_name, e)
try:
spec.loader.exec_module(mod)
except Exception as e:
logger.debug("Failed to exec_module %s: %s", module_name, e)
sys.modules.pop(module_name, None)
return None
# Try register(ctx) pattern first (how our plugins are written)
if hasattr(mod, "register"):
collector = _ProviderCollector()
try:
mod.register(collector)
if collector.provider:
return collector.provider
except Exception as e:
logger.debug("register() failed for %s: %s", name, e)
# Fallback: find a MemoryProvider subclass and instantiate it
from agent.memory_provider import MemoryProvider
for attr_name in dir(mod):
attr = getattr(mod, attr_name, None)
if (isinstance(attr, type) and issubclass(attr, MemoryProvider)
and attr is not MemoryProvider):
try:
return attr()
except Exception:
pass
return None
class _ProviderCollector:
"""Fake plugin context that captures register_memory_provider calls."""
def __init__(self):
self.provider = None
def register_memory_provider(self, provider):
self.provider = provider
# No-op for other registration methods
def register_tool(self, *args, **kwargs):
pass
def register_hook(self, *args, **kwargs):
pass
plugins/byterover-memory/__init__.py → plugins/memory/byterover/__init__.py
View File
plugins/byterover-memory/plugin.yaml → plugins/memory/byterover/plugin.yaml
View File
plugins/hindsight-memory/__init__.py → plugins/memory/hindsight/__init__.py
View File
plugins/hindsight-memory/plugin.yaml → plugins/memory/hindsight/plugin.yaml
View File
plugins/hermes-memory-store/__init__.py → plugins/memory/holographic/__init__.py
View File
plugins/hermes-memory-store/holographic.py → plugins/memory/holographic/holographic.py
View File
plugins/hermes-memory-store/plugin.yaml → plugins/memory/holographic/plugin.yaml
View File
plugins/hermes-memory-store/retrieval.py → plugins/memory/holographic/retrieval.py
View File
plugins/hermes-memory-store/store.py → plugins/memory/holographic/store.py
View File
plugins/honcho-memory/__init__.py → plugins/memory/honcho/__init__.py
View File
plugins/honcho-memory/plugin.yaml → plugins/memory/honcho/plugin.yaml
View File
plugins/mem0-memory/__init__.py → plugins/memory/mem0/__init__.py
View File
plugins/mem0-memory/plugin.yaml → plugins/memory/mem0/plugin.yaml
View File
plugins/openviking-memory/__init__.py → plugins/memory/openviking/__init__.py
View File
plugins/openviking-memory/plugin.yaml → plugins/memory/openviking/plugin.yaml
View File
plugins/retaindb-memory/__init__.py → plugins/memory/retaindb/__init__.py
View File
plugins/retaindb-memory/plugin.yaml → plugins/memory/retaindb/plugin.yaml
View File
run_agent.py
+21 -7
View File
@@ -1019,14 +1019,30 @@ class AIAgent:
if not skip_memory:
try:
_mem_provider_name = mem_config.get("provider", "") if mem_config else ""
# Migration notice for users with Honcho configured but not migrated
if not _mem_provider_name and not self.quiet_mode:
try:
from hermes_constants import get_hermes_home as _ghh2
_honcho_paths = [
_ghh2() / "honcho.json",
Path.home() / ".honcho" / "config.json",
]
if any(p.exists() for p in _honcho_paths):
print(" ℹ️ Detected existing Honcho configuration.")
print(" Honcho is now a memory provider plugin.")
print(" Run 'hermes memory setup' to activate it.")
print()
except Exception:
pass
if _mem_provider_name:
from agent.memory_manager import MemoryManager as _MemoryManager
from hermes_cli.plugins import get_plugin_memory_providers as _get_mem_providers
from plugins.memory import load_memory_provider as _load_mem
self._memory_manager = _MemoryManager()
for _mp in _get_mem_providers():
if _mp.name == _mem_provider_name and _mp.is_available():
self._memory_manager.add_provider(_mp)
break
_mp = _load_mem(_mem_provider_name)
if _mp and _mp.is_available():
self._memory_manager.add_provider(_mp)
if self._memory_manager.providers:
from hermes_constants import get_hermes_home as _ghh
self._memory_manager.initialize_all(
@@ -4695,8 +4711,6 @@ class AIAgent:
return
if "memory" not in self.valid_tool_names or not self._memory_store:
return
if False: # placeholder
return
effective_min = min_turns if min_turns is not None else self._memory_flush_min_turns
if self._user_turn_count < effective_min:
return
tests/gateway/test_honcho_lifecycle.py
-131
View File
@@ -1,131 +0,0 @@
"""Tests for gateway-owned Honcho lifecycle helpers."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from gateway.config import Platform
from gateway.platforms.base import MessageEvent
from gateway.session import SessionSource
def _make_runner():
from gateway.run import GatewayRunner
runner = object.__new__(GatewayRunner)
runner._honcho_managers = {}
runner._honcho_configs = {}
runner._running_agents = {}
runner._pending_messages = {}
runner._pending_approvals = {}
runner.adapters = {}
runner.hooks = MagicMock()
runner.hooks.emit = AsyncMock()
return runner
def _make_event(text="/reset"):
return MessageEvent(
text=text,
source=SessionSource(
platform=Platform.TELEGRAM,
chat_id="chat-1",
user_id="user-1",
user_name="alice",
),
)
class TestGatewayHonchoLifecycle:
def test_gateway_reuses_honcho_manager_for_session_key(self):
runner = _make_runner()
hcfg = SimpleNamespace(
enabled=True,
api_key="honcho-key",
ai_peer="hermes",
peer_name="alice",
context_tokens=123,
peer_memory_mode=lambda peer: "hybrid",
)
manager = MagicMock()
with (
patch("honcho_integration.client.HonchoClientConfig.from_global_config", return_value=hcfg),
patch("honcho_integration.client.get_honcho_client", return_value=MagicMock()),
patch("honcho_integration.session.HonchoSessionManager", return_value=manager) as mock_mgr_cls,
):
first_mgr, first_cfg = runner._get_or_create_gateway_honcho("session-key")
second_mgr, second_cfg = runner._get_or_create_gateway_honcho("session-key")
assert first_mgr is manager
assert second_mgr is manager
assert first_cfg is hcfg
assert second_cfg is hcfg
mock_mgr_cls.assert_called_once()
def test_gateway_skips_honcho_manager_when_disabled(self):
runner = _make_runner()
hcfg = SimpleNamespace(
enabled=False,
api_key="honcho-key",
ai_peer="hermes",
peer_name="alice",
)
with (
patch("honcho_integration.client.HonchoClientConfig.from_global_config", return_value=hcfg),
patch("honcho_integration.client.get_honcho_client") as mock_client,
patch("honcho_integration.session.HonchoSessionManager") as mock_mgr_cls,
):
manager, cfg = runner._get_or_create_gateway_honcho("session-key")
assert manager is None
assert cfg is hcfg
mock_client.assert_not_called()
mock_mgr_cls.assert_not_called()
@pytest.mark.asyncio
async def test_reset_shuts_down_gateway_honcho_manager(self):
runner = _make_runner()
event = _make_event()
runner._shutdown_gateway_honcho = MagicMock()
runner._async_flush_memories = AsyncMock()
runner.session_store = MagicMock()
runner.session_store._generate_session_key.return_value = "gateway-key"
runner.session_store._entries = {
"gateway-key": SimpleNamespace(session_id="old-session"),
}
runner.session_store.reset_session.return_value = SimpleNamespace(session_id="new-session")
result = await runner._handle_reset_command(event)
runner._shutdown_gateway_honcho.assert_called_once_with("gateway-key")
runner._async_flush_memories.assert_called_once_with("old-session", "gateway-key")
assert "Session reset" in result
def test_flush_memories_reuses_gateway_session_key_and_skips_honcho_sync(self):
runner = _make_runner()
runner.session_store = MagicMock()
runner.session_store.load_transcript.return_value = [
{"role": "user", "content": "a"},
{"role": "assistant", "content": "b"},
{"role": "user", "content": "c"},
{"role": "assistant", "content": "d"},
]
tmp_agent = MagicMock()
with (
patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}),
patch("gateway.run._resolve_gateway_model", return_value="model-name"),
patch("run_agent.AIAgent", return_value=tmp_agent) as mock_agent_cls,
):
runner._flush_memories_for_session("old-session", "gateway-key")
mock_agent_cls.assert_called_once()
_, kwargs = mock_agent_cls.call_args
assert kwargs["session_id"] == "old-session"
assert kwargs["honcho_session_key"] == "gateway-key"
tmp_agent.run_conversation.assert_called_once()
_, run_kwargs = tmp_agent.run_conversation.call_args
assert run_kwargs["sync_honcho"] is False
tools/honcho_tools.py
-279
View File
@@ -1,279 +0,0 @@
"""Honcho tools for user context retrieval.
Registers three complementary tools, ordered by capability:
honcho_context — dialectic Q&A (LLM-powered, direct answers)
honcho_search — semantic search (fast, no LLM, raw excerpts)
honcho_profile — peer card (fast, no LLM, structured facts)
Use honcho_context when you need Honcho to synthesize an answer.
Use honcho_search or honcho_profile when you want raw data to reason
over yourself.
The session key is injected at runtime by the agent loop via
``set_session_context()``.
"""
import json
import logging
logger = logging.getLogger(__name__)
# ── Module-level state (injected by AIAgent at init time) ──
_session_manager = None # HonchoSessionManager instance
_session_key: str | None = None # Current session key (e.g., "telegram:123456")
def set_session_context(session_manager, session_key: str) -> None:
"""Register the active Honcho session manager and key.
Called by AIAgent.__init__ when Honcho is enabled.
"""
global _session_manager, _session_key
_session_manager = session_manager
_session_key = session_key
def clear_session_context() -> None:
"""Clear session context (for testing or shutdown)."""
global _session_manager, _session_key
_session_manager = None
_session_key = None
# ── Availability check ──
def _check_honcho_available() -> bool:
"""Tool is available when Honcho is active OR configured.
At banner time the session context hasn't been injected yet, but if
a valid config exists the tools *will* activate once the agent starts.
Returning True for "configured" prevents the banner from marking
honcho tools as red/disabled when they're actually going to work.
"""
# Fast path: session already active (mid-conversation)
if _session_manager is not None and _session_key is not None:
return True
# Slow path: check if Honcho is configured (banner time)
try:
from honcho_integration.client import HonchoClientConfig
cfg = HonchoClientConfig.from_global_config()
return cfg.enabled and bool(cfg.api_key or cfg.base_url)
except Exception:
return False
def _resolve_session_context(**kwargs):
"""Prefer the calling agent's session context over module-global fallback."""
session_manager = kwargs.get("honcho_manager") or _session_manager
session_key = kwargs.get("honcho_session_key") or _session_key
return session_manager, session_key
# ── honcho_profile ──
_PROFILE_SCHEMA = {
"name": "honcho_profile",
"description": (
"Retrieve the user's peer card from Honcho — a curated list of key facts "
"about them (name, role, preferences, communication style, patterns). "
"Fast, no LLM reasoning, minimal cost. "
"Use this at conversation start or when you need a quick factual snapshot. "
"Use honcho_context instead when you need Honcho to synthesize an answer."
),
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
}
def _handle_honcho_profile(args: dict, **kw) -> str:
session_manager, session_key = _resolve_session_context(**kw)
if not session_manager or not session_key:
return json.dumps({"error": "Honcho is not active for this session."})
try:
card = session_manager.get_peer_card(session_key)
if not card:
return json.dumps({"result": "No profile facts available yet. The user's profile builds over time through conversations."})
return json.dumps({"result": card})
except Exception as e:
logger.error("Error fetching Honcho peer card: %s", e)
return json.dumps({"error": f"Failed to fetch profile: {e}"})
# ── honcho_search ──
_SEARCH_SCHEMA = {
"name": "honcho_search",
"description": (
"Semantic search over Honcho's stored context about the user. "
"Returns raw excerpts ranked by relevance to your query — no LLM synthesis. "
"Cheaper and faster than honcho_context. "
"Good when you want to find specific past facts and reason over them yourself. "
"Use honcho_context when you need a direct synthesized answer."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "What to search for in Honcho's memory (e.g. 'programming languages', 'past projects', 'timezone').",
},
"max_tokens": {
"type": "integer",
"description": "Token budget for returned context (default 800, max 2000).",
},
},
"required": ["query"],
},
}
def _handle_honcho_search(args: dict, **kw) -> str:
query = args.get("query", "")
if not query:
return json.dumps({"error": "Missing required parameter: query"})
session_manager, session_key = _resolve_session_context(**kw)
if not session_manager or not session_key:
return json.dumps({"error": "Honcho is not active for this session."})
max_tokens = min(int(args.get("max_tokens", 800)), 2000)
try:
result = session_manager.search_context(session_key, query, max_tokens=max_tokens)
if not result:
return json.dumps({"result": "No relevant context found."})
return json.dumps({"result": result})
except Exception as e:
logger.error("Error searching Honcho context: %s", e)
return json.dumps({"error": f"Failed to search context: {e}"})
# ── honcho_context (dialectic — LLM-powered) ──
_QUERY_SCHEMA = {
"name": "honcho_context",
"description": (
"Ask Honcho a natural language question and get a synthesized answer. "
"Uses Honcho's LLM (dialectic reasoning) — higher cost than honcho_profile or honcho_search. "
"Can query about any peer: the user (default), the AI assistant, or any named peer. "
"Examples: 'What are the user's main goals?', 'What has hermes been working on?', "
"'What is the user's technical expertise level?'"
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "A natural language question.",
},
"peer": {
"type": "string",
"description": "Which peer to query about: 'user' (default) or 'ai'. Omit for user.",
},
},
"required": ["query"],
},
}
def _handle_honcho_context(args: dict, **kw) -> str:
query = args.get("query", "")
if not query:
return json.dumps({"error": "Missing required parameter: query"})
session_manager, session_key = _resolve_session_context(**kw)
if not session_manager or not session_key:
return json.dumps({"error": "Honcho is not active for this session."})
peer_target = args.get("peer", "user")
try:
result = session_manager.dialectic_query(session_key, query, peer=peer_target)
return json.dumps({"result": result or "No result from Honcho."})
except Exception as e:
logger.error("Error querying Honcho context: %s", e)
return json.dumps({"error": f"Failed to query context: {e}"})
# ── honcho_conclude ──
_CONCLUDE_SCHEMA = {
"name": "honcho_conclude",
"description": (
"Write a conclusion about the user back to Honcho's memory. "
"Conclusions are persistent facts that build the user's profile — "
"preferences, corrections, clarifications, project context, or anything "
"the user tells you that should be remembered across sessions. "
"Use this when the user explicitly states a preference, corrects you, "
"or shares something they want remembered. "
"Examples: 'User prefers dark mode', 'User's project uses Python 3.11', "
"'User corrected: their name is spelled Eri not Eric'."
),
"parameters": {
"type": "object",
"properties": {
"conclusion": {
"type": "string",
"description": "A factual statement about the user to persist in memory.",
}
},
"required": ["conclusion"],
},
}
def _handle_honcho_conclude(args: dict, **kw) -> str:
conclusion = args.get("conclusion", "")
if not conclusion:
return json.dumps({"error": "Missing required parameter: conclusion"})
session_manager, session_key = _resolve_session_context(**kw)
if not session_manager or not session_key:
return json.dumps({"error": "Honcho is not active for this session."})
try:
ok = session_manager.create_conclusion(session_key, conclusion)
if ok:
return json.dumps({"result": f"Conclusion saved: {conclusion}"})
return json.dumps({"error": "Failed to save conclusion."})
except Exception as e:
logger.error("Error creating Honcho conclusion: %s", e)
return json.dumps({"error": f"Failed to save conclusion: {e}"})
# ── Registration ──
from tools.registry import registry
registry.register(
name="honcho_profile",
toolset="honcho",
schema=_PROFILE_SCHEMA,
handler=_handle_honcho_profile,
check_fn=_check_honcho_available,
emoji="🔮",
)
registry.register(
name="honcho_search",
toolset="honcho",
schema=_SEARCH_SCHEMA,
handler=_handle_honcho_search,
check_fn=_check_honcho_available,
emoji="🔮",
)
registry.register(
name="honcho_context",
toolset="honcho",
schema=_QUERY_SCHEMA,
handler=_handle_honcho_context,
check_fn=_check_honcho_available,
emoji="🔮",
)
registry.register(
name="honcho_conclude",
toolset="honcho",
schema=_CONCLUDE_SCHEMA,
handler=_handle_honcho_conclude,
check_fn=_check_honcho_available,
emoji="🔮",
)