hermes-agent/agent/memory_provider.py

"""Abstract base class for pluggable memory providers.

Plugins ship in ``plugins/memory/<name>/``, 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.

    Multiplexing contract (#140058). One gateway process can serve several profiles, and each agent's
    ``MemoryManager`` owns its own provider instance, initialized for one profile. A provider must:

    - hand every load a NEW instance (``register(ctx)`` / the loader constructs one per agent). The
      manager refuses an instance another profile's manager already bound, rather than rebinding it;
    - bind every store, path, client, bank or tenant from ``initialize()``'s ``hermes_home`` (and
      scoped secrets read during ``initialize``) and keep that binding on the instance;
    - keep no profile-specific mutable state at module or class level, and never "rebind" a shared
      binding per call: two profiles' calls interleave on different threads.

    The manager runs tool calls and lifecycle hooks only from the profile the instance was initialized
    for (refusing callers that name another profile, or none on a multiplexed process), runs its
    background hooks under that profile's ``HERMES_HOME``, and passes ``hermes_home`` /
    ``agent_identity`` / ``caller_gsk`` to ``handle_tool_call`` when it accepts ``**kwargs``. Those
    keywords name the same profile the instance is bound to; they are not a reason to rebind. Core
    cannot see inside a provider: one that keeps a process-wide binding still breaks this contract.
    """

    # 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. A memory-tool patch arrives as a replace whose
        ``content`` is the whole entry after the edit. 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 []

    # -- Profile lifecycle (uninitialized instance; that profile's home and secrets bound; errors are
    # logged). Keep them fast: the dashboard and Desktop run them inside the create/rename request,
    # so put network work (creating a remote peer) on a background thread. --

    def on_profile_created(self, profile_name: str, hermes_home: str) -> None:
        """A profile whose config selects this provider was just created by ``create_profile`` (CLI,
        dashboard, Desktop/TUI; a clone carries ``memory.provider``). Not called for profile import.
        Set up per-profile provider state here (a peer, a host entry)."""

    def on_profile_renamed(self, old_name: str, new_name: str, hermes_home: str) -> None:
        """A profile whose config selects this provider was renamed; ``hermes_home`` is its new
        directory. Re-key anything the provider stores under the old profile name."""

    # -- Dashboard config panel for a ``config_schema.py`` declaring ``storage=STORAGE_PROVIDER`` --

    def read_config_fields(self, hermes_home: str) -> dict[str, Any]:
        """``{"values": {field key: stored native value}, "placeholders": {field key: text}}``. A key
        absent from ``values`` reads as unset; a placeholder shows what a blank field resolves to.
        A stored secret may be included: it only marks the field as set and is never sent out."""
        return {"values": {}, "placeholders": {}}

    def write_config_fields(self, values: dict[str, Any], hermes_home: str) -> None:
        """Persist the submitted fields (only those present): natives already coerced, ``None`` clears a
        key, non-blank secrets included (core saves each to ``.env`` under its ``env_key`` once this
        returns). Raise ``ValueError`` to refuse the save (the dashboard shows it as a 400)."""
        raise NotImplementedError(f"memory provider {self.name} declares storage=provider but cannot write it")