"""Abstract base class for pluggable memory providers. Plugins ship in ``plugins/memory//``, activated via ``memory.provider`` (ONE external provider at a time). Lifecycle, driven by MemoryManager: initialize -> system_prompt_block / prefetch / sync_turn per turn -> tool dispatch -> shutdown, plus optional ``on_*`` hooks. """ from __future__ import annotations import contextvars import logging import re import threading from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional logger = logging.getLogger(__name__) def ctx_bound(fn: Callable[..., Any]) -> Callable[..., Any]: """Bind ``fn`` to the CALLER's contextvars for another thread/executor. Profile isolation is a ContextVar-scoped HERMES_HOME override plus the per-turn secret scope; a worker started with an empty context silently lands on the default profile (or fails closed on secrets).""" ctx = contextvars.copy_context() return lambda *args, **kwargs: ctx.run(fn, *args, **kwargs) def spawn_context_thread(target: Callable[..., Any], *, name: str, daemon: bool = True, args: tuple = (), kwargs: Optional[Dict[str, Any]] = None) -> threading.Thread: """Unstarted thread running *target* under the spawner's contextvars (see :func:`ctx_bound`). Every memory-provider background job (prefetch, sync, writer loops) must go through this.""" return threading.Thread(target=ctx_bound(target), args=args, kwargs=kwargs, name=name, daemon=daemon) # v1 = best-effort on_pre_compress() with the raw message list; v2 = opt-in fail-closed # checkpoint (normalized evidence handoff + strict-mode failure propagation). PRE_COMPRESS_CHECKPOINT_API_VERSION = 2 # Default glyph for recall indicators; providers may use their own brand mark. INDICATOR_GLYPH = "🧠" # ``memory.provider`` values that mean "the built-in store, no external plugin". The built-in # store is core: doctor, migration and dependency refresh must never look these up as plugins. CORE_MEMORY_PROVIDER_SENTINELS = frozenset({"", "default", "builtin", "built-in", "none"}) def is_core_memory_provider(name: Optional[str]) -> bool: """True when ``memory.provider`` selects the built-in store rather than an external plugin.""" return str(name or "").strip().lower() in CORE_MEMORY_PROVIDER_SENTINELS @dataclass(frozen=True) class RecallStatus: """What the last prefetch injected, for the deterministic recall indicator (``MemoryManager.describe_recall``). ``count == 0`` means content without a discrete count (e.g. a synthesized reflect answer) and renders generically.""" provider_label: str count: int glyph: str = INDICATOR_GLYPH # Prompts with no semantic signal; single source of truth for the core prefetch gate and # provider-side classifiers. Anchored and followed only by whitespace/punctuation, so # "k8s"/"yolo"/"note" do NOT match while "hi!"/"thanks :)"/"done???" do. TRIVIAL_PROMPT_RE = re.compile( r'^(yes|no|ok|okay|sure|thanks|thank you|y|n|yep|nope|yeah|nah|' r'hi|hey|hello|yo|sup|' r'continue|go ahead|do it|proceed|got it|cool|nice|great|done|next|lgtm|k)' r'[\s!?.:;,"' + "'" + r'~\u2018\u2019\u201c\u201d\u2014\u2013\u2026()\[\]{}<>*&^%$#@!+=`\u00a0]*$', re.IGNORECASE, ) def is_trivial_prompt(text: Optional[str]) -> bool: """True for empty input, slash commands and bare greetings/acknowledgements (skipping recall saves a round-trip and keeps stale context from derailing one-word replies).""" stripped = (text or "").strip() if not stripped or stripped.startswith("/"): return True return bool(TRIVIAL_PROMPT_RE.match(stripped)) class MemoryProvider(ABC): """Abstract base class for memory providers.""" # Providers that durably checkpoint every successful on_pre_compress() set this to # PRE_COMPRESS_CHECKPOINT_API_VERSION; 1 = best-effort legacy. pre_compress_checkpoint_api_version = 1 @property @abstractmethod def name(self) -> str: """Short identifier for this provider (e.g. 'builtin', 'honcho', 'hindsight').""" # -- Core lifecycle (implement these) ------------------------------------ @abstractmethod def is_available(self) -> bool: """Configured, credentialed and ready? Gates activation; check config/deps only, no network.""" @abstractmethod def initialize(self, session_id: str, **kwargs) -> None: """Initialize once at agent startup (connections, resources, threads). kwargs always include ``hermes_home`` (profile-scoped storage; never hardcode ``~/.hermes``) and ``platform``; may include ``agent_context`` ("primary" | "subagent" | "cron" | "flush" — skip writes for non-primary contexts), ``agent_identity``, ``agent_workspace``, ``parent_session_id``, ``user_id``, ``user_id_alt``. """ def unavailable_reason(self) -> str: """User-facing hint for the "provider unavailable" warning (``initialize()`` never runs then).""" return "" def system_prompt_block(self) -> str: """STATIC system-prompt text; "" to skip. Recalled context goes through prefetch(), not here.""" return "" def prefetch(self, query: str, *, session_id: str = "") -> str: """Formatted recall context for the upcoming turn ("" if none). Must be fast — recall in the background and return cached results; ``session_id`` scopes concurrent sessions.""" return "" def queue_prefetch(self, query: str, *, session_id: str = "") -> None: """Queue a background recall after each turn; prefetch() consumes it next turn.""" def recall_status(self) -> Optional[RecallStatus]: """What the most recent :meth:`prefetch` injected (``None`` = no indicator). Must reflect only the LAST prefetch, never a stale prior count.""" return None def sync_turn( self, user_content: str, assistant_content: str, *, session_id: str = "", messages: Optional[List[Dict[str, Any]]] = None, turn_author: Optional[Dict[str, Any]] = None, ) -> None: """Persist a completed turn (non-blocking). ``messages`` is the OpenAI-style list so far. ``turn_author`` (``{"id", "name", "is_bot"}``) is who wrote the user side; the manager sends it only to signatures that accept it.""" @abstractmethod def get_tool_schemas(self) -> List[Dict[str, Any]]: """OpenAI function-calling schemas ({"name", "description", "parameters"}); [] if none.""" def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str: """Handle one of this provider's tools; must return a JSON string.""" raise NotImplementedError(f"Provider {self.name} does not handle tool {tool_name}") def shutdown(self) -> None: """Clean shutdown — flush queues, close connections.""" # -- Optional hooks (override to opt in) --------------------------------- def on_turn_start(self, turn_number: int, message: str, **kwargs) -> None: """Per-turn tick. kwargs may include remaining_tokens, model, platform, tool_count, author_id, author_name, author_is_bot. The author trio names who wrote THIS turn (None, None, False without one): a shared session carries several participants, so a provider keying durable state on identity must read it per turn.""" def identity_signature(self) -> Dict[str, Any]: """Identity-mapping values that must bust a cached gateway agent when they change (writer identity, alias tables, session-name prefixing). Provider-namespaced keys, JSON-serializable values. The gateway calls this on an uninitialized instance on every inbound message, so keep it cheap and read-only.""" return {} def on_session_end(self, messages: List[Dict[str, Any]]) -> None: """End-of-session extraction; fires only at real session boundaries, never per-turn.""" def on_session_switch( self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False, rewound: bool = False, **kwargs, ) -> None: """session_id reassigned mid-process (/resume, /branch, /reset, /new, compression) without teardown: rebind per-session state so later writes land in the right record. ``reset`` is True only for a genuinely new conversation (flush buffers); ``rewound``: same id but the transcript was truncated.""" def on_pre_compress(self, messages: List[Dict[str, Any]]) -> str: """Extract insights from ``messages`` about to be compressed, fed into the summary prompt.""" return "" def on_delegation(self, task: str, result: str, *, child_session_id: str = "", **kwargs) -> None: """PARENT-side observation of a completed delegation (the subagent has no provider session).""" def get_config_schema(self) -> List[Dict[str, Any]]: """Setup fields for ``hermes memory setup`` ([] if none): ``key``, ``description``, optional ``secret`` (goes to .env), ``required``, ``default``, ``choices``, ``type`` (text | integer | number | boolean), ``minimum``/``maximum``/``step``, ``url``, ``env_var`` (explicit secret env var; default auto-generated).""" return [] def save_config(self, values: Dict[str, Any], hermes_home: str) -> None: """Write non-secret setup ``values`` to the provider's native config. Plugins MUST either override this or use only env vars (every schema field carrying ``env_var``).""" def on_memory_write(self, action: str, target: str, content: str, metadata: Optional[Dict[str, Any]] = None) -> None: """Mirror a built-in memory-tool write (``action``: add | replace | remove; ``target``: memory | user; ``metadata``: provenance such as write_origin, session_id, tool_name). For replace/remove, ``metadata["previous_content"]`` is the full entry selected under the native-store lock. Notifications follow a successful complete write or batch; each batch operation sees the preceding operation's result. Older callers may omit this field: ``old_text`` alone is not authoritative identity. """ def backup_paths(self) -> List[str]: """Absolute paths of provider state OUTSIDE HERMES_HOME for ``hermes backup``/``import`` (paths outside the home dir are skipped). MUST work without ``initialize()`` or network.""" return []