diff --git a/cli.py b/cli.py index edd0b6640b9d..4fdedd674869 100644 --- a/cli.py +++ b/cli.py @@ -66,6 +66,17 @@ ) from hermes_cli.banner import _format_context_length + +def _print_ascii_banner_fallback(console, model: str, session_id: str | None) -> None: + """Fallback banner for terminals that cannot render the default Rich output.""" + stream = getattr(sys, "__stdout__", None) or sys.stdout + stream.write("Hermes Agent\n") + stream.write(f"Model: {model}\n") + if session_id: + stream.write(f"Session: {session_id}\n") + stream.write("Banner rendering fell back to ASCII because this terminal encoding does not support the default banner.\n") + stream.flush() + _COMMAND_SPINNER_FRAMES = ("⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏") @@ -1962,50 +1973,53 @@ def _init_agent(self, *, model_override: str = None, runtime_override: dict = No def show_banner(self): """Display the welcome banner in Claude Code style.""" - self.console.clear() - if self.preloaded_skills and not self._startup_skills_line_shown: - skills_label = ", ".join(self.preloaded_skills) - self.console.print( - f"[bold {_accent_hex()}]Activated skills:[/] {skills_label}" - ) + try: + self.console.clear() + if self.preloaded_skills and not self._startup_skills_line_shown: + skills_label = ", ".join(self.preloaded_skills) + self.console.print( + f"[bold {_accent_hex()}]Activated skills:[/] {skills_label}" + ) + self.console.print() + self._startup_skills_line_shown = True + + # Auto-compact for narrow terminals — the full banner with caduceus + # + tool list needs ~80 columns minimum to render without wrapping. + term_width = shutil.get_terminal_size().columns + use_compact = self.compact or term_width < 80 + + if use_compact: + self.console.print(_build_compact_banner()) + self._show_status() + else: + # Get tools for display + tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, quiet_mode=True) + + # Get terminal working directory (where commands will execute) + cwd = os.getenv("TERMINAL_CWD", os.getcwd()) + + # Get context length for display + ctx_len = None + if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'): + ctx_len = self.agent.context_compressor.context_length + + # Build and display the banner + build_welcome_banner( + console=self.console, + model=self.model, + cwd=cwd, + tools=tools, + enabled_toolsets=self.enabled_toolsets, + session_id=self.session_id, + context_length=ctx_len, + ) + + # Show tool availability warnings if any tools are disabled + self._show_tool_availability_warnings() + self.console.print() - self._startup_skills_line_shown = True - - # Auto-compact for narrow terminals — the full banner with caduceus - # + tool list needs ~80 columns minimum to render without wrapping. - term_width = shutil.get_terminal_size().columns - use_compact = self.compact or term_width < 80 - - if use_compact: - self.console.print(_build_compact_banner()) - self._show_status() - else: - # Get tools for display - tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, quiet_mode=True) - - # Get terminal working directory (where commands will execute) - cwd = os.getenv("TERMINAL_CWD", os.getcwd()) - - # Get context length for display - ctx_len = None - if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'): - ctx_len = self.agent.context_compressor.context_length - - # Build and display the banner - build_welcome_banner( - console=self.console, - model=self.model, - cwd=cwd, - tools=tools, - enabled_toolsets=self.enabled_toolsets, - session_id=self.session_id, - context_length=ctx_len, - ) - - # Show tool availability warnings if any tools are disabled - self._show_tool_availability_warnings() - - self.console.print() + except UnicodeEncodeError: + _print_ascii_banner_fallback(self.console, self.model, self.session_id) def _preload_resumed_session(self) -> bool: """Load a resumed session's history from the DB early (before first chat). diff --git a/gateway/run.py b/gateway/run.py index b4b6c6ef05d8..9ec874d622e9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -368,6 +368,8 @@ def __init__(self, config: Optional[GatewayConfig] = None): # per-message AIAgent instances. self._honcho_managers: Dict[str, Any] = {} self._honcho_configs: Dict[str, Any] = {} + self._retaindb_managers: Dict[str, Any] = {} + self._retaindb_configs: Dict[str, Any] = {} # Ensure tirith security scanner is available (downloads if needed) try: @@ -426,6 +428,49 @@ def _get_or_create_gateway_honcho(self, session_key: str): logger.debug("Gateway Honcho init failed for %s: %s", session_key, e) return None, None + def _build_retaindb_identity(self, source, session_id: str) -> dict: + """Build RetainDB runtime identity hints from the gateway session source.""" + return { + "platform": source.platform.value if getattr(source, "platform", None) else "", + "platform_user_id": getattr(source, "user_id", None), + "user_name": getattr(source, "user_name", None), + "chat_id": getattr(source, "chat_id", None), + "peer_name": getattr(source, "user_name", None), + "session_id": session_id, + } + + def _get_or_create_gateway_retaindb(self, session_key: str, identity: Optional[dict] = None): + """Return a persistent RetainDB manager/config pair for this gateway session.""" + if not hasattr(self, "_retaindb_managers"): + self._retaindb_managers = {} + if not hasattr(self, "_retaindb_configs"): + self._retaindb_configs = {} + + if session_key in self._retaindb_managers: + manager = self._retaindb_managers[session_key] + if identity and hasattr(manager, "set_runtime_identity"): + manager.set_runtime_identity(identity) + return manager, self._retaindb_configs.get(session_key) + + try: + from retaindb_integration.client import RetainDBClientConfig + from retaindb_integration.session import RetainDBSessionManager + + rcfg = RetainDBClientConfig.from_global_config() + if not rcfg.should_activate(): + return None, rcfg + + manager = RetainDBSessionManager( + config=rcfg, + runtime_identity=identity or {}, + ) + self._retaindb_managers[session_key] = manager + self._retaindb_configs[session_key] = rcfg + return manager, rcfg + except Exception as e: + logger.debug("Gateway RetainDB 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) @@ -442,6 +487,22 @@ def _shutdown_gateway_honcho(self, session_key: str) -> None: except Exception as e: logger.debug("Gateway Honcho shutdown failed for %s: %s", session_key, e) + def _shutdown_gateway_retaindb(self, session_key: str) -> None: + """Flush and close the persistent RetainDB manager for a gateway session.""" + managers = getattr(self, "_retaindb_managers", None) + configs = getattr(self, "_retaindb_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 RetainDB 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) @@ -449,6 +510,14 @@ def _shutdown_all_gateway_honcho(self) -> None: return for session_key in list(managers.keys()): self._shutdown_gateway_honcho(session_key) + + def _shutdown_all_gateway_retaindb(self) -> None: + """Flush and close all persistent RetainDB managers.""" + managers = getattr(self, "_retaindb_managers", None) + if not managers: + return + for session_key in list(managers.keys()): + self._shutdown_gateway_retaindb(session_key) # -- Setup skill availability ---------------------------------------- @@ -515,6 +584,7 @@ def _flush_memories_for_session( self, old_session_id: str, honcho_session_key: Optional[str] = None, + source: Optional[SessionSource] = None, ): """Prompt the agent to save memories/skills before context is lost. @@ -536,6 +606,17 @@ def _flush_memories_for_session( # active provider is openai-codex. model = _resolve_gateway_model() + retaindb_identity = ( + self._build_retaindb_identity(source, old_session_id) if source else None + ) + retaindb_manager = None + retaindb_config = None + if honcho_session_key: + retaindb_manager, retaindb_config = self._get_or_create_gateway_retaindb( + honcho_session_key, + retaindb_identity, + ) + tmp_agent = AIAgent( **runtime_kwargs, model=model, @@ -544,6 +625,10 @@ def _flush_memories_for_session( enabled_toolsets=["memory", "skills"], session_id=old_session_id, honcho_session_key=honcho_session_key, + retaindb_session_key=old_session_id, + retaindb_manager=retaindb_manager, + retaindb_config=retaindb_config, + retaindb_identity=retaindb_identity, ) # Build conversation history from transcript @@ -587,6 +672,7 @@ async def _async_flush_memories( self, old_session_id: str, honcho_session_key: Optional[str] = None, + source: Optional[SessionSource] = None, ): """Run the sync memory flush in a thread pool so it won't block the event loop.""" loop = asyncio.get_event_loop() @@ -595,6 +681,7 @@ async def _async_flush_memories( self._flush_memories_for_session, old_session_id, honcho_session_key, + source, ) @property @@ -1055,8 +1142,9 @@ async def _session_expiry_watcher(self, interval: int = 300): entry.session_id, key, ) try: - await self._async_flush_memories(entry.session_id, key) + await self._async_flush_memories(entry.session_id, key, getattr(entry, "origin", None)) self._shutdown_gateway_honcho(key) + self._shutdown_gateway_retaindb(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) @@ -1098,6 +1186,7 @@ async def stop(self) -> None: self._pending_messages.clear() self._pending_approvals.clear() self._shutdown_all_gateway_honcho() + self._shutdown_all_gateway_retaindb() self._shutdown_event.set() from gateway.status import remove_pid_file, write_runtime_status @@ -2479,12 +2568,13 @@ async def _handle_reset_command(self, event: MessageEvent) -> str: old_entry = self.session_store._entries.get(session_key) if old_entry: asyncio.create_task( - self._async_flush_memories(old_entry.session_id, session_key) + self._async_flush_memories(old_entry.session_id, session_key, getattr(old_entry, "origin", None)) ) except Exception as e: logger.debug("Gateway memory flush on reset failed: %s", e) self._shutdown_gateway_honcho(session_key) + self._shutdown_gateway_retaindb(session_key) self._evict_cached_agent(session_key) # Reset the session @@ -3891,12 +3981,13 @@ async def _handle_resume_command(self, event: MessageEvent) -> str: # Flush memories for current session before switching try: asyncio.create_task( - self._async_flush_memories(current_entry.session_id, session_key) + self._async_flush_memories(current_entry.session_id, session_key, getattr(current_entry, "origin", None)) ) except Exception as e: logger.debug("Memory flush on resume failed: %s", e) self._shutdown_gateway_honcho(session_key) + self._shutdown_gateway_retaindb(session_key) # Clear any running agent for this session key if session_key in self._running_agents: @@ -4957,6 +5048,11 @@ def run_sync(): pr = self._provider_routing honcho_manager, honcho_config = self._get_or_create_gateway_honcho(session_key) + retaindb_identity = self._build_retaindb_identity(source, session_id) + retaindb_manager, retaindb_config = self._get_or_create_gateway_retaindb( + session_key, + retaindb_identity, + ) reasoning_config = self._load_reasoning_config() self._reasoning_config = reasoning_config # Set up streaming consumer if enabled @@ -5032,6 +5128,10 @@ def run_sync(): honcho_session_key=session_key, honcho_manager=honcho_manager, honcho_config=honcho_config, + retaindb_session_key=session_id, + retaindb_manager=retaindb_manager, + retaindb_config=retaindb_config, + retaindb_identity=retaindb_identity, session_db=self._session_db, fallback_model=self._fallback_model, ) diff --git a/hermes b/hermes index f0feeb2bad80..4445fbc18bd5 100755 --- a/hermes +++ b/hermes @@ -7,6 +7,5 @@ Usage: ./hermes [options] """ if __name__ == "__main__": - from cli import main - import fire - fire.Fire(main) + from hermes_cli.main import main + main() diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 4c6179c5f5d2..245f04ee7b21 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -322,6 +322,23 @@ def ensure_hermes_home(): # (apiKey, workspace, peerName, sessions, enabled) comes from the global config. "honcho": {}, + # Native RetainDB memory integration. + # RetainDB adds optional cross-session recall on top of Hermes local memory. + "retaindb": { + "enabled": True, + "base_url": "https://api.retaindb.com", + "project": "", + "memory_mode": "hybrid", + "recall_mode": "hybrid", + "write_frequency": "async", + "context_tokens": 1200, + "prefetch_timeout_ms": 1500, + "flush_batch_size": 50, + "disable_tool_exposure": False, + "debug_recall_trace": False, + "agent_id": "hermes", + }, + # IANA timezone (e.g. "Asia/Kolkata", "America/New_York"). # Empty string means use server-local time. "timezone": "", @@ -676,6 +693,19 @@ def ensure_hermes_home(): "prompt": "Honcho base URL (e.g. http://localhost:8000)", "category": "tool", }, + "RETAINDB_API_KEY": { + "description": "RetainDB API key for native Hermes deep memory", + "prompt": "RetainDB API key", + "url": "https://retaindb.com/dashboard", + "tools": ["retaindb_context"], + "password": True, + "category": "tool", + }, + "RETAINDB_BASE_URL": { + "description": "Base URL for self-hosted RetainDB instances", + "prompt": "RetainDB base URL (leave empty for cloud)", + "category": "tool", + }, # ── Messaging platforms ── "TELEGRAM_BOT_TOKEN": { diff --git a/hermes_cli/doctor.py b/hermes_cli/doctor.py index a28433dd15f0..29a540df3453 100644 --- a/hermes_cli/doctor.py +++ b/hermes_cli/doctor.py @@ -66,18 +66,35 @@ def _honcho_is_configured_for_doctor() -> bool: return False +def _retaindb_is_configured_for_doctor() -> bool: + """Return True when RetainDB is configured, even if this process has no active session.""" + try: + from retaindb_integration.client import RetainDBClientConfig + + cfg = RetainDBClientConfig.from_global_config() + return bool(cfg.enabled and cfg.api_key and cfg.project) + except Exception: + return False + + def _apply_doctor_tool_availability_overrides(available: list[str], unavailable: list[dict]) -> tuple[list[str], list[dict]]: """Adjust runtime-gated tool availability for doctor diagnostics.""" - if not _honcho_is_configured_for_doctor(): + has_honcho = _honcho_is_configured_for_doctor() + has_retaindb = _retaindb_is_configured_for_doctor() + if not has_honcho and not has_retaindb: return available, unavailable updated_available = list(available) updated_unavailable = [] for item in unavailable: if item.get("name") == "honcho": - if "honcho" not in updated_available: + if has_honcho and "honcho" not in updated_available: updated_available.append("honcho") - continue + continue + if item.get("name") == "retaindb": + if has_retaindb and "retaindb" not in updated_available: + updated_available.append("retaindb") + continue updated_unavailable.append(item) return updated_available, updated_unavailable @@ -745,6 +762,39 @@ def run_doctor(args): except Exception as _e: check_warn("Honcho check failed", str(_e)) + # ========================================================================= + # RetainDB memory + # ========================================================================= + print() + print(color("â—† RetainDB Memory", Colors.CYAN, Colors.BOLD)) + + try: + from retaindb_integration.client import RetainDBClientConfig + from retaindb_integration.session import RetainDBSessionManager + + rcfg = RetainDBClientConfig.from_global_config() + + if not rcfg.enabled: + check_info("RetainDB disabled (set retaindb.enabled: true in ~/.hermes/config.yaml to activate)") + elif not rcfg.project: + check_warn("RetainDB project not set", "run: hermes retaindb setup") + issues.append("No RetainDB project - run 'hermes retaindb setup'") + elif not rcfg.api_key: + check_fail("RetainDB API key not set", "run: hermes retaindb setup") + issues.append("No RetainDB API key - run 'hermes retaindb setup'") + else: + status = RetainDBSessionManager(config=rcfg).connection_status() + if status.get("ok"): + check_ok( + "RetainDB connected", + f"project={rcfg.project} mode={rcfg.memory_mode} freq={rcfg.write_frequency}", + ) + else: + check_fail("RetainDB connection failed", str(status.get("error") or "unknown error")) + issues.append(f"RetainDB unreachable: {status.get('error')}") + except Exception as _e: + check_warn("RetainDB check failed", str(_e)) + # ========================================================================= # Summary # ========================================================================= diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 9a2989484f2a..67ed193343f5 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -3698,6 +3698,62 @@ def cmd_honcho(args): honcho_parser.set_defaults(func=cmd_honcho) + # ========================================================================= + # retaindb command + # ========================================================================= + retaindb_parser = subparsers.add_parser( + "retaindb", + help="Manage native RetainDB memory integration", + description=( + "RetainDB adds optional cross-session memory to Hermes.\n\n" + "Hermes keeps local MEMORY.md / USER.md and state.db, while RetainDB adds " + "background session ingestion and deeper recall across sessions.\n\n" + "Modes: hybrid (RetainDB + local memory) and retaindb (RetainDB only)." + ), + formatter_class=__import__("argparse").RawDescriptionHelpFormatter, + ) + retaindb_subparsers = retaindb_parser.add_subparsers(dest="retaindb_command") + + retaindb_setup = retaindb_subparsers.add_parser("setup", help="Interactive setup wizard for RetainDB") + retaindb_setup.add_argument("--yes", action="store_true", help="Non-interactive setup using env vars / flags") + retaindb_setup.add_argument("--api-key", help="RetainDB API key") + retaindb_setup.add_argument("--project", help="RetainDB project slug/name (creates it if missing)") + retaindb_setup.add_argument("--base-url", help="RetainDB base URL") + + retaindb_subparsers.add_parser("status", help="Show current RetainDB config and connection status") + retaindb_subparsers.add_parser("test", help="Run RetainDB read/write smoke tests") + + retaindb_mode = retaindb_subparsers.add_parser( + "mode", help="Show or set memory mode (hybrid/retaindb)" + ) + retaindb_mode.add_argument( + "mode", nargs="?", metavar="MODE", + choices=("hybrid", "retaindb"), + help="Memory mode to set (hybrid/retaindb). Omit to show current.", + ) + + retaindb_tokens = retaindb_subparsers.add_parser( + "tokens", help="Show or set RetainDB context token budget" + ) + retaindb_tokens.add_argument( + "--context", type=int, metavar="N", + help="Max tokens for RetainDB query context injection", + ) + + retaindb_identity = retaindb_subparsers.add_parser( + "identity", help="Show the resolved RetainDB identity mapping" + ) + retaindb_identity.add_argument( + "--session-id", metavar="SESSION", + help="Optional Hermes session id to inspect", + ) + + def cmd_retaindb(args): + from retaindb_integration.cli import retaindb_command + retaindb_command(args) + + retaindb_parser.set_defaults(func=cmd_retaindb) + # ========================================================================= # tools command # ========================================================================= diff --git a/model_tools.py b/model_tools.py index c651d93ed73d..7571fd5d27b9 100644 --- a/model_tools.py +++ b/model_tools.py @@ -157,6 +157,7 @@ def _discover_tools(): "tools.process_registry", "tools.send_message_tool", "tools.honcho_tools", + "tools.retaindb_tools", "tools.homeassistant_tool", ] import importlib @@ -373,6 +374,8 @@ def handle_function_call( enabled_tools: Optional[List[str]] = None, honcho_manager: Optional[Any] = None, honcho_session_key: Optional[str] = None, + retaindb_manager: Optional[Any] = None, + retaindb_session_key: Optional[str] = None, ) -> str: """ Main function call dispatcher that routes calls to the tool registry. @@ -419,6 +422,8 @@ def handle_function_call( enabled_tools=sandbox_enabled, honcho_manager=honcho_manager, honcho_session_key=honcho_session_key, + retaindb_manager=retaindb_manager, + retaindb_session_key=retaindb_session_key, ) else: result = registry.dispatch( @@ -427,6 +432,8 @@ def handle_function_call( user_task=user_task, honcho_manager=honcho_manager, honcho_session_key=honcho_session_key, + retaindb_manager=retaindb_manager, + retaindb_session_key=retaindb_session_key, ) try: diff --git a/pyproject.toml b/pyproject.toml index cb5141829af0..861e4f95d1cb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,7 +97,7 @@ hermes-acp = "acp_adapter.entry:main" py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajectory_compressor", "toolset_distributions", "cli", "hermes_constants", "hermes_state", "hermes_time", "mini_swe_runner", "minisweagent_path", "rl_cli", "utils"] [tool.setuptools.packages.find] -include = ["agent", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "cron", "honcho_integration", "acp_adapter"] +include = ["agent", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "cron", "honcho_integration", "retaindb_integration", "acp_adapter"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/retaindb_integration/__init__.py b/retaindb_integration/__init__.py new file mode 100644 index 000000000000..c1072548f053 --- /dev/null +++ b/retaindb_integration/__init__.py @@ -0,0 +1,14 @@ +"""Native RetainDB integration for Hermes Agent.""" + +from retaindb_integration.client import RetainDBClient, RetainDBClientConfig, RetainDBClientError +from retaindb_integration.identity import ResolvedRetainDBIdentity, RetainDBIdentityResolver +from retaindb_integration.session import RetainDBSessionManager + +__all__ = [ + "ResolvedRetainDBIdentity", + "RetainDBClient", + "RetainDBClientConfig", + "RetainDBClientError", + "RetainDBIdentityResolver", + "RetainDBSessionManager", +] diff --git a/retaindb_integration/cli.py b/retaindb_integration/cli.py new file mode 100644 index 000000000000..35a25c556fd7 --- /dev/null +++ b/retaindb_integration/cli.py @@ -0,0 +1,332 @@ +"""CLI commands for Hermes' native RetainDB integration.""" + +from __future__ import annotations + +import os +import sys +from dataclasses import replace +from pathlib import Path + +from retaindb_integration.client import DEFAULT_BASE_URL, RetainDBClient, RetainDBClientConfig +from retaindb_integration.session import RetainDBSessionManager + + +def _prompt(label: str, default: str | None = None, secret: bool = False) -> str: + suffix = f" [{default}]" if default else "" + sys.stdout.write(f" {label}{suffix}: ") + sys.stdout.flush() + if secret and sys.stdin.isatty(): + import getpass + + value = getpass.getpass(prompt="") + else: + value = sys.stdin.readline().strip() + return value or (default or "") + + +def _load_hermes_config() -> dict: + from hermes_cli.config import load_config + + return load_config() or {} + + +def _save_hermes_config(cfg: dict) -> None: + from hermes_cli.config import save_config + + save_config(cfg) + + +def _config_path() -> Path: + from hermes_cli.config import get_config_path + + return get_config_path() + + +def _env_path() -> Path: + from hermes_cli.config import get_env_path + + return get_env_path() + + +def _write_runtime_config( + cfg: dict, + *, + project: str, + base_url: str, + enabled: bool = True, +) -> None: + block = cfg.setdefault("retaindb", {}) + block.update( + { + "enabled": enabled, + "base_url": base_url or DEFAULT_BASE_URL, + "project": project, + "memory_mode": block.get("memory_mode", "hybrid") or "hybrid", + "recall_mode": block.get("recall_mode", "hybrid") or "hybrid", + "write_frequency": block.get("write_frequency", "async") or "async", + "context_tokens": int(block.get("context_tokens", 1200) or 1200), + "prefetch_timeout_ms": int(block.get("prefetch_timeout_ms", 1500) or 1500), + "flush_batch_size": int(block.get("flush_batch_size", 50) or 50), + "disable_tool_exposure": bool(block.get("disable_tool_exposure", False)), + "debug_recall_trace": bool(block.get("debug_recall_trace", False)), + "agent_id": str(block.get("agent_id", "hermes") or "hermes"), + } + ) + _save_hermes_config(cfg) + + +def _save_env(api_key: str, base_url: str) -> None: + from hermes_cli.config import save_env_value + + save_env_value("RETAINDB_API_KEY", api_key) + save_env_value( + "RETAINDB_BASE_URL", + base_url if base_url and base_url != DEFAULT_BASE_URL else "", + ) + + +def _select_or_create_project( + client: RetainDBClient, + args, + *, + interactive: bool, +) -> str: + projects = client.list_projects() + explicit_project = str(getattr(args, "project", None) or "").strip() + if explicit_project: + for project in projects: + if explicit_project in { + str(project.get("id") or ""), + str(project.get("slug") or ""), + str(project.get("name") or ""), + }: + return str(project.get("slug") or project.get("name") or project.get("id")) + created = client.create_project(explicit_project) + return str(created.get("slug") or created.get("name") or created.get("id")) + + if not projects: + if not interactive: + raise RuntimeError("No RetainDB projects found. Re-run with --project to auto-create one.") + project_name = _prompt("Project name", default="hermes-agent") + created = client.create_project(project_name) + return str(created.get("slug") or created.get("name") or created.get("id")) + + if not interactive: + if len(projects) == 1: + project = projects[0] + return str(project.get("slug") or project.get("name") or project.get("id")) + raise RuntimeError("Multiple RetainDB projects found. Re-run with --project .") + + print("\n Available RetainDB projects:") + for idx, project in enumerate(projects, start=1): + label = project.get("slug") or project.get("name") or project.get("id") + print(f" {idx}. {label}") + print(f" {len(projects) + 1}. Create new project") + + choice = _prompt("Choose project", default="1") + try: + selected = int(choice) + except ValueError: + selected = 1 + + if selected == len(projects) + 1: + project_name = _prompt("New project name", default="hermes-agent") + created = client.create_project(project_name) + return str(created.get("slug") or created.get("name") or created.get("id")) + + project = projects[max(0, min(len(projects) - 1, selected - 1))] + return str(project.get("slug") or project.get("name") or project.get("id")) + + +def _smoke_test(client: RetainDBClient, config: RetainDBClientConfig) -> tuple[bool, str]: + smoke_config = replace( + config, + prefetch_timeout_ms=max(int(config.prefetch_timeout_ms or 0), 8000), + ) + manager = RetainDBSessionManager(client=RetainDBClient(smoke_config), config=smoke_config) + identity = manager.resolve_identity("hermes-retaindb-setup") + try: + manager.get_profile(identity.session_id) + except Exception as exc: + return False, f"Read smoke test failed: {exc}" + + try: + write_result = manager.remember( + identity.session_id, + "Hermes setup smoke test", + memory_type="factual", + importance=0.1, + metadata={"source": "hermes.retaindb.setup"}, + ) + memory_id = ( + write_result.get("memory", {}) or {} + ).get("id") or write_result.get("memory_id") or write_result.get("id") + if memory_id: + try: + manager.forget(str(memory_id)) + except Exception: + pass + except Exception as exc: + return False, f"Write smoke test failed: {exc}" + + return True, "Read/write smoke test passed." + + +def cmd_setup(args) -> None: + cfg = _load_hermes_config() + current = RetainDBClientConfig.from_global_config() + interactive = not bool(getattr(args, "yes", False)) + config_path = _config_path() + env_path = _env_path() + + print("\nRetainDB setup\n" + "-" * 40) + print(" Native RetainDB memory for Hermes.") + print(f" Config: {config_path}") + print(f" Env: {env_path}\n") + + api_key = str(getattr(args, "api_key", None) or current.api_key or "").strip() + if interactive and not api_key: + api_key = _prompt("RetainDB API key", secret=True) + if not api_key: + print(" No API key provided. Set RETAINDB_API_KEY or re-run with --api-key.\n") + return + + base_url = str(getattr(args, "base_url", None) or current.base_url or DEFAULT_BASE_URL).strip() or DEFAULT_BASE_URL + if interactive and not getattr(args, "base_url", None): + base_url = _prompt("RetainDB base URL", default=base_url) + + test_config = RetainDBClientConfig.from_global_config() + test_config.api_key = api_key + test_config.base_url = base_url + client = RetainDBClient(test_config) + + print(" Validating API key... ", end="", flush=True) + try: + client.validate_api_key() + except Exception as exc: + print("FAILED") + print(f" {exc}\n") + return + print("OK") + + try: + project = _select_or_create_project(client, args, interactive=interactive) + except Exception as exc: + print(f" {exc}\n") + return + + _write_runtime_config(cfg, project=project, base_url=base_url) + _save_env(api_key, base_url) + + final_config = RetainDBClientConfig.from_global_config() + final_config.api_key = api_key + final_config.base_url = base_url + final_config.project = project + + ok, message = _smoke_test(RetainDBClient(final_config), final_config) + print(f"\n Config written to {config_path}") + print(f" Env written to {env_path}") + print(f" Project: {project}") + print(f" Base URL: {base_url}") + print(f" Status: {'ready' if ok else 'not ready'}") + print(f" {message}\n") + + +def cmd_status(args) -> None: + config = RetainDBClientConfig.from_global_config() + api_key = config.api_key or "" + masked = f"...{api_key[-8:]}" if len(api_key) > 8 else ("set" if api_key else "not set") + print("\nRetainDB status\n" + "-" * 40) + print(f" Enabled: {config.enabled}") + print(f" API key: {masked}") + print(f" Base URL: {config.base_url}") + print(f" Project: {config.project or 'not set'}") + print(f" Memory mode: {config.memory_mode}") + print(f" Recall mode: {config.recall_mode}") + print(f" Write freq: {config.write_frequency}") + print(f" Context tokens: {config.context_tokens}") + print(f" Tool exposure: {not config.disable_tool_exposure}") + + if not config.should_activate(): + print("\n Not connected (missing enabled/project/api key).\n") + return + + status = RetainDBSessionManager(config=config).connection_status() + if status.get("ok"): + print(f"\n Connection... OK ({len(status.get('projects') or [])} project(s) visible)\n") + else: + print(f"\n Connection... FAILED ({status.get('error')})\n") + + +def cmd_test(args) -> None: + config = RetainDBClientConfig.from_global_config() + if not config.should_activate(): + print(" RetainDB is not configured. Run 'hermes retaindb setup' first.\n") + return + ok, message = _smoke_test(RetainDBClient(config), config) + print(f" {'PASS' if ok else 'FAIL'}: {message}\n") + + +def cmd_mode(args) -> None: + cfg = _load_hermes_config() + current = str((cfg.get("retaindb", {}) or {}).get("memory_mode") or "hybrid") + mode = getattr(args, "mode", None) + if not mode: + print("\nRetainDB memory mode\n" + "-" * 40) + print(f" Current: {current}") + print(" Options: hybrid, retaindb\n") + return + if mode not in {"hybrid", "retaindb"}: + print(" Invalid mode. Options: hybrid, retaindb\n") + return + cfg.setdefault("retaindb", {})["memory_mode"] = mode + _save_hermes_config(cfg) + print(f" Memory mode -> {mode}\n") + + +def cmd_tokens(args) -> None: + cfg = _load_hermes_config() + current = int((cfg.get("retaindb", {}) or {}).get("context_tokens") or 1200) + context_tokens = getattr(args, "context", None) + if context_tokens is None: + print("\nRetainDB token budget\n" + "-" * 40) + print(f" Context tokens: {current}\n") + return + cfg.setdefault("retaindb", {})["context_tokens"] = int(context_tokens) + _save_hermes_config(cfg) + print(f" Context tokens -> {int(context_tokens)}\n") + + +def cmd_identity(args) -> None: + config = RetainDBClientConfig.from_global_config() + session_id = getattr(args, "session_id", None) or "hermes-retaindb-identity" + runtime_identity = { + "platform": "cli", + "user_name": os.getenv("USER") or os.getenv("USERNAME") or "", + } + identity = RetainDBSessionManager(config=config, runtime_identity=runtime_identity).resolve_identity(session_id) + print("\nRetainDB identity\n" + "-" * 40) + print(f" user_id: {identity.user_id}") + print(f" session_id: {identity.session_id}") + print(f" agent_id: {identity.agent_id}") + print(f" project: {identity.project or 'not set'}") + print(f" source: {identity.source}") + print(f" cache: {config.identity_cache_path}\n") + + +def retaindb_command(args) -> None: + action = getattr(args, "retaindb_command", None) + if action == "setup": + cmd_setup(args) + elif action == "status": + cmd_status(args) + elif action == "test": + cmd_test(args) + elif action == "mode": + cmd_mode(args) + elif action == "tokens": + cmd_tokens(args) + elif action == "identity": + cmd_identity(args) + else: + print("Usage: hermes retaindb [setup|status|test|mode|tokens|identity]\n") diff --git a/retaindb_integration/client.py b/retaindb_integration/client.py new file mode 100644 index 000000000000..379c4d242e28 --- /dev/null +++ b/retaindb_integration/client.py @@ -0,0 +1,469 @@ +"""HTTP client + config resolution for the native RetainDB integration.""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import requests + +logger = logging.getLogger(__name__) + +DEFAULT_BASE_URL = "https://api.retaindb.com" +DEFAULT_PREFETCH_TIMEOUT_MS = 1500 +DEFAULT_CONTEXT_TOKENS = 1200 +DEFAULT_FLUSH_BATCH_SIZE = 50 +_VALID_MEMORY_MODES = {"hybrid", "retaindb", "local"} +_VALID_RECALL_MODES = {"hybrid", "context", "tools"} + + +def _get_hermes_home() -> Path: + return Path(os.getenv("HERMES_HOME", Path.home() / ".hermes")) + + +def normalize_base_url(url: str | None) -> str: + value = (url or DEFAULT_BASE_URL).strip() + if not value: + value = DEFAULT_BASE_URL + value = re.sub(r"/+$", "", value) + value = re.sub(r"/api/v1$", "", value, flags=re.IGNORECASE) + value = re.sub(r"/v1$", "", value, flags=re.IGNORECASE) + value = re.sub(r"/api$", "", value, flags=re.IGNORECASE) + return value + + +def _coerce_int(value: Any, default: int) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _coerce_bool(value: Any, default: bool) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +class RetainDBClientError(RuntimeError): + """Raised when RetainDB returns a non-success response.""" + + def __init__( + self, + message: str, + *, + status_code: int | None = None, + payload: Any = None, + ) -> None: + super().__init__(message) + self.status_code = status_code + self.payload = payload + + +@dataclass +class RetainDBClientConfig: + """Resolved config for Hermes' native RetainDB integration.""" + + enabled: bool = True + api_key: str | None = None + base_url: str = DEFAULT_BASE_URL + project: str | None = None + memory_mode: str = "hybrid" + recall_mode: str = "hybrid" + write_frequency: str | int = "async" + context_tokens: int = DEFAULT_CONTEXT_TOKENS + prefetch_timeout_ms: int = DEFAULT_PREFETCH_TIMEOUT_MS + flush_batch_size: int = DEFAULT_FLUSH_BATCH_SIZE + disable_tool_exposure: bool = False + debug_recall_trace: bool = False + user_id_override: str | None = None + agent_id: str = "hermes" + + @classmethod + def from_global_config(cls) -> "RetainDBClientConfig": + """Resolve RetainDB settings from Hermes config.yaml + ~/.hermes/.env.""" + try: + from hermes_cli.config import get_env_value, load_config + + cfg = load_config() or {} + raw = cfg.get("retaindb", {}) if isinstance(cfg, dict) else {} + env_api_key = get_env_value("RETAINDB_API_KEY") + env_base_url = get_env_value("RETAINDB_BASE_URL") + env_project = get_env_value("RETAINDB_PROJECT") + except Exception: + raw = {} + env_api_key = os.getenv("RETAINDB_API_KEY") + env_base_url = os.getenv("RETAINDB_BASE_URL") + env_project = os.getenv("RETAINDB_PROJECT") + + write_frequency = raw.get("write_frequency", "async") + try: + write_frequency = int(write_frequency) + except (TypeError, ValueError): + write_frequency = str(write_frequency or "async").strip().lower() or "async" + + memory_mode = str(raw.get("memory_mode", "hybrid") or "hybrid").strip().lower() + if memory_mode not in _VALID_MEMORY_MODES: + memory_mode = "hybrid" + + recall_mode = str(raw.get("recall_mode", "hybrid") or "hybrid").strip().lower() + if recall_mode not in _VALID_RECALL_MODES: + recall_mode = "hybrid" + + return cls( + enabled=_coerce_bool(raw.get("enabled"), True), + api_key=(env_api_key or "").strip() or None, + base_url=normalize_base_url(env_base_url or raw.get("base_url") or DEFAULT_BASE_URL), + project=(str(env_project or raw.get("project") or "").strip() or None), + memory_mode=memory_mode, + recall_mode=recall_mode, + write_frequency=write_frequency, + context_tokens=max(200, _coerce_int(raw.get("context_tokens"), DEFAULT_CONTEXT_TOKENS)), + prefetch_timeout_ms=max(100, _coerce_int(raw.get("prefetch_timeout_ms"), DEFAULT_PREFETCH_TIMEOUT_MS)), + flush_batch_size=max(1, _coerce_int(raw.get("flush_batch_size"), DEFAULT_FLUSH_BATCH_SIZE)), + disable_tool_exposure=_coerce_bool(raw.get("disable_tool_exposure"), False), + debug_recall_trace=_coerce_bool(raw.get("debug_recall_trace"), False), + user_id_override=(str(raw.get("user_id_override") or "").strip() or None), + agent_id=(str(raw.get("agent_id") or "hermes").strip() or "hermes"), + ) + + @property + def queue_db_path(self) -> Path: + return _get_hermes_home() / "retaindb_queue.db" + + @property + def identity_cache_path(self) -> Path: + return _get_hermes_home() / "retaindb_identity.json" + + def should_activate(self) -> bool: + return bool( + self.enabled + and self.api_key + and self.project + and self.memory_mode != "local" + ) + + +class RetainDBClient: + """Minimal requests-based client for Hermes' native RetainDB integration.""" + + def __init__(self, config: RetainDBClientConfig): + self.config = config + self.base_url = normalize_base_url(config.base_url) + self.api_key = (config.api_key or "").strip() + + def _headers(self, endpoint: str) -> dict[str, str]: + attach_api_key_header = endpoint.startswith("/v1/memory") or endpoint.startswith("/v1/context/query") + token = self.api_key.replace("Bearer ", "").strip() + headers = { + "Content-Type": "application/json", + "Authorization": self.api_key if self.api_key.startswith("Bearer ") else f"Bearer {self.api_key}", + "x-sdk-runtime": "hermes-agent", + "x-sdk-version": "native-retaindb-v1", + } + if attach_api_key_header and token: + headers["X-API-Key"] = token + return headers + + def _request( + self, + method: str, + endpoint: str, + *, + params: dict[str, Any] | None = None, + json_body: dict[str, Any] | None = None, + timeout_ms: int | None = None, + ) -> Any: + if not self.api_key: + raise RetainDBClientError("RETAINDB_API_KEY is not configured.") + + url = f"{self.base_url}{endpoint}" + timeout = max(0.2, float(timeout_ms or self.config.prefetch_timeout_ms) / 1000.0) + response = requests.request( + method=method.upper(), + url=url, + params=params, + json=json_body if method.upper() not in {"GET", "DELETE"} else None, + headers=self._headers(endpoint), + timeout=timeout, + ) + + payload: Any + try: + payload = response.json() + except ValueError: + payload = response.text + + if response.ok: + return payload + + message = "" + if isinstance(payload, dict): + message = ( + str(payload.get("message") or "") + or str(payload.get("error") or "") + or str(payload.get("detail") or "") + ).strip() + elif isinstance(payload, str): + message = payload.strip() + + if not message: + message = f"RetainDB request failed with HTTP {response.status_code}" + + if response.status_code in {401, 403}: + message = f"RetainDB authentication failed: {message}" + elif response.status_code == 404 and "project" not in message.lower(): + message = f"RetainDB endpoint not found at {url}" + + raise RetainDBClientError(message, status_code=response.status_code, payload=payload) + + def validate_api_key(self) -> list[dict[str, Any]]: + response = self._request("GET", "/v1/projects", timeout_ms=10000) + return list((response or {}).get("projects") or []) + + def list_projects(self) -> list[dict[str, Any]]: + response = self._request("GET", "/v1/projects", timeout_ms=10000) + return list((response or {}).get("projects") or []) + + def create_project(self, name: str) -> dict[str, Any]: + return self._request( + "POST", + "/v1/projects", + json_body={"name": name}, + timeout_ms=10000, + ) + + def query( + self, + *, + query: str, + project: str | None = None, + user_id: str | None = None, + session_id: str | None = None, + agent_id: str | None = None, + include_memories: bool = True, + max_tokens: int | None = None, + top_k: int = 6, + include_graph: bool = False, + include_parent_content: bool = False, + retrieval_profile: str = "precision_v1", + timeout_ms: int | None = None, + ) -> dict[str, Any]: + project_ref = project or self.config.project + if not project_ref: + raise RetainDBClientError("RetainDB project is not configured.") + return self._request( + "POST", + "/v1/context/query", + json_body={ + "project": project_ref, + "query": query, + "user_id": user_id, + "session_id": session_id, + "agent_id": agent_id, + "include_memories": include_memories, + "top_k": top_k, + "include_graph": include_graph, + "include_parent_content": include_parent_content, + "max_tokens": max_tokens or self.config.context_tokens, + "retrieval_profile": retrieval_profile, + }, + timeout_ms=timeout_ms, + ) + + def search_memories( + self, + *, + query: str, + project: str | None = None, + user_id: str | None = None, + session_id: str | None = None, + agent_id: str | None = None, + memory_type: str | None = None, + top_k: int = 8, + include_pending: bool = True, + profile: str = "balanced", + timeout_ms: int | None = None, + ) -> dict[str, Any]: + project_ref = project or self.config.project + if not project_ref: + raise RetainDBClientError("RetainDB project is not configured.") + + body = { + "project": project_ref, + "query": query, + "user_id": user_id, + "session_id": session_id, + "agent_id": agent_id, + "top_k": top_k, + "profile": profile, + "include_pending": include_pending, + "memory_types": [memory_type] if memory_type else None, + } + try: + return self._request( + "POST", + "/v1/memory/search", + json_body=body, + timeout_ms=timeout_ms, + ) + except RetainDBClientError as exc: + if exc.status_code != 404: + raise + legacy_body = { + "project": project_ref, + "query": query, + "user_id": user_id, + "session_id": session_id, + "agent_id": agent_id, + "top_k": top_k, + "memory_type": memory_type, + } + return self._request( + "POST", + "/v1/memories/search", + json_body=legacy_body, + timeout_ms=timeout_ms, + ) + + def get_user_profile( + self, + *, + user_id: str, + project: str | None = None, + include_pending: bool = True, + memory_types: str | None = None, + timeout_ms: int | None = None, + ) -> dict[str, Any]: + project_ref = project or self.config.project + if not project_ref: + raise RetainDBClientError("RetainDB project is not configured.") + params = {"project": project_ref, "include_pending": str(include_pending).lower()} + if memory_types: + params["memory_types"] = memory_types + try: + return self._request( + "GET", + f"/v1/memory/profile/{quote(user_id, safe='')}", + params=params, + timeout_ms=timeout_ms, + ) + except RetainDBClientError as exc: + if exc.status_code != 404: + raise + legacy = self._request( + "GET", + "/v1/memories", + params={ + "project": project_ref, + "user_id": user_id, + "limit": "200", + }, + timeout_ms=timeout_ms, + ) + memories = list((legacy or {}).get("memories") or []) + return {"user_id": user_id, "memories": memories, "count": len(memories)} + + def add_memory( + self, + *, + content: str, + project: str | None = None, + memory_type: str = "factual", + user_id: str | None = None, + session_id: str | None = None, + agent_id: str | None = None, + importance: float | None = None, + metadata: dict[str, Any] | None = None, + write_mode: str = "sync", + timeout_ms: int | None = None, + ) -> dict[str, Any]: + project_ref = project or self.config.project + if not project_ref: + raise RetainDBClientError("RetainDB project is not configured.") + try: + return self._request( + "POST", + "/v1/memory", + json_body={ + "project": project_ref, + "content": content, + "memory_type": memory_type, + "user_id": user_id, + "session_id": session_id, + "agent_id": agent_id, + "importance": importance, + "metadata": metadata or {}, + "write_mode": write_mode, + }, + timeout_ms=timeout_ms, + ) + except RetainDBClientError as exc: + if exc.status_code != 404: + raise + return self._request( + "POST", + "/v1/memories", + json_body={ + "project": project_ref, + "content": content, + "memory_type": memory_type, + "user_id": user_id, + "session_id": session_id, + "agent_id": agent_id, + "importance": importance, + "metadata": metadata or {}, + }, + timeout_ms=timeout_ms, + ) + + def ingest_session( + self, + *, + session_id: str, + messages: list[dict[str, Any]], + project: str | None = None, + user_id: str | None = None, + write_mode: str = "sync", + timeout_ms: int | None = None, + ) -> dict[str, Any]: + project_ref = project or self.config.project + if not project_ref: + raise RetainDBClientError("RetainDB project is not configured.") + payload = { + "project": project_ref, + "session_id": session_id, + "user_id": user_id, + "messages": messages, + "write_mode": write_mode, + } + return self._request( + "POST", + "/v1/memory/ingest/session", + json_body=payload, + timeout_ms=timeout_ms or 10000, + ) + + def delete_memory(self, memory_id: str, *, timeout_ms: int | None = None) -> dict[str, Any]: + try: + return self._request( + "DELETE", + f"/v1/memory/{quote(memory_id, safe='')}", + timeout_ms=timeout_ms or 5000, + ) + except RetainDBClientError as exc: + if exc.status_code != 404: + raise + self._request( + "DELETE", + f"/v1/memories/{quote(memory_id, safe='')}", + timeout_ms=timeout_ms or 5000, + ) + return {"success": True, "deleted": memory_id} diff --git a/retaindb_integration/identity.py b/retaindb_integration/identity.py new file mode 100644 index 000000000000..429ec89db3de --- /dev/null +++ b/retaindb_integration/identity.py @@ -0,0 +1,134 @@ +"""Identity resolution and persistence for Hermes' RetainDB integration.""" + +from __future__ import annotations + +import getpass +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any + +from retaindb_integration.client import RetainDBClientConfig + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _sanitize_fallback(value: str | None) -> str: + text = (value or "").strip() + if not text: + return "hermes-user" + cleaned = [] + for char in text: + if char.isalnum() or char in {"-", "_", ".", ":"}: + cleaned.append(char) + elif char.isspace(): + cleaned.append("-") + collapsed = "".join(cleaned).strip("-") + while "--" in collapsed: + collapsed = collapsed.replace("--", "-") + return collapsed or "hermes-user" + + +@dataclass +class ResolvedRetainDBIdentity: + user_id: str + session_id: str + agent_id: str + project: str + source: str + peer_name: str | None = None + platform: str | None = None + chat_id: str | None = None + + +class RetainDBIdentityResolver: + """Resolve and persist a stable user identity for RetainDB lookups.""" + + def __init__(self, config: RetainDBClientConfig): + self.config = config + self.cache_path = config.identity_cache_path + + def _read_cache(self) -> dict[str, Any]: + try: + return json.loads(self.cache_path.read_text(encoding="utf-8")) + except Exception: + return {} + + def _write_cache(self, payload: dict[str, Any]) -> None: + self.cache_path.parent.mkdir(parents=True, exist_ok=True) + self.cache_path.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + def resolve( + self, + *, + session_id: str, + runtime_identity: dict[str, Any] | None = None, + ) -> ResolvedRetainDBIdentity: + runtime = dict(runtime_identity or {}) + cached = self._read_cache() + + explicit_override = ( + str(runtime.get("user_id_override") or "").strip() + or str(self.config.user_id_override or "").strip() + ) + platform_user_id = str( + runtime.get("platform_user_id") + or runtime.get("user_id") + or "" + ).strip() + peer_name = str( + runtime.get("peer_name") + or runtime.get("user_name") + or "" + ).strip() or None + os_username = str(runtime.get("os_username") or getpass.getuser() or "").strip() + + if explicit_override: + resolved_user_id = explicit_override + source = "config_override" + elif platform_user_id: + resolved_user_id = platform_user_id + source = "platform" + elif peer_name: + if cached.get("source") in {"peer", "os"} and cached.get("user_id"): + resolved_user_id = str(cached["user_id"]) + else: + resolved_user_id = _sanitize_fallback(peer_name) + source = "peer" + else: + if cached.get("source") in {"peer", "os"} and cached.get("user_id"): + resolved_user_id = str(cached["user_id"]) + else: + resolved_user_id = _sanitize_fallback(os_username) + source = "os" + + result = ResolvedRetainDBIdentity( + user_id=resolved_user_id, + session_id=session_id, + agent_id=str(runtime.get("agent_id") or self.config.agent_id or "hermes"), + project=str(runtime.get("project") or self.config.project or "").strip(), + source=source, + peer_name=peer_name, + platform=str(runtime.get("platform") or "").strip() or None, + chat_id=str(runtime.get("chat_id") or "").strip() or None, + ) + + self._write_cache( + { + "user_id": result.user_id, + "session_id": result.session_id, + "agent_id": result.agent_id, + "project": result.project, + "source": result.source, + "peer_name": result.peer_name, + "platform": result.platform, + "chat_id": result.chat_id, + "updated_at": _now_iso(), + } + ) + return result diff --git a/retaindb_integration/recall.py b/retaindb_integration/recall.py new file mode 100644 index 000000000000..c1c7173e7b42 --- /dev/null +++ b/retaindb_integration/recall.py @@ -0,0 +1,134 @@ +"""Formatting and dedupe helpers for RetainDB turn-time recall.""" + +from __future__ import annotations + +import re +from typing import Any + + +def _compact_text(value: str | None) -> str: + text = re.sub(r"\s+", " ", str(value or "")).strip() + return text[:320].rstrip() + + +def _normalize_text(value: str | None) -> str: + text = _compact_text(value).lower() + text = re.sub(r"[^a-z0-9]+", " ", text) + return re.sub(r"\s+", " ", text).strip() + + +def _is_duplicate(candidate_norm: str, corpus: list[str]) -> bool: + if not candidate_norm: + return True + for existing in corpus: + if not existing: + continue + if candidate_norm == existing: + return True + if len(candidate_norm) > 18 and candidate_norm in existing: + return True + if len(existing) > 18 and existing in candidate_norm: + return True + return False + + +def _dedupe_items(items: list[str], corpus: list[str], max_items: int) -> list[str]: + deduped: list[str] = [] + for item in items: + compact = _compact_text(item) + norm = _normalize_text(compact) + if not norm or _is_duplicate(norm, corpus): + continue + corpus.append(norm) + deduped.append(compact) + if len(deduped) >= max_items: + break + return deduped + + +def _extract_profile_items(profile: dict[str, Any] | None) -> list[str]: + memories = list((profile or {}).get("memories") or []) + return [ + _compact_text( + (memory or {}).get("content") + or (memory or {}).get("memory", {}).get("content") + ) + for memory in memories + if _compact_text( + (memory or {}).get("content") + or (memory or {}).get("memory", {}).get("content") + ) + ] + + +def _extract_query_items(query_result: dict[str, Any] | None) -> list[str]: + items: list[str] = [] + for result in list((query_result or {}).get("results") or []): + content = _compact_text((result or {}).get("content")) + if content: + items.append(content) + + if items: + return items + + context = _compact_text((query_result or {}).get("context")) + if not context: + return [] + return [ + segment.strip() + for segment in re.split(r"(?<=[.!?])\s+", context) + if segment.strip() + ] + + +def _extract_update_items(profile: dict[str, Any] | None, query_result: dict[str, Any] | None) -> list[str]: + candidates = _extract_profile_items(profile) + _extract_query_items(query_result) + update_markers = ( + "correct", + "changed", + "no longer", + "instead", + "prefer", + "updated", + "switch", + "moved", + "now uses", + ) + return [ + item + for item in candidates + if any(marker in item.lower() for marker in update_markers) + ] + + +def build_retaindb_overlay( + *, + profile: dict[str, Any] | None, + query_result: dict[str, Any] | None, + local_entries: list[str] | None = None, + recent_texts: list[str] | None = None, + max_profile_items: int = 5, + max_memory_items: int = 5, + max_update_items: int = 3, +) -> str: + """Build the strict RetainDB overlay block for a single turn.""" + + corpus = [ + _normalize_text(item) + for item in (local_entries or []) + (recent_texts or []) + if _normalize_text(item) + ] + profile_items = _dedupe_items(_extract_profile_items(profile), corpus, max_profile_items) + relevant_items = _dedupe_items(_extract_query_items(query_result), corpus, max_memory_items) + update_items = _dedupe_items(_extract_update_items(profile, query_result), corpus, max_update_items) + + if not profile_items and not relevant_items and not update_items: + return "" + + lines = ["[RetainDB Context]", "Profile:"] + lines.extend(f"- {item}" for item in (profile_items or ["None"])) + lines.append("Relevant memories:") + lines.extend(f"- {item}" for item in (relevant_items or ["None"])) + lines.append("Open corrections / recent updates:") + lines.extend(f"- {item}" for item in (update_items or ["None"])) + return "\n".join(lines) diff --git a/retaindb_integration/session.py b/retaindb_integration/session.py new file mode 100644 index 000000000000..87dcc2edfcfe --- /dev/null +++ b/retaindb_integration/session.py @@ -0,0 +1,272 @@ +"""Runtime session manager for Hermes' native RetainDB integration.""" + +from __future__ import annotations + +import hashlib +import json +import logging +import threading +from datetime import datetime, timezone +from typing import Any + +from retaindb_integration.client import RetainDBClient, RetainDBClientConfig +from retaindb_integration.identity import ResolvedRetainDBIdentity, RetainDBIdentityResolver +from retaindb_integration.recall import build_retaindb_overlay +from retaindb_integration.write_queue import DurableRetainDBWriteQueue + +logger = logging.getLogger(__name__) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +class RetainDBSessionManager: + """Turn-time recall + durable write-behind manager for a Hermes session.""" + + def __init__( + self, + client: RetainDBClient | None = None, + config: RetainDBClientConfig | None = None, + runtime_identity: dict[str, Any] | None = None, + ): + self._config = config or RetainDBClientConfig.from_global_config() + self._client = client or RetainDBClient(self._config) + self._runtime_identity = dict(runtime_identity or {}) + self._identity_resolver = RetainDBIdentityResolver(self._config) + self._write_queue = DurableRetainDBWriteQueue(self._client, self._config) + self._prefetch_cache: dict[str, dict[str, Any]] = {} + self._prefetch_lock = threading.Lock() + + @property + def config(self) -> RetainDBClientConfig: + return self._config + + def set_runtime_identity(self, runtime_identity: dict[str, Any] | None) -> None: + self._runtime_identity = dict(runtime_identity or {}) + + def resolve_identity(self, session_id: str) -> ResolvedRetainDBIdentity: + runtime = dict(self._runtime_identity) + runtime.setdefault("session_id", session_id) + return self._identity_resolver.resolve( + session_id=session_id, + runtime_identity=runtime, + ) + + def connection_status(self) -> dict[str, Any]: + try: + projects = self._client.list_projects() + return {"ok": True, "projects": projects} + except Exception as exc: + return {"ok": False, "error": str(exc)} + + def get_profile(self, session_id: str) -> dict[str, Any]: + identity = self.resolve_identity(session_id) + return self._client.get_user_profile( + project=identity.project, + user_id=identity.user_id, + include_pending=True, + timeout_ms=self._config.prefetch_timeout_ms, + ) + + def search(self, session_id: str, query: str, *, top_k: int = 8) -> dict[str, Any]: + identity = self.resolve_identity(session_id) + return self._client.search_memories( + project=identity.project, + query=query, + user_id=identity.user_id, + session_id=identity.session_id, + agent_id=identity.agent_id, + top_k=top_k, + include_pending=True, + timeout_ms=self._config.prefetch_timeout_ms, + ) + + def remember( + self, + session_id: str, + content: str, + *, + memory_type: str = "factual", + importance: float = 0.6, + metadata: dict[str, Any] | None = None, + ) -> dict[str, Any]: + identity = self.resolve_identity(session_id) + return self._client.add_memory( + project=identity.project, + content=content, + memory_type=memory_type, + user_id=identity.user_id, + session_id=identity.session_id, + agent_id=identity.agent_id, + importance=importance, + metadata=metadata or {}, + write_mode="sync", + timeout_ms=max(2500, self._config.prefetch_timeout_ms), + ) + + def forget(self, memory_id: str) -> dict[str, Any]: + return self._client.delete_memory(memory_id) + + def get_context( + self, + session_id: str, + query: str, + *, + local_entries: list[str] | None = None, + recent_texts: list[str] | None = None, + ) -> dict[str, Any]: + identity = self.resolve_identity(session_id) + query_result = self._client.query( + project=identity.project, + query=query, + user_id=identity.user_id, + session_id=identity.session_id, + agent_id=identity.agent_id, + include_memories=True, + max_tokens=self._config.context_tokens, + timeout_ms=self._config.prefetch_timeout_ms, + ) + profile = self._client.get_user_profile( + project=identity.project, + user_id=identity.user_id, + include_pending=True, + timeout_ms=self._config.prefetch_timeout_ms, + ) + overlay = build_retaindb_overlay( + profile=profile, + query_result=query_result, + local_entries=local_entries, + recent_texts=recent_texts, + ) + return { + "identity": identity, + "profile": profile, + "query": query_result, + "context": overlay, + } + + def _prefetch_worker( + self, + session_id: str, + query: str, + local_entries: list[str] | None, + recent_texts: list[str] | None, + generation: str, + ) -> None: + result_payload: dict[str, Any] + try: + result_payload = self.get_context( + session_id, + query, + local_entries=local_entries, + recent_texts=recent_texts, + ) + except Exception as exc: + logger.debug("RetainDB prefetch failed for %s: %s", session_id, exc) + result_payload = {"context": "", "error": str(exc)} + + with self._prefetch_lock: + cache = self._prefetch_cache.get(session_id) + if not cache or cache.get("generation") != generation: + return + cache["result"] = result_payload.get("context", "") + cache["details"] = result_payload + cache["event"].set() + + if self._config.debug_recall_trace and result_payload.get("context"): + logger.info( + "RetainDB recall trace session=%s profile_memories=%s query_results=%s", + session_id, + len((result_payload.get("profile") or {}).get("memories") or []), + len((result_payload.get("query") or {}).get("results") or []), + ) + + def prefetch_context( + self, + session_id: str, + query: str, + *, + local_entries: list[str] | None = None, + recent_texts: list[str] | None = None, + ) -> None: + generation = hashlib.sha1( + f"{session_id}:{query}:{_now_iso()}".encode("utf-8") + ).hexdigest() + event = threading.Event() + with self._prefetch_lock: + self._prefetch_cache[session_id] = { + "generation": generation, + "event": event, + "result": "", + "details": {}, + } + thread = threading.Thread( + target=self._prefetch_worker, + args=(session_id, query, local_entries, recent_texts, generation), + name=f"retaindb-prefetch-{session_id}", + daemon=True, + ) + thread.start() + + def pop_context_result(self, session_id: str, *, wait_ms: int | None = None) -> str: + with self._prefetch_lock: + cache = self._prefetch_cache.get(session_id) + if not cache: + return "" + event: threading.Event = cache["event"] + timeout = max(0.0, float(wait_ms or self._config.prefetch_timeout_ms) / 1000.0) + if not event.wait(timeout=timeout): + with self._prefetch_lock: + self._prefetch_cache.pop(session_id, None) + return "" + with self._prefetch_lock: + final = self._prefetch_cache.pop(session_id, None) or {} + return str(final.get("result") or "") + + def enqueue_turn( + self, + session_id: str, + user_content: str, + assistant_content: str, + *, + message_index: int, + turn_id: str, + ) -> None: + identity = self.resolve_identity(session_id) + messages = [ + {"role": "user", "content": user_content, "timestamp": _now_iso()}, + {"role": "assistant", "content": assistant_content, "timestamp": _now_iso()}, + ] + payload_checksum = hashlib.sha1( + json.dumps(messages, ensure_ascii=False, sort_keys=True).encode("utf-8") + ).hexdigest() + self._write_queue.enqueue( + identity, + turn_id=turn_id, + message_index=message_index, + payload_checksum=payload_checksum, + messages=messages, + ) + + if self._config.write_frequency == "turn": + self.flush_session(session_id) + + def save_user_observation(self, session_id: str, content: str) -> dict[str, Any]: + return self.remember( + session_id, + content, + memory_type="factual", + importance=0.7, + metadata={"source": "hermes.memory_tool"}, + ) + + def flush_session(self, session_id: str) -> None: + identity = self.resolve_identity(session_id) + self._write_queue.flush_session(identity) + + def flush_all(self) -> None: + self._write_queue.flush_all() + + def shutdown(self) -> None: + self._write_queue.shutdown() diff --git a/retaindb_integration/write_queue.py b/retaindb_integration/write_queue.py new file mode 100644 index 000000000000..5c1089c62120 --- /dev/null +++ b/retaindb_integration/write_queue.py @@ -0,0 +1,245 @@ +"""Durable async write-behind queue for the native RetainDB integration.""" + +from __future__ import annotations + +import json +import logging +import queue +import sqlite3 +import threading +import time +from datetime import datetime, timezone +from typing import Any + +from retaindb_integration.client import RetainDBClient, RetainDBClientConfig +from retaindb_integration.identity import ResolvedRetainDBIdentity + +logger = logging.getLogger(__name__) + +_ASYNC_SHUTDOWN = object() + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +class DurableRetainDBWriteQueue: + """Crash-safe local spool for background RetainDB session ingestion.""" + + def __init__(self, client: RetainDBClient, config: RetainDBClientConfig): + self._client = client + self._config = config + self._db_path = config.queue_db_path + self._identity_by_session: dict[str, ResolvedRetainDBIdentity] = {} + self._queue: queue.Queue | None = None + self._thread: threading.Thread | None = None + + self._init_db() + + if config.write_frequency == "async": + self._queue = queue.Queue() + self._thread = threading.Thread( + target=self._writer_loop, + name="retaindb-async-writer", + daemon=True, + ) + self._thread.start() + for session_id in self.pending_session_ids(): + self._queue.put(session_id) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self._db_path, timeout=30) + conn.row_factory = sqlite3.Row + return conn + + def _init_db(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS pending_ingest ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + user_id TEXT, + agent_id TEXT, + project TEXT NOT NULL, + turn_id TEXT NOT NULL, + message_index INTEGER NOT NULL, + payload_checksum TEXT NOT NULL, + messages_json TEXT NOT NULL, + created_at TEXT NOT NULL, + last_error TEXT + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS flush_state ( + session_id TEXT PRIMARY KEY, + last_flushed_message_index INTEGER, + last_flushed_turn_id TEXT, + payload_checksum TEXT, + updated_at TEXT NOT NULL + ) + """ + ) + conn.commit() + + def _mark_error(self, row_ids: list[int], message: str) -> None: + if not row_ids: + return + with self._connect() as conn: + conn.executemany( + "UPDATE pending_ingest SET last_error = ? WHERE id = ?", + [(message, row_id) for row_id in row_ids], + ) + conn.commit() + + def enqueue( + self, + identity: ResolvedRetainDBIdentity, + *, + turn_id: str, + message_index: int, + payload_checksum: str, + messages: list[dict[str, Any]], + ) -> None: + self._identity_by_session[identity.session_id] = identity + with self._connect() as conn: + conn.execute( + """ + INSERT INTO pending_ingest ( + session_id, user_id, agent_id, project, turn_id, + message_index, payload_checksum, messages_json, created_at, last_error + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL) + """, + ( + identity.session_id, + identity.user_id, + identity.agent_id, + identity.project, + turn_id, + message_index, + payload_checksum, + json.dumps(messages, ensure_ascii=False), + _now_iso(), + ), + ) + conn.commit() + + if self._config.write_frequency == "async" and self._queue is not None: + self._queue.put(identity.session_id) + + def pending_session_ids(self) -> list[str]: + with self._connect() as conn: + rows = conn.execute( + "SELECT DISTINCT session_id FROM pending_ingest ORDER BY id ASC" + ).fetchall() + return [str(row["session_id"]) for row in rows] + + def _load_rows(self, session_id: str) -> list[sqlite3.Row]: + with self._connect() as conn: + return conn.execute( + """ + SELECT * FROM pending_ingest + WHERE session_id = ? + ORDER BY id ASC + LIMIT ? + """, + (session_id, int(self._config.flush_batch_size)), + ).fetchall() + + def flush_session(self, identity: ResolvedRetainDBIdentity) -> bool: + rows = self._load_rows(identity.session_id) + if not rows: + return True + + row_ids = [int(row["id"]) for row in rows] + message_index = max(int(row["message_index"]) for row in rows) + last_turn_id = str(rows[-1]["turn_id"]) + payload_checksum = str(rows[-1]["payload_checksum"]) + messages: list[dict[str, Any]] = [] + for row in rows: + messages.extend(json.loads(str(row["messages_json"]))) + + try: + self._client.ingest_session( + project=identity.project, + session_id=identity.session_id, + user_id=identity.user_id, + messages=messages, + write_mode="sync", + timeout_ms=max(4000, self._config.prefetch_timeout_ms * 4), + ) + except Exception as exc: + self._mark_error(row_ids, str(exc)) + logger.warning("RetainDB ingest failed for %s: %s", identity.session_id, exc) + return False + + with self._connect() as conn: + conn.executemany( + "DELETE FROM pending_ingest WHERE id = ?", + [(row_id,) for row_id in row_ids], + ) + conn.execute( + """ + INSERT INTO flush_state ( + session_id, last_flushed_message_index, last_flushed_turn_id, + payload_checksum, updated_at + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + last_flushed_message_index = excluded.last_flushed_message_index, + last_flushed_turn_id = excluded.last_flushed_turn_id, + payload_checksum = excluded.payload_checksum, + updated_at = excluded.updated_at + """, + ( + identity.session_id, + message_index, + last_turn_id, + payload_checksum, + _now_iso(), + ), + ) + conn.commit() + + return True + + def flush_all(self) -> None: + for session_id in self.pending_session_ids(): + identity = self._identity_by_session.get(session_id) + if identity is None: + continue + self.flush_session(identity) + + def get_flush_state(self, session_id: str) -> dict[str, Any]: + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM flush_state WHERE session_id = ?", + (session_id,), + ).fetchone() + return dict(row) if row is not None else {} + + def _writer_loop(self) -> None: + while True: + try: + item = self._queue.get(timeout=5) + if item is _ASYNC_SHUTDOWN: + break + session_id = str(item) + identity = self._identity_by_session.get(session_id) + if identity is None: + continue + success = self.flush_session(identity) + if not success: + time.sleep(2) + except queue.Empty: + continue + except Exception as exc: + logger.error("RetainDB async writer error: %s", exc) + + def shutdown(self) -> None: + self.flush_all() + if self._queue is not None and self._thread is not None: + self._queue.put(_ASYNC_SHUTDOWN) + self._thread.join(timeout=10) diff --git a/run_agent.py b/run_agent.py index 7c8d9208b135..e9060d218180 100644 --- a/run_agent.py +++ b/run_agent.py @@ -106,6 +106,14 @@ "honcho_conclude", } +RETAINDB_TOOL_NAMES = { + "retaindb_profile", + "retaindb_search", + "retaindb_context", + "retaindb_remember", + "retaindb_forget", +} + class _SafeWriter: """Transparent stdio wrapper that catches OSError/ValueError from broken pipes. @@ -354,6 +362,27 @@ def _inject_honcho_turn_context(content, turn_context: str): return f"{text}\n\n{note}" +def _inject_retaindb_turn_context(content, turn_context: str): + """Append RetainDB recall to the current-turn user message without mutating history.""" + if not turn_context: + return content + + note = ( + "[System note: The following RetainDB memory was retrieved from prior " + "sessions. It is continuity context for this turn only, not new user " + "input.]\n\n" + f"{turn_context}" + ) + + if isinstance(content, list): + return list(content) + [{"type": "text", "text": note}] + + text = "" if content is None else str(content) + if not text.strip(): + return note + return f"{text}\n\n{note}" + + class AIAgent: """ AI Agent with tool calling capabilities. @@ -416,6 +445,10 @@ def __init__( honcho_session_key: str = None, honcho_manager=None, honcho_config=None, + retaindb_session_key: str = None, + retaindb_manager=None, + retaindb_config=None, + retaindb_identity: Dict[str, Any] = None, iteration_budget: "IterationBudget" = None, fallback_model: Dict[str, Any] = None, checkpoints_enabled: bool = False, @@ -464,6 +497,10 @@ def __init__( When provided and Honcho is enabled in config, enables persistent cross-session user modeling. honcho_manager: Optional shared HonchoSessionManager owned by the caller. honcho_config: Optional HonchoClientConfig corresponding to honcho_manager. + retaindb_session_key (str): Session id for RetainDB recall + write-behind. + retaindb_manager: Optional shared RetainDBSessionManager owned by the caller. + retaindb_config: Optional RetainDBClientConfig corresponding to retaindb_manager. + retaindb_identity (Dict): Optional runtime identity hints for user_id resolution. """ _install_safe_stdio() @@ -982,6 +1019,58 @@ def __init__( if not self._honcho: self._strip_honcho_tools_from_surface() + # Native RetainDB deep memory (cross-session semantic recall) + self._retaindb = None + self._retaindb_session_key = retaindb_session_key + self._retaindb_config = None + self._retaindb_exit_hook_registered = False + self._retaindb_runtime_identity = dict(retaindb_identity or {}) + if not skip_memory: + try: + if retaindb_manager is not None: + rcfg = retaindb_config or getattr(retaindb_manager, "_config", None) or getattr(retaindb_manager, "config", None) + self._retaindb_config = rcfg + if rcfg and self._retaindb_should_activate(rcfg): + self._retaindb = retaindb_manager + self._activate_retaindb( + rcfg, + enabled_toolsets=enabled_toolsets, + disabled_toolsets=disabled_toolsets, + ) + else: + from retaindb_integration.client import RetainDBClientConfig + from retaindb_integration.session import RetainDBSessionManager + + rcfg = RetainDBClientConfig.from_global_config() + self._retaindb_config = rcfg + if self._retaindb_should_activate(rcfg): + self._retaindb = RetainDBSessionManager( + config=rcfg, + runtime_identity=self._retaindb_runtime_identity, + ) + self._activate_retaindb( + rcfg, + enabled_toolsets=enabled_toolsets, + disabled_toolsets=disabled_toolsets, + ) + else: + if not rcfg.enabled: + logger.debug("RetainDB disabled in config") + elif not rcfg.project: + logger.debug("RetainDB enabled but no project configured") + elif not rcfg.api_key: + logger.debug("RetainDB enabled but no API key configured") + else: + logger.debug("RetainDB enabled but inactive due to incomplete config") + except Exception as e: + logger.warning("RetainDB init failed - memory disabled: %s", e) + print(f" RetainDB init failed: {e}") + print(" Run 'hermes retaindb setup' to reconfigure.") + self._retaindb = None + + if not self._retaindb: + self._strip_retaindb_tools_from_surface() + # Gate local memory writes based on per-peer memory modes. # AI peer governs MEMORY.md; user peer governs USER.md. # "honcho" = Honcho only, disable local writes. @@ -996,6 +1085,12 @@ def __init__( if _user_mode == "honcho": self._user_profile_enabled = False logger.debug("peer %s memory_mode=honcho: local USER.md writes disabled", _hcfg.peer_name or "user") + if self._retaindb_config and self._retaindb: + if self._retaindb_config.memory_mode == "retaindb": + self._memory_flush_min_turns = 0 + self._memory_enabled = False + self._user_profile_enabled = False + logger.debug("RetainDB memory_mode=retaindb: local MEMORY.md and USER.md writes disabled") # Skills config: nudge interval for skill creation reminders self._skill_nudge_interval = 10 @@ -2227,6 +2322,183 @@ def _honcho_sync(self, user_content: str, assistant_content: str) -> None: if not self.quiet_mode: print(f" Honcho write failed: {e}") + def _retaindb_should_activate(self, rcfg) -> bool: + """Return True when native RetainDB should be active.""" + return bool(rcfg and getattr(rcfg, "should_activate", lambda: False)()) + + def _strip_retaindb_tools_from_surface(self) -> None: + """Remove RetainDB tools from the active tool surface.""" + if not self.tools: + self.valid_tool_names = set() + return + + self.tools = [ + tool for tool in self.tools + if tool.get("function", {}).get("name") not in RETAINDB_TOOL_NAMES + ] + self.valid_tool_names = { + tool["function"]["name"] for tool in self.tools + } if self.tools else set() + + def _activate_retaindb( + self, + rcfg, + *, + enabled_toolsets: Optional[List[str]], + disabled_toolsets: Optional[List[str]], + ) -> None: + """Finish RetainDB setup once a session manager is available.""" + if not self._retaindb: + return + + if not self._retaindb_session_key: + self._retaindb_session_key = self.session_id or "hermes-default" + + self._retaindb_runtime_identity.setdefault("session_id", self._retaindb_session_key) + self._retaindb_runtime_identity.setdefault("platform", self.platform or "cli") + self._retaindb_runtime_identity.setdefault("agent_id", getattr(rcfg, "agent_id", "hermes")) + if hasattr(self._retaindb, "set_runtime_identity"): + self._retaindb.set_runtime_identity(self._retaindb_runtime_identity) + + from tools.retaindb_tools import set_session_context + + set_session_context(self._retaindb, self._retaindb_session_key) + + self.tools = get_tool_definitions( + enabled_toolsets=enabled_toolsets, + disabled_toolsets=disabled_toolsets, + quiet_mode=True, + ) + self.valid_tool_names = { + tool["function"]["name"] for tool in self.tools + } if self.tools else set() + + if rcfg.recall_mode == "context" or rcfg.disable_tool_exposure: + self._strip_retaindb_tools_from_surface() + if not self.quiet_mode: + print(" RetainDB active - recall_mode: context (RetainDB tools hidden)") + else: + if not self.quiet_mode: + print(f" RetainDB active - recall_mode: {rcfg.recall_mode}") + + logger.info( + "RetainDB active (session: %s, project: %s, write_frequency: %s, memory_mode: %s)", + self._retaindb_session_key, + rcfg.project, + rcfg.write_frequency, + rcfg.memory_mode, + ) + + self._register_retaindb_exit_hook() + + def _register_retaindb_exit_hook(self) -> None: + """Register a process-exit flush hook for RetainDB without clobbering signals.""" + if self._retaindb_exit_hook_registered or not self._retaindb: + return + + retaindb_ref = weakref.ref(self._retaindb) + + def _flush_retaindb_on_exit(): + manager = retaindb_ref() + if manager is None: + return + try: + manager.flush_all() + except Exception as exc: + logger.debug("RetainDB flush on exit failed (non-fatal): %s", exc) + + atexit.register(_flush_retaindb_on_exit) + self._retaindb_exit_hook_registered = True + + def _collect_retaindb_dedupe_inputs(self, messages: list[dict]) -> tuple[list[str], list[str]]: + """Collect local memory and recent transcript text for RetainDB dedupe.""" + local_entries: list[str] = [] + if self._memory_store: + local_entries.extend(getattr(self._memory_store, "memory_entries", []) or []) + local_entries.extend(getattr(self._memory_store, "user_entries", []) or []) + + recent_texts: list[str] = [] + for msg in messages[-12:]: + if msg.get("role") not in ("user", "assistant"): + continue + content = msg.get("content") + if not content: + continue + recent_texts.append(str(content)) + return local_entries, recent_texts + + def _queue_retaindb_prefetch(self, user_message: str, messages: list[dict]) -> None: + """Queue same-turn RetainDB prefetch so local prompt assembly can proceed in parallel.""" + if not self._retaindb or not self._retaindb_session_key: + return + + recall_mode = (self._retaindb_config.recall_mode if self._retaindb_config else "hybrid") + if recall_mode == "tools": + return + + try: + local_entries, recent_texts = self._collect_retaindb_dedupe_inputs(messages) + self._retaindb.prefetch_context( + self._retaindb_session_key, + user_message, + local_entries=local_entries, + recent_texts=recent_texts, + ) + except Exception as exc: + logger.debug("RetainDB background prefetch failed (non-fatal): %s", exc) + + def _retaindb_prefetch(self) -> str: + """Consume the same-turn RetainDB overlay if it completed within budget.""" + if not self._retaindb or not self._retaindb_session_key: + return "" + try: + wait_ms = getattr(self._retaindb_config, "prefetch_timeout_ms", 1500) + return self._retaindb.pop_context_result( + self._retaindb_session_key, + wait_ms=wait_ms, + ) + except Exception as exc: + logger.debug("RetainDB prefetch failed (non-fatal): %s", exc) + return "" + + def _retaindb_save_user_observation(self, content: str) -> str: + """Route a memory tool target=user add into RetainDB explicit memory.""" + if not content or not content.strip(): + return json.dumps({"success": False, "error": "Content cannot be empty."}) + try: + result = self._retaindb.save_user_observation( + self._retaindb_session_key, + content.strip(), + ) + return json.dumps({ + "success": True, + "target": "user", + "message": "Saved to RetainDB user memory.", + "result": result, + }) + except Exception as exc: + logger.debug("RetainDB user observation failed: %s", exc) + return json.dumps({"success": False, "error": f"RetainDB save failed: {exc}"}) + + def _retaindb_sync(self, user_content: str, assistant_content: str) -> None: + """Queue the current turn for durable RetainDB write-behind ingestion.""" + if not self._retaindb or not self._retaindb_session_key: + return + try: + turn_id = f"{self._retaindb_session_key}:{self._user_turn_count}:{int(time.time() * 1000)}" + self._retaindb.enqueue_turn( + self._retaindb_session_key, + user_content, + assistant_content, + message_index=self._user_turn_count, + turn_id=turn_id, + ) + logger.info("RetainDB sync queued for session %s", self._retaindb_session_key) + except Exception as exc: + logger.warning("RetainDB sync failed: %s", exc) + if not self.quiet_mode: + print(f" RetainDB write failed: {exc}") + def _build_system_prompt(self, system_message: str = None) -> str: """ Assemble the full system prompt from all layers. @@ -4523,6 +4795,8 @@ def flush_memories(self, messages: list = None, min_turns: int = None): ) if self._honcho and flush_target == "user" and args.get("action") == "add": self._honcho_save_user_observation(args.get("content", "")) + if self._retaindb and flush_target == "user" and args.get("action") == "add": + self._retaindb_save_user_observation(args.get("content", "")) if not self.quiet_mode: print(f" 🧠 Memory flush: saved to {args.get('target', 'memory')}") except Exception as e: @@ -4651,6 +4925,8 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i # Also send user observations to Honcho when active if self._honcho and target == "user" and function_args.get("action") == "add": self._honcho_save_user_observation(function_args.get("content", "")) + if self._retaindb and target == "user" and function_args.get("action") == "add": + self._retaindb_save_user_observation(function_args.get("content", "")) return result elif function_name == "clarify": from tools.clarify_tool import clarify_tool as _clarify_tool @@ -4675,6 +4951,8 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, honcho_manager=self._honcho, honcho_session_key=self._honcho_session_key, + retaindb_manager=self._retaindb, + retaindb_session_key=self._retaindb_session_key, ) def _execute_tool_calls_concurrent(self, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0) -> None: @@ -4982,6 +5260,8 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe # Also send user observations to Honcho when active if self._honcho and target == "user" and function_args.get("action") == "add": self._honcho_save_user_observation(function_args.get("content", "")) + if self._retaindb and target == "user" and function_args.get("action") == "add": + self._retaindb_save_user_observation(function_args.get("content", "")) tool_duration = time.time() - tool_start_time if self.quiet_mode: self._vprint(f" {_get_cute_tool_message_impl('memory', function_args, tool_duration, result=function_result)}") @@ -5043,6 +5323,8 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, honcho_manager=self._honcho, honcho_session_key=self._honcho_session_key, + retaindb_manager=self._retaindb, + retaindb_session_key=self._retaindb_session_key, ) _spinner_result = function_result except Exception as tool_error: @@ -5059,6 +5341,8 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe enabled_tools=list(self.valid_tool_names) if self.valid_tool_names else None, honcho_manager=self._honcho, honcho_session_key=self._honcho_session_key, + retaindb_manager=self._retaindb, + retaindb_session_key=self._retaindb_session_key, ) except Exception as tool_error: function_result = f"Error executing tool '{function_name}': {tool_error}" @@ -5455,6 +5739,7 @@ def run_conversation( # to consume background prefetch results from turn N-1. self._honcho_context = "" self._honcho_turn_context = "" + self._retaindb_turn_context = "" _recall_mode = (self._honcho_config.recall_mode if self._honcho_config else "hybrid") if self._honcho and self._honcho_session_key and _recall_mode != "tools": try: @@ -5467,6 +5752,13 @@ def run_conversation( except Exception as e: logger.debug("Honcho prefetch failed (non-fatal): %s", e) + _retaindb_recall_mode = (self._retaindb_config.recall_mode if self._retaindb_config else "hybrid") + if self._retaindb and self._retaindb_session_key and _retaindb_recall_mode != "tools": + try: + self._queue_retaindb_prefetch(original_user_message, messages) + except Exception as e: + logger.debug("RetainDB prefetch scheduling failed (non-fatal): %s", e) + # Add user message user_msg = {"role": "user", "content": user_message} messages.append(user_msg) @@ -5565,6 +5857,12 @@ def run_conversation( if _preflight_tokens < self.context_compressor.threshold_tokens: break # Under threshold + if self._retaindb and self._retaindb_session_key and _retaindb_recall_mode != "tools": + try: + self._retaindb_turn_context = self._retaindb_prefetch() + except Exception as e: + logger.debug("RetainDB prefetch failed (non-fatal): %s", e) + # Main conversation loop api_call_count = 0 final_response = None @@ -5625,10 +5923,17 @@ def run_conversation( for idx, msg in enumerate(messages): api_msg = msg.copy() - if idx == current_turn_user_idx and msg.get("role") == "user" and self._honcho_turn_context: - api_msg["content"] = _inject_honcho_turn_context( - api_msg.get("content", ""), self._honcho_turn_context - ) + if idx == current_turn_user_idx and msg.get("role") == "user": + current_content = api_msg.get("content", "") + if self._honcho_turn_context: + current_content = _inject_honcho_turn_context( + current_content, self._honcho_turn_context + ) + if self._retaindb_turn_context: + current_content = _inject_retaindb_turn_context( + current_content, self._retaindb_turn_context + ) + api_msg["content"] = current_content # For ALL assistant messages, pass reasoning back to the API # This ensures multi-turn reasoning context is preserved @@ -7073,6 +7378,7 @@ def _stop_spinner(): if final_response and not interrupted and sync_honcho: self._honcho_sync(original_user_message, final_response) self._queue_honcho_prefetch(original_user_message) + self._retaindb_sync(original_user_message, final_response) # Extract reasoning from the last assistant message (if any) last_reasoning = None diff --git a/tests/gateway/test_honcho_lifecycle.py b/tests/gateway/test_honcho_lifecycle.py index 01cff91826aa..9db1014a25d0 100644 --- a/tests/gateway/test_honcho_lifecycle.py +++ b/tests/gateway/test_honcho_lifecycle.py @@ -101,7 +101,8 @@ async def test_reset_shuts_down_gateway_honcho_manager(self): 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") + runner._async_flush_memories.assert_called_once() + assert runner._async_flush_memories.call_args.args == ("old-session", "gateway-key", None) assert "Session reset" in result def test_flush_memories_reuses_gateway_session_key_and_skips_honcho_sync(self): diff --git a/tests/gateway/test_retaindb_lifecycle.py b/tests/gateway/test_retaindb_lifecycle.py new file mode 100644 index 000000000000..b0145792c61a --- /dev/null +++ b/tests/gateway/test_retaindb_lifecycle.py @@ -0,0 +1,132 @@ +"""Tests for gateway-owned RetainDB 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._retaindb_managers = {} + runner._retaindb_configs = {} + runner._running_agents = {} + runner._pending_messages = {} + runner._pending_approvals = {} + runner.adapters = {} + runner.hooks = MagicMock() + runner.hooks.emit = AsyncMock() + return runner + + +def _make_source(): + return SessionSource( + platform=Platform.TELEGRAM, + chat_id="chat-1", + user_id="user-1", + user_name="alice", + ) + + +def _make_event(text="/reset"): + return MessageEvent(text=text, source=_make_source()) + + +class TestGatewayRetainDBLifecycle: + def test_gateway_reuses_retaindb_manager_for_session_key(self): + runner = _make_runner() + identity_one = {"platform_user_id": "user-1", "session_id": "session-1"} + identity_two = {"platform_user_id": "user-1", "session_id": "session-2"} + rcfg = SimpleNamespace(should_activate=lambda: True) + manager = MagicMock() + + with patch( + "retaindb_integration.client.RetainDBClientConfig.from_global_config", + return_value=rcfg, + ), patch( + "retaindb_integration.session.RetainDBSessionManager", + return_value=manager, + ) as mock_mgr_cls: + first_mgr, first_cfg = runner._get_or_create_gateway_retaindb("session-key", identity_one) + second_mgr, second_cfg = runner._get_or_create_gateway_retaindb("session-key", identity_two) + + assert first_mgr is manager + assert second_mgr is manager + assert first_cfg is rcfg + assert second_cfg is rcfg + mock_mgr_cls.assert_called_once_with(config=rcfg, runtime_identity=identity_one) + manager.set_runtime_identity.assert_called_once_with(identity_two) + + def test_gateway_skips_retaindb_manager_when_inactive(self): + runner = _make_runner() + rcfg = SimpleNamespace(should_activate=lambda: False) + + with patch( + "retaindb_integration.client.RetainDBClientConfig.from_global_config", + return_value=rcfg, + ), patch("retaindb_integration.session.RetainDBSessionManager") as mock_mgr_cls: + manager, cfg = runner._get_or_create_gateway_retaindb("session-key") + + assert manager is None + assert cfg is rcfg + mock_mgr_cls.assert_not_called() + + @pytest.mark.asyncio + async def test_reset_shuts_down_gateway_retaindb_manager(self): + runner = _make_runner() + event = _make_event() + origin = _make_source() + runner._shutdown_gateway_honcho = MagicMock() + runner._shutdown_gateway_retaindb = MagicMock() + runner._async_flush_memories = AsyncMock() + runner._evict_cached_agent = MagicMock() + 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", origin=origin), + } + runner.session_store.reset_session.return_value = SimpleNamespace(session_id="new-session") + + result = await runner._handle_reset_command(event) + + runner._shutdown_gateway_retaindb.assert_called_once_with("gateway-key") + runner._async_flush_memories.assert_called_once() + assert runner._async_flush_memories.call_args.args == ("old-session", "gateway-key", origin) + assert "Session reset" in result + + def test_flush_memories_passes_retaindb_identity_and_session_key(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"}, + ] + runner._get_or_create_gateway_retaindb = MagicMock( + return_value=(MagicMock(), SimpleNamespace()) + ) + tmp_agent = MagicMock() + source = _make_source() + + 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", source) + + _, kwargs = mock_agent_cls.call_args + assert kwargs["session_id"] == "old-session" + assert kwargs["retaindb_session_key"] == "old-session" + assert kwargs["retaindb_identity"]["platform_user_id"] == "user-1" + assert kwargs["retaindb_identity"]["session_id"] == "old-session" + runner._get_or_create_gateway_retaindb.assert_called_once() diff --git a/tests/hermes_cli/test_doctor.py b/tests/hermes_cli/test_doctor.py index f91d17811759..c292a58953f7 100644 --- a/tests/hermes_cli/test_doctor.py +++ b/tests/hermes_cli/test_doctor.py @@ -75,6 +75,28 @@ def test_reports_not_configured_without_api_key(self, monkeypatch): assert not doctor._honcho_is_configured_for_doctor() +class TestRetainDBDoctorConfigDetection: + def test_reports_configured_when_enabled_with_api_key_and_project(self, monkeypatch): + fake_config = SimpleNamespace(enabled=True, api_key="***", project="demo") + + monkeypatch.setattr( + "retaindb_integration.client.RetainDBClientConfig.from_global_config", + lambda: fake_config, + ) + + assert doctor._retaindb_is_configured_for_doctor() + + def test_reports_not_configured_without_project(self, monkeypatch): + fake_config = SimpleNamespace(enabled=True, api_key="***", project="") + + monkeypatch.setattr( + "retaindb_integration.client.RetainDBClientConfig.from_global_config", + lambda: fake_config, + ) + + assert not doctor._retaindb_is_configured_for_doctor() + + def test_run_doctor_sets_interactive_env_for_tool_checks(monkeypatch, tmp_path): """Doctor should present CLI-gated tools as available in CLI context.""" project_root = tmp_path / "project" diff --git a/tests/tools/test_retaindb_tools.py b/tests/tools/test_retaindb_tools.py new file mode 100644 index 000000000000..d014b34def9e --- /dev/null +++ b/tests/tools/test_retaindb_tools.py @@ -0,0 +1,76 @@ +"""Regression tests for per-call RetainDB tool session routing.""" + +import json +from unittest.mock import MagicMock + +from retaindb_integration.identity import ResolvedRetainDBIdentity +from tools import retaindb_tools + + +class TestRetainDBToolSessionContext: + def setup_method(self): + self.orig_manager = retaindb_tools._session_manager + self.orig_key = retaindb_tools._session_key + + def teardown_method(self): + retaindb_tools._session_manager = self.orig_manager + retaindb_tools._session_key = self.orig_key + + def test_explicit_call_context_wins_over_module_global_state(self): + global_manager = MagicMock() + global_manager.get_profile.return_value = {"name": "global"} + explicit_manager = MagicMock() + explicit_manager.get_profile.return_value = {"name": "explicit"} + + retaindb_tools.set_session_context(global_manager, "global-session") + + result = json.loads( + retaindb_tools._handle_retaindb_profile( + {}, + retaindb_manager=explicit_manager, + retaindb_session_key="explicit-session", + ) + ) + + assert result == {"result": {"name": "explicit"}} + explicit_manager.get_profile.assert_called_once_with("explicit-session") + global_manager.get_profile.assert_not_called() + + def test_context_tool_serializes_identity_dataclass_payload(self): + manager = MagicMock() + manager.get_context.return_value = { + "identity": ResolvedRetainDBIdentity( + user_id="user-1", + session_id="session-1", + agent_id="hermes", + project="default", + source="platform", + peer_name="alice", + platform="telegram", + chat_id="chat-1", + ), + "context": "[RetainDB Context]\nRelevant memories:\n- cedar-42", + "profile": {"memories": []}, + "query": {"results": []}, + } + + result = json.loads( + retaindb_tools._handle_retaindb_context( + {"query": "What matters now?"}, + retaindb_manager=manager, + retaindb_session_key="session-1", + ) + ) + + assert result["result"]["identity"] == { + "user_id": "user-1", + "session_id": "session-1", + "agent_id": "hermes", + "project": "default", + "source": "platform", + "peer_name": "alice", + "platform": "telegram", + "chat_id": "chat-1", + } + assert "RetainDB Context" in result["result"]["context"] + manager.get_context.assert_called_once_with("session-1", "What matters now?") diff --git a/tools/retaindb_tools.py b/tools/retaindb_tools.py new file mode 100644 index 000000000000..b29ffd2dde17 --- /dev/null +++ b/tools/retaindb_tools.py @@ -0,0 +1,258 @@ +"""RetainDB tools for Hermes' native deep-memory integration.""" + +from __future__ import annotations + +import json +import logging +from dataclasses import asdict, is_dataclass +from datetime import date, datetime + +logger = logging.getLogger(__name__) + +_session_manager = None +_session_key: str | None = None + + +def set_session_context(session_manager, session_key: str) -> None: + global _session_manager, _session_key + _session_manager = session_manager + _session_key = session_key + + +def clear_session_context() -> None: + global _session_manager, _session_key + _session_manager = None + _session_key = None + + +def _check_retaindb_available() -> bool: + return _session_manager is not None and _session_key is not None + + +def _resolve_session_context(**kwargs): + session_manager = kwargs.get("retaindb_manager") or _session_manager + session_key = kwargs.get("retaindb_session_key") or _session_key + return session_manager, session_key + + +def _json_default(value): + """Serialize a few common Python/runtime types used by RetainDB helpers.""" + if is_dataclass(value): + return asdict(value) + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, set): + return sorted(value) + raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") + + +def _json_response(payload: dict) -> str: + return json.dumps(payload, ensure_ascii=False, default=_json_default) + + +_PROFILE_SCHEMA = { + "name": "retaindb_profile", + "description": ( + "Retrieve a compact stable profile for this user from RetainDB. " + "Use this for preferences, working style, and durable user facts." + ), + "parameters": {"type": "object", "properties": {}, "required": []}, +} + + +def _handle_retaindb_profile(args: dict, **kwargs) -> str: + session_manager, session_key = _resolve_session_context(**kwargs) + if not session_manager or not session_key: + return _json_response({"error": "RetainDB is not active for this session."}) + try: + return _json_response({"result": session_manager.get_profile(session_key)}) + except Exception as exc: + logger.error("RetainDB profile lookup failed: %s", exc) + return _json_response({"error": f"Failed to fetch profile: {exc}"}) + + +_SEARCH_SCHEMA = { + "name": "retaindb_search", + "description": ( + "Search RetainDB memory for facts related to a query. " + "Use this when you need deeper cross-session recall beyond the injected turn context." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What to search for."}, + "top_k": { + "type": "integer", + "description": "How many matching memories to return (default 8, max 20).", + }, + }, + "required": ["query"], + }, +} + + +def _handle_retaindb_search(args: dict, **kwargs) -> str: + query = str(args.get("query") or "").strip() + if not query: + return _json_response({"error": "Missing required parameter: query"}) + session_manager, session_key = _resolve_session_context(**kwargs) + if not session_manager or not session_key: + return _json_response({"error": "RetainDB is not active for this session."}) + top_k = max(1, min(int(args.get("top_k", 8)), 20)) + try: + return _json_response({"result": session_manager.search(session_key, query, top_k=top_k)}) + except Exception as exc: + logger.error("RetainDB search failed: %s", exc) + return _json_response({"error": f"Failed to search RetainDB: {exc}"}) + + +_CONTEXT_SCHEMA = { + "name": "retaindb_context", + "description": ( + "Ask RetainDB for the best compact 'what matters now' memory context for this query. " + "Use this when you need a synthesized memory block rather than raw search results." + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "What context to retrieve."}, + }, + "required": ["query"], + }, +} + + +def _handle_retaindb_context(args: dict, **kwargs) -> str: + query = str(args.get("query") or "").strip() + if not query: + return _json_response({"error": "Missing required parameter: query"}) + session_manager, session_key = _resolve_session_context(**kwargs) + if not session_manager or not session_key: + return _json_response({"error": "RetainDB is not active for this session."}) + try: + return _json_response({"result": session_manager.get_context(session_key, query)}) + except Exception as exc: + logger.error("RetainDB context lookup failed: %s", exc) + return _json_response({"error": f"Failed to fetch context: {exc}"}) + + +_REMEMBER_SCHEMA = { + "name": "retaindb_remember", + "description": ( + "Persist an explicit fact, preference, correction, or instruction into RetainDB. " + "Use this only for durable information that should survive future sessions." + ), + "parameters": { + "type": "object", + "properties": { + "content": {"type": "string", "description": "The fact or preference to remember."}, + "memory_type": { + "type": "string", + "description": "Optional type: factual, preference, event, relationship, opinion, goal, instruction.", + }, + "importance": { + "type": "number", + "description": "Optional importance from 0.0 to 1.0.", + }, + }, + "required": ["content"], + }, +} + + +def _handle_retaindb_remember(args: dict, **kwargs) -> str: + content = str(args.get("content") or "").strip() + if not content: + return _json_response({"error": "Missing required parameter: content"}) + session_manager, session_key = _resolve_session_context(**kwargs) + if not session_manager or not session_key: + return _json_response({"error": "RetainDB is not active for this session."}) + memory_type = str(args.get("memory_type") or "factual").strip() or "factual" + importance = float(args.get("importance", 0.7)) + try: + return _json_response( + { + "result": session_manager.remember( + session_key, + content, + memory_type=memory_type, + importance=importance, + ) + } + ) + except Exception as exc: + logger.error("RetainDB remember failed: %s", exc) + return _json_response({"error": f"Failed to remember: {exc}"}) + + +_FORGET_SCHEMA = { + "name": "retaindb_forget", + "description": ( + "Delete a specific RetainDB memory by id. " + "Use this when the user explicitly asks to forget or remove stored information." + ), + "parameters": { + "type": "object", + "properties": { + "memory_id": {"type": "string", "description": "The memory id to delete."}, + }, + "required": ["memory_id"], + }, +} + + +def _handle_retaindb_forget(args: dict, **kwargs) -> str: + memory_id = str(args.get("memory_id") or "").strip() + if not memory_id: + return _json_response({"error": "Missing required parameter: memory_id"}) + session_manager, _session = _resolve_session_context(**kwargs) + if not session_manager: + return _json_response({"error": "RetainDB is not active for this session."}) + try: + return _json_response({"result": session_manager.forget(memory_id)}) + except Exception as exc: + logger.error("RetainDB forget failed: %s", exc) + return _json_response({"error": f"Failed to forget memory: {exc}"}) + + +from tools.registry import registry + +registry.register( + name="retaindb_profile", + toolset="retaindb", + schema=_PROFILE_SCHEMA, + handler=_handle_retaindb_profile, + check_fn=_check_retaindb_available, +) + +registry.register( + name="retaindb_search", + toolset="retaindb", + schema=_SEARCH_SCHEMA, + handler=_handle_retaindb_search, + check_fn=_check_retaindb_available, +) + +registry.register( + name="retaindb_context", + toolset="retaindb", + schema=_CONTEXT_SCHEMA, + handler=_handle_retaindb_context, + check_fn=_check_retaindb_available, +) + +registry.register( + name="retaindb_remember", + toolset="retaindb", + schema=_REMEMBER_SCHEMA, + handler=_handle_retaindb_remember, + check_fn=_check_retaindb_available, +) + +registry.register( + name="retaindb_forget", + toolset="retaindb", + schema=_FORGET_SCHEMA, + handler=_handle_retaindb_forget, + check_fn=_check_retaindb_available, +) diff --git a/toolsets.py b/toolsets.py index a314f277b738..f337d90447fe 100644 --- a/toolsets.py +++ b/toolsets.py @@ -62,6 +62,8 @@ "send_message", # Honcho memory tools (gated on honcho being active via check_fn) "honcho_context", "honcho_profile", "honcho_search", "honcho_conclude", + # RetainDB memory tools (gated on RetainDB being active via check_fn) + "retaindb_profile", "retaindb_search", "retaindb_context", "retaindb_remember", "retaindb_forget", # Home Assistant smart home control (gated on HASS_TOKEN via check_fn) "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service", ] @@ -202,6 +204,12 @@ "includes": [] }, + "retaindb": { + "description": "RetainDB deep semantic memory for persistent cross-session recall", + "tools": ["retaindb_profile", "retaindb_search", "retaindb_context", "retaindb_remember", "retaindb_forget"], + "includes": [] + }, + "homeassistant": { "description": "Home Assistant smart home control and monitoring", "tools": ["ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service"], diff --git a/website/docs/reference/cli-commands.md b/website/docs/reference/cli-commands.md index b76859081645..d968be4dff6a 100644 --- a/website/docs/reference/cli-commands.md +++ b/website/docs/reference/cli-commands.md @@ -44,6 +44,7 @@ hermes [global-options] [subcommand/options] | `hermes pairing` | Approve or revoke messaging pairing codes. | | `hermes skills` | Browse, install, publish, audit, and configure skills. | | `hermes honcho` | Manage Honcho cross-session memory integration. | +| `hermes retaindb` | Manage RetainDB cross-session memory integration. | | `hermes acp` | Run Hermes as an ACP server for editor integration. | | `hermes tools` | Configure enabled tools per platform. | | `hermes sessions` | Browse, export, prune, rename, and delete sessions. | @@ -303,6 +304,23 @@ Subcommands: | `identity` | Seed or show the AI peer identity representation. | | `migrate` | Migration guide from openclaw-honcho to Hermes Honcho. | +## `hermes retaindb` + +```bash +hermes retaindb +``` + +Subcommands: + +| Subcommand | Description | +|------------|-------------| +| `setup` | Interactive RetainDB setup wizard. | +| `status` | Show current RetainDB config and connection status. | +| `test` | Run RetainDB read/write smoke tests. | +| `mode` | Show or set memory mode: `hybrid` or `retaindb`. | +| `tokens` | Show or set the RetainDB context token budget. | +| `identity` | Show the resolved RetainDB identity mapping. | + ## `hermes acp` ```bash diff --git a/website/docs/user-guide/features/memory.md b/website/docs/user-guide/features/memory.md index c0810b69345c..3ee9f7d05406 100644 --- a/website/docs/user-guide/features/memory.md +++ b/website/docs/user-guide/features/memory.md @@ -209,10 +209,20 @@ memory: ## Honcho Integration (Cross-Session User Modeling) -For deeper, AI-generated user understanding that works across sessions and platforms, you can enable [Honcho Memory](./honcho.md). Honcho runs alongside built-in memory in `hybrid` mode (the default) — `MEMORY.md` and `USER.md` stay as-is, and Honcho adds a persistent user modeling layer on top. +For deeper, AI-generated user understanding that works across sessions and platforms, you can enable [Honcho Memory](./honcho.md). Honcho runs alongside built-in memory in `hybrid` mode (the default) - `MEMORY.md` and `USER.md` stay as-is, and Honcho adds a persistent user modeling layer on top. ```bash hermes honcho setup ``` See the [Honcho Memory](./honcho.md) docs for full configuration, tools, and CLI reference. + +## RetainDB Integration (Cross-Session Session Memory) + +RetainDB adds a native cross-session memory layer for Hermes. It can ingest session turns in the background, retrieve a compact memory overlay for the current turn, and expose explicit RetainDB memory tools when enabled. + +```bash +hermes retaindb setup +``` + +See the [RetainDB Memory](./retaindb.md) docs for setup, modes, and CLI commands. diff --git a/website/docs/user-guide/features/retaindb.md b/website/docs/user-guide/features/retaindb.md new file mode 100644 index 000000000000..8bd80555eec4 --- /dev/null +++ b/website/docs/user-guide/features/retaindb.md @@ -0,0 +1,112 @@ +--- +title: RetainDB Memory +description: Native RetainDB integration for cross-session memory in Hermes. +sidebar_label: RetainDB Memory +sidebar_position: 9 +--- + +# RetainDB Memory + +RetainDB adds an optional cross-session memory layer to Hermes. Hermes still keeps its local memory files (`MEMORY.md`, `USER.md`) and `state.db`; RetainDB adds deeper recall, background session ingestion, and explicit memory tools on top. + +## What It Does + +- Prefetches a compact memory overlay for the current turn +- Queues user and assistant turns for durable background ingestion +- Supports explicit memory writes and deletes through native Hermes tools +- Falls back cleanly to Hermes local memory when RetainDB is unavailable + +## Setup + +### Interactive setup + +```bash +hermes retaindb setup +``` + +### Non-interactive setup + +```bash +hermes retaindb setup --yes --api-key --project +``` + +The setup command writes: + +- `~/.hermes/config.yaml` for RetainDB runtime settings +- `~/.hermes/.env` for `RETAINDB_API_KEY` and optional `RETAINDB_BASE_URL` + +## Configuration + +Hermes stores RetainDB settings under `retaindb:` in `~/.hermes/config.yaml`: + +```yaml +retaindb: + enabled: true + base_url: https://api.retaindb.com + project: my-project + memory_mode: hybrid + recall_mode: hybrid + write_frequency: async + context_tokens: 1200 + prefetch_timeout_ms: 1500 + flush_batch_size: 50 + disable_tool_exposure: false + debug_recall_trace: false + agent_id: hermes +``` + +### Memory modes + +| Mode | Behavior | +|------|----------| +| `hybrid` | Keep Hermes local memory and RetainDB enabled together | +| `retaindb` | Disable local `MEMORY.md` / `USER.md` writes and rely on RetainDB | + +### Recall modes + +| Mode | Behavior | +|------|----------| +| `hybrid` | Inject RetainDB context and keep the RetainDB tools available | +| `context` | Inject RetainDB context, hide the RetainDB tools | +| `tools` | Expose RetainDB tools only, skip automatic context injection | + +### Write frequency + +| Setting | Behavior | +|---------|----------| +| `async` | Queue session ingestion in the background | +| `turn` | Flush after every turn | +| integer `N` | Flush every `N` turns | + +## Tools + +When RetainDB is active, Hermes can expose these tools: + +- `retaindb_profile` +- `retaindb_search` +- `retaindb_context` +- `retaindb_remember` +- `retaindb_forget` + +Use `retaindb_profile` and `retaindb_search` for raw recall. Use `retaindb_context` when you want a compact synthesized memory block for the current question. + +## CLI Commands + +```bash +hermes retaindb setup +hermes retaindb status +hermes retaindb test +hermes retaindb mode [hybrid|retaindb] +hermes retaindb tokens --context N +hermes retaindb identity [--session-id SESSION] +``` + +## How It Fits With Hermes Memory + +RetainDB does not replace Hermes session storage. Hermes still uses its own session DB and local memory files unless you switch to `memory_mode: retaindb`. + +The common default is: + +- Hermes local memory for hot local context +- RetainDB for deeper cross-session recall +- async session ingestion so useful context can survive into later sessions