agent/agent_init_memory_provider.py
Rama main · réplica de lectura del repositorio original.
"""Memory provider at agent init: resolve the configured one (telling the user once why it will not run)
and build the kwargs ``MemoryProvider.initialize`` receives."""
from __future__ import annotations
import logging
from contextlib import suppress
from typing import Any
from agent.agent_init_params import _GATEWAY_IDENTITY_PARAMS
from hermes_constants import get_hermes_home
logger = logging.getLogger("run_agent")
# Deduped: the gateway builds a fresh AIAgent per message, so it would warn every turn.
_warned_unavailable_providers: set[tuple[str, str, str]] = set()
def _unavailable_warning_key(name: str, kind: str = "unavailable") -> tuple[str, str, str]:
"""Once per profile home, provider and kind of trouble: one multiplexed gateway/Desktop backend
serves several profiles, each one running without its memory must be told, and a provider the
user turns back on that is then unavailable is news even after the "turned off" notice."""
from hermes_constants import get_hermes_home, hermes_home_key
return hermes_home_key(get_hermes_home()), name, kind
def _warn_once(name: str, message: str, say=None, kind: str = "unavailable") -> None:
"""Log *message* and hand it to the user's sink once per home, provider and kind."""
key = _unavailable_warning_key(name, kind)
if key in _warned_unavailable_providers:
return
_warned_unavailable_providers.add(key)
logger.warning(message)
if say is not None:
say(message)
def _warn_memory_provider_unavailable(name: str, reason: str = "", say=None) -> None:
"""Warn once per home and provider that a configured memory provider is unavailable.
``is_available()`` is a side-effect-free hot-path check and can't log itself; without this
the provider is silently dropped. ``reason`` (the provider's ``unavailable_reason()`` hint)
can only reach the user here, so it is appended when present. *say* is the agent's
user-facing sink: a log line alone leaves the user running without memory unaware.
"""
_warn_once(name, (
f"⚠ Memory provider {name!r} is selected but reports unavailable — external memory "
"is disabled for this session (built-in memory still works). Check the "
"provider's credentials/config with 'hermes memory status'. Note: "
"systemd/gateway services do not inherit ~/.hermes/.env automatically; set "
f"any required variables in the service environment.{f' {reason}' if reason else ''}"
), say=say)
def resolve_memory_provider(name: str, say=None):
"""The configured provider ready to add, or None after telling the user why not (once per home)."""
from plugins.memory import find_provider_dir, load_memory_provider, parked_provider_notice
provider = load_memory_provider(name)
if provider is None:
parked = parked_provider_notice(name)
if parked:
# Installed but turned off (by the user or by an update): say so, never re-install it.
_warn_once(name, parked, say=say, kind="parked")
return None
if find_provider_dir(name) is None:
# The provider left core for the catalog (or was never installed): fetch it once.
from hermes_cli.memory_provider_migration import recover_at_startup
if recover_at_startup(name, say=say):
provider = load_memory_provider(name)
# Installed (now or before) yet not loading: neither "not installed" nor an install notice explains that.
if provider is None and find_provider_dir(name) is not None:
_warn_once(name, f"⚠ Memory provider '{name}' is installed but failed to load, so external memory is off "
"for this session (built-in memory still works). The log names the error; "
"`hermes memory status` shows the plugin's state.", say=say, kind="load")
if provider is None or provider.is_available():
return provider
if _unavailable_warning_key(name) not in _warned_unavailable_providers:
# unavailable_reason() reads config/probes importlib — skip it once warned.
reason = ""
try:
reason = provider.unavailable_reason()
except Exception: # health: allow BLE001 -- a provider's own hint must never break agent init
logger.debug("unavailable_reason() failed for %s", name, exc_info=True)
_warn_memory_provider_unavailable(name, reason, say=say)
return None
def _stored_transcript_kwargs(agent) -> dict[str, Any]:
"""``resumed`` / ``message_count`` for ``MemoryProvider.initialize``. ``message_count`` is how many
messages the session store holds for this session id when the agent is built; ``resumed`` is
whether one of them is an assistant reply, i.e. an earlier turn ran. TUI/Desktop store the prompt
being submitted before the build, so a brand-new chat can count 1 and still be not resumed."""
db, sid = agent._session_db, agent.session_id
count, resumed = 0, False
if db:
with suppress(Exception):
count = int((db.get_session_list_row(sid) or {}).get("message_count") or 0)
resumed = count > 0 and db.latest_message_row_id(sid, role="assistant", require_text=False) is not None
return {"resumed": resumed, "message_count": count}
def _memory_provider_init_kwargs(agent, platform) -> dict[str, Any]:
"""Scoping kwargs for ``MemoryManager.initialize_all`` (status_callback is CLI-only:
gateway status travels a different path and the indicator no-ops without it)."""
kwargs = {
"session_id": agent.session_id,
"platform": platform or "cli",
"hermes_home": str(get_hermes_home()),
# platform="cron" (scheduler) / "subagent" (delegate_task) → providers skip writes (MemoryProvider.initialize);
# a fork that loads a provider only for the checkpoint gate names its scope (memory_agent_context).
"agent_context": getattr(agent, "memory_agent_context", None)
or (platform if platform in ("cron", "subagent") else "primary"),
}
# Documented in MemoryProvider.initialize: forks tell their shape apart by platform + parent.
if getattr(agent, "_parent_session_id", None):
kwargs["parent_session_id"] = agent._parent_session_id
if kwargs["platform"] == "cli":
kwargs["warning_callback"] = agent._emit_warning
kwargs["status_callback"] = agent._emit_status
kwargs.update(_stored_transcript_kwargs(agent))
# Session title (e.g. honcho derives chat-scoped session keys from it).
if agent._session_db:
with suppress(Exception):
_st = agent._session_db.get_session_title(agent.session_id)
if _st:
kwargs["session_title"] = _st
_source = agent._session_db.get_session_title_source(agent.session_id)
if _source:
kwargs["session_title_source"] = _source
# Gateway user/chat identity for per-user scoping (gateway_session_key: stable per-chat
# Honcho session isolation).
for _ident in _GATEWAY_IDENTITY_PARAMS:
_val = getattr(agent, f"_{_ident}")
if _val:
kwargs[_ident] = _val
if agent.session_cwd:
kwargs["cwd"] = agent.session_cwd
# Profile identity for per-profile provider scoping
with suppress(Exception):
from hermes_cli.profiles import get_active_profile_name
kwargs["agent_identity"] = get_active_profile_name()
kwargs["agent_workspace"] = "hermes"
return kwargs