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

Merge branch 'simp/r3-27-W3' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:11:46 -07:00
parent 4dc7737ab3 66c09a1a7c
commit db070745fc
6 changed files with 268 additions and 480 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/memory/hindsight/__init__.py
+176 -299
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
import asyncio
import atexit
import contextlib
import contextvars
import json
import logging
@@ -20,7 +21,6 @@ import queue
import sys
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
@@ -35,14 +35,13 @@ from tools.registry import tool_error
# Re-exported (tests patch/import these via this module).
from .embedded import ( # noqa: F401
_PORT_HEALTH_GRACE_ENV, _RETRIABLE_CONNECTION_MARKERS, _build_embedded_profile_env,
_check_local_runtime, _embedded_profile_env_path, _export_port_health_grace_timeout,
_load_simple_env, _local_runtime_hint, _materialize_embedded_profile_env,
_secure_write_profile_env, _validate_profile_env_permissions,
_check_local_runtime, _embedded_llm_api_key, _embedded_profile_env_path,
_export_port_health_grace_timeout, _load_simple_env, _local_runtime_hint, _materialize_embedded_profile_env,
)
from .settings import ( # noqa: F401
_DEFAULT_API_URL, _DEFAULT_IDLE_TIMEOUT, _DEFAULT_LOCAL_URL, _DEFAULT_RETAIN_SOURCE,
_DEFAULT_TIMEOUT, _HINDSIGHT_GLYPH, _MIN_CLIENT_VERSION, _MIN_VERSION_FOR_UPDATE_MODE_APPEND,
_OBSERVATION_SCOPE_KEYWORDS, _PROVIDER_DEFAULT_MODELS, _VALID_BUDGETS, _daemon_llm_provider,
_PROVIDER_DEFAULT_MODELS, _VALID_BUDGETS, _daemon_llm_provider,
_normalize_observation_scopes, _normalize_retain_tags, _parse_int_setting,
_resolve_bank_id_template, _sanitize_bank_segment,
)
@@ -53,14 +52,6 @@ _LOCAL_MODES = {"local", "local_embedded"}
_RETAIN_CONTEXT_DEFAULT = "conversation between Hermes Agent and the User"
@dataclass(frozen=True)
class _RecallResult:
"""Recall text + memory count for the recall indicator (0 for reflect/error)."""
text: str
count: int
def _ensure_client_dependency() -> None:
"""Lazily install the Hindsight client (``tools.lazy_deps``) before importing it."""
try:
@@ -90,11 +81,12 @@ def _maybe_upgrade_client() -> None:
outcome = install_specs([f"hindsight-client>={_MIN_CLIENT_VERSION}"], timeout=120)
if outcome.ok:
logger.info("hindsight-client upgraded to >=%s", _MIN_CLIENT_VERSION)
elif outcome.blocked:
logger.warning("Auto-upgrade unavailable: %s. Run: uv pip install 'hindsight-client>=%s'",
outcome.reason, _MIN_CLIENT_VERSION)
else:
what, why = (("unavailable", outcome.reason) if outcome.blocked
else ("failed", (outcome.stderr or "").strip() or "install error"))
logger.warning("Auto-upgrade %s: %s. Run: uv pip install 'hindsight-client>=%s'",
what, why, _MIN_CLIENT_VERSION)
logger.warning("Auto-upgrade failed: %s. Run: uv pip install 'hindsight-client>=%s'",
(outcome.stderr or "").strip() or "install error", _MIN_CLIENT_VERSION)
except Exception:
pass # packaging not available or other issue — proceed anyway
@@ -105,17 +97,6 @@ _append_capability_cache: Dict[str, bool] = {}
_append_capability_lock = threading.Lock()
def _meets_minimum_version(actual: str | None, required: str) -> bool:
"""True if *actual* >= *required* (semver). False on missing/invalid."""
if not actual:
return False
try:
from packaging.version import Version
return Version(actual) >= Version(required)
except Exception:
return False
def _fetch_hindsight_api_version(api_url: str, api_key: str | None = None,
timeout: float = 5.0) -> str | None:
"""GET ``<api_url>/version`` -> version string, or None on any failure (= legacy API)."""
@@ -123,16 +104,14 @@ def _fetch_hindsight_api_version(api_url: str, api_key: str | None = None,
if not api_url:
return None
url = api_url.rstrip("/") + "/version"
req = urllib.request.Request(url)
if api_key:
req.add_header("Authorization", f"Bearer {api_key}")
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {api_key}"} if api_key else {})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310
data = json.loads(resp.read().decode("utf-8", errors="replace"))
except Exception as exc:
logger.debug("Hindsight /version probe failed for %s: %s", url, exc)
return None
version = data.get("version") or data.get("api_version") if isinstance(data, dict) else None
version = (data.get("version") or data.get("api_version")) if isinstance(data, dict) else None
return str(version) if version else None
@@ -145,21 +124,23 @@ def _check_api_supports_update_mode_append(api_url: str, api_key: str | None = N
if api_url in _append_capability_cache:
return _append_capability_cache[api_url]
version = _fetch_hindsight_api_version(api_url, api_key)
supported = _meets_minimum_version(version, _MIN_VERSION_FOR_UPDATE_MODE_APPEND)
try: # missing/invalid version -> unsupported
from packaging.version import Version
supported = bool(version) and Version(version) >= Version(_MIN_VERSION_FOR_UPDATE_MODE_APPEND)
except Exception:
supported = False
with _append_capability_lock:
# A concurrent probe may have filled the cache meanwhile; its answer wins.
supported = _append_capability_cache.setdefault(api_url, supported)
if supported:
logger.debug("Hindsight API %s version %s supports update_mode='append'", api_url, version)
else:
logger.warning(
"Hindsight API at %s reports version %r, older than %s. "
"Falling back to per-process document_id — retains across "
"processes/sessions create separate documents instead of "
"appending to a session-scoped one. Upgrade Hindsight to "
"%s+ to enable update_mode='append' deduplication.",
api_url, version, _MIN_VERSION_FOR_UPDATE_MODE_APPEND, _MIN_VERSION_FOR_UPDATE_MODE_APPEND,
)
logger.warning("Hindsight API at %s reports version %r, older than %s. "
"Falling back to per-process document_id — retains across "
"processes/sessions create separate documents instead of "
"appending to a session-scoped one. Upgrade Hindsight to "
"%s+ to enable update_mode='append' deduplication.",
api_url, version, _MIN_VERSION_FOR_UPDATE_MODE_APPEND, _MIN_VERSION_FOR_UPDATE_MODE_APPEND)
return supported
@@ -179,13 +160,10 @@ def _get_loop() -> asyncio.AbstractEventLoop:
with _loop_lock:
if _loop is not None and _loop.is_running():
return _loop
_loop = asyncio.new_event_loop()
def _run():
asyncio.set_event_loop(_loop)
_loop.run_forever()
_loop_thread = threading.Thread(target=_run, daemon=True, name="hindsight-loop")
loop = _loop = asyncio.new_event_loop()
_loop_thread = threading.Thread(
target=lambda: (asyncio.set_event_loop(loop), loop.run_forever()), daemon=True, name="hindsight-loop",
)
_loop_thread.start()
return _loop
@@ -204,9 +182,7 @@ def _context_thread(target, name: str) -> threading.Thread:
Threads start with an EMPTY Context; under multiplex_profiles get_secret fails
closed without the profile's secret scope + HERMES_HOME override. (The shared
loop needs no wrap: run_coroutine_threadsafe inherits the submitter's context.)"""
return threading.Thread(
target=contextvars.copy_context().run, args=(target,), daemon=True, name=name,
)
return threading.Thread(target=contextvars.copy_context().run, args=(target,), daemon=True, name=name)
RETAIN_SCHEMA = {
@@ -220,21 +196,15 @@ RETAIN_SCHEMA = {
"properties": {
"content": {"type": "string", "description": "The information to store."},
"context": {"type": "string", "description": "Short label (e.g. 'user preference', 'project decision')."},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Optional per-call tags to merge with configured default retain tags.",
},
"occurred_at": {
"type": "string",
"description": (
"When the remembered event actually happened, as an ISO-8601 date "
"or datetime (e.g. '2026-08-20' or '2026-08-20T14:30:00+02:00'). "
"Pass this whenever the memory references a specific event time "
"('yesterday', 'last Tuesday', 'on March 3rd') so Hindsight can "
"anchor it on the timeline. Omit for timeless facts/preferences."
),
},
"tags": {"type": "array", "items": {"type": "string"},
"description": "Optional per-call tags to merge with configured default retain tags."},
"occurred_at": {"type": "string", "description": (
"When the remembered event actually happened, as an ISO-8601 date "
"or datetime (e.g. '2026-08-20' or '2026-08-20T14:30:00+02:00'). "
"Pass this whenever the memory references a specific event time "
"('yesterday', 'last Tuesday', 'on March 3rd') so Hindsight can "
"anchor it on the timeline. Omit for timeless facts/preferences."
)},
},
"required": ["content"],
},
@@ -246,13 +216,8 @@ RECALL_SCHEMA = {
"Search long-term memory. Returns memories ranked by relevance using "
"semantic search, keyword matching, entity graph traversal, and reranking."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "What to search for."},
},
"required": ["query"],
},
"parameters": {"type": "object", "required": ["query"],
"properties": {"query": {"type": "string", "description": "What to search for."}}},
}
REFLECT_SCHEMA = {
@@ -261,13 +226,8 @@ REFLECT_SCHEMA = {
"Synthesize a reasoned answer from long-term memories. Unlike recall, "
"this reasons across all stored memories to produce a coherent response."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "The question to reflect on."},
},
"required": ["query"],
},
"parameters": {"type": "object", "required": ["query"],
"properties": {"query": {"type": "string", "description": "The question to reflect on."}}},
}
@@ -276,10 +236,8 @@ def _load_config() -> dict:
(legacy, shared), else environment variables."""
for path in (get_hermes_home() / "hindsight" / "config.json", Path.home() / ".hindsight" / "config.json"):
if path.exists():
try:
with contextlib.suppress(Exception):
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
pass
return {
"mode": os.environ.get("HINDSIGHT_MODE", "cloud"),
"apiKey": get_secret("HINDSIGHT_API_KEY", ""),
@@ -295,11 +253,6 @@ def _load_config() -> dict:
}
def _utc_timestamp() -> str:
"""UTC write/audit time for retain metadata."""
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def _event_timestamp() -> str:
"""Configured Hermes event time with an explicit UTC offset."""
event_time = _hermes_now()
@@ -333,22 +286,19 @@ _SYSTEM_PROMPT_TAILS = {
"Use hindsight_recall to search, hindsight_reflect for synthesis, "
"hindsight_retain to store facts."),
}
_TOOL_ERRORS = {
"hindsight_retain": "Failed to store memory",
"hindsight_recall": "Failed to search memory",
"hindsight_reflect": "Failed to reflect",
}
class HindsightMemoryProvider(MemoryProvider):
"""Hindsight long-term memory with knowledge graph and multi-strategy retrieval."""
# Each server-side op status poll is a round trip — coarser than the 0.05s queue poll.
_RETAIN_OP_POLL_INTERVAL_S = 0.5
def backup_paths(self) -> List[str]:
"""Legacy shared config + embedded-mode profile env files live under ~/.hindsight."""
try:
with contextlib.suppress(Exception):
return [str(Path.home() / ".hindsight")]
except Exception:
return []
return []
def __init__(self):
self._config = self._api_key = self._client = None
@@ -371,13 +321,10 @@ class HindsightMemoryProvider(MemoryProvider):
self._sync_thread = None # legacy alias external callers may join; points at the writer
self._shutting_down = threading.Event()
self._atexit_registered = False
self._auto_retain = self._retain_async = self._retain_indicator = True
self._retain_every_n_turns = 1
self._retain_tags: List[str] = []
self._tags: list[str] | None = None
self._retain_source = _DEFAULT_RETAIN_SOURCE
self._retain_user_prefix, self._retain_assistant_prefix = "User", "Assistant"
self._retain_context = _RETAIN_CONTEXT_DEFAULT
self._turn_counter = self._turn_index = 0
self._session_turns: list[str] = [] # ALL turns for the session
self._last_retained_turn_count = 0 # append-mode delta watermark
@@ -387,29 +334,14 @@ class HindsightMemoryProvider(MemoryProvider):
self._pending_retain_ops: set[str] = set()
self._pending_retain_ops_lock = threading.Lock()
self._retain_ops_bank_id = ""
# Each status poll is a server round trip — coarser than the 0.05s queue poll.
self._RETAIN_OP_POLL_INTERVAL_S = 0.5
# The next turn's warm prefetch could read BEFORE an async retain is
# recall-visible; when True it first waits (bounded, off the reply path)
# for the queue to drain AND the server-side op(s) to complete.
self._prefetch_waits_for_retain = True
self._prefetch_retain_drain_timeout = 10.0
self._apply_retain_policy({})
# Recall: pending prefetch block + count, and the indicator state (recall_status()).
self._prefetch_result, self._prefetch_count = "", 0
self._prefetch_lock = threading.Lock()
self._prefetch_thread = None
self._last_recall_returned, self._last_recall_count = False, 0
self._auto_recall = self._recall_indicator = True
self._recall_sync = False
self._recall_tags: list[str] | None = None
self._recall_tags_match = "any"
self._recall_max_tokens, self._recall_max_input_chars = 4096, 800
# Observation-only by default: observations are Hindsight's consolidated,
# deduplicated layer; raw world/experience facts re-ship the evidence
# they summarize and burn the recall_max_tokens budget.
self._recall_types: list[str] = ["observation"]
self._recall_prompt_preamble = ""
self._apply_recall_settings({})
@property
def name(self) -> str:
@@ -422,8 +354,7 @@ class HindsightMemoryProvider(MemoryProvider):
if mode in _LOCAL_MODES:
return _check_local_runtime()[0]
return mode == "local_external" or bool(
_cloud_api_key(cfg) or cfg.get("api_url") or os.environ.get("HINDSIGHT_API_URL", "")
)
_cloud_api_key(cfg) or cfg.get("api_url") or os.environ.get("HINDSIGHT_API_URL", ""))
except Exception:
return False
@@ -431,11 +362,10 @@ class HindsightMemoryProvider(MemoryProvider):
"""Install hint for an unavailable local_embedded runtime (is_available() gates
initialize() out, so the hint it would log never fires; agent_init shows this)."""
try:
mode = _load_config().get("mode", "cloud")
if _load_config().get("mode", "cloud") not in _LOCAL_MODES:
return ""
except Exception:
return ""
if mode not in _LOCAL_MODES:
return ""
available, reason = _check_local_runtime()
return "" if available else _local_runtime_hint(reason).strip()
@@ -444,10 +374,10 @@ class HindsightMemoryProvider(MemoryProvider):
from utils import atomic_json_write
config_path = Path(hermes_home) / "hindsight" / "config.json"
config_path.parent.mkdir(parents=True, exist_ok=True)
try:
existing = json.loads(config_path.read_text(encoding="utf-8")) if config_path.exists() else {}
except Exception:
existing = {}
existing = {}
if config_path.exists():
with contextlib.suppress(Exception):
existing = json.loads(config_path.read_text(encoding="utf-8"))
existing.update(values)
atomic_json_write(config_path, existing, mode=0o600)
@@ -524,13 +454,9 @@ class HindsightMemoryProvider(MemoryProvider):
self._idle_timeout = self._int_setting(
"idle_timeout", "HINDSIGHT_IDLE_TIMEOUT", _DEFAULT_IDLE_TIMEOUT, env_default=self._idle_timeout,
)
kwargs = dict(
profile=cfg.get("profile", "hermes"),
llm_provider=llm_provider,
llm_api_key=cfg.get("llmApiKey") or cfg.get("llm_api_key") or get_secret("HINDSIGHT_LLM_API_KEY", ""),
llm_model=cfg.get("llm_model", ""),
idle_timeout=self._idle_timeout,
)
kwargs = dict(profile=cfg.get("profile", "hermes"), llm_provider=llm_provider,
llm_api_key=_embedded_llm_api_key(cfg), llm_model=cfg.get("llm_model", ""),
idle_timeout=self._idle_timeout)
if self._llm_base_url:
kwargs["llm_base_url"] = self._llm_base_url
return HindsightEmbedded(**kwargs)
@@ -548,9 +474,7 @@ class HindsightMemoryProvider(MemoryProvider):
def _get_client(self):
"""Return the cached Hindsight client (created once, reused)."""
if self._client is None:
self._client = (
self._new_embedded_client() if self._mode == "local_embedded" else self._new_cloud_client()
)
self._client = self._new_embedded_client() if self._mode == "local_embedded" else self._new_cloud_client()
return self._client
def _run_sync(self, coro):
@@ -566,10 +490,7 @@ class HindsightMemoryProvider(MemoryProvider):
text = f"{type(exc).__name__}: {exc}".lower()
if self._mode != "local_embedded" or not any(m in text for m in _RETRIABLE_CONNECTION_MARKERS):
raise
logger.info(
"Hindsight embedded daemon appears unreachable; recreating client and retrying once: %s",
exc,
)
logger.info("Hindsight embedded daemon appears unreachable; recreating client and retrying once: %s", exc)
self._client = None
self._client = client = self._get_client()
return self._run_sync(operation(client))
@@ -578,8 +499,7 @@ class HindsightMemoryProvider(MemoryProvider):
def _ensure_writer(self) -> None:
"""Lazy-start the single retain-writer thread (tools-only providers never pay for it)."""
thread = self._writer_thread
if thread is not None and thread.is_alive():
if (thread := self._writer_thread) is not None and thread.is_alive():
return
# A previous writer may have exited after shutdown(); allow the fresh one to drain.
self._shutting_down.clear()
@@ -607,18 +527,16 @@ class HindsightMemoryProvider(MemoryProvider):
try:
if job is _WRITER_SENTINEL:
return
try:
job()
except Exception as exc:
logger.warning("Hindsight retain failed: %s", exc, exc_info=True)
job()
except Exception as exc:
logger.warning("Hindsight retain failed: %s", exc, exc_info=True)
finally:
self._retain_queue.task_done()
def _atexit_shutdown(self) -> None:
if self._shutting_down.is_set():
return
try:
self.shutdown()
if not self._shutting_down.is_set():
self.shutdown()
except Exception as exc:
logger.debug("Hindsight atexit shutdown failed: %s", exc)
@@ -627,12 +545,10 @@ class HindsightMemoryProvider(MemoryProvider):
(pending until recall-visible). No id (older API / sync completion) leaves
only the local queue drain as a signal."""
raw_ids = [getattr(retain_response, "operation_id", None), *(getattr(retain_response, "operation_ids", None) or [])]
ids = [str(op) for op in raw_ids if op]
if not ids:
return
self._retain_ops_bank_id = bank_id
with self._pending_retain_ops_lock:
self._pending_retain_ops.update(ids)
if ids := [str(op) for op in raw_ids if op]:
self._retain_ops_bank_id = bank_id
with self._pending_retain_ops_lock:
self._pending_retain_ops.update(ids)
def _is_retain_op_complete(self, bank_id: str, op_id: str) -> bool:
"""True when a server-side retain op is done or gone (completed ops are evicted,
@@ -658,25 +574,23 @@ class HindsightMemoryProvider(MemoryProvider):
server-side async ops complete (async retain returns on acceptance, not
durability). False on timeout/shutdown."""
deadline = None if timeout <= 0 else time.monotonic() + timeout
expired = lambda: deadline is not None and time.monotonic() >= deadline # noqa: E731
while self._retain_queue.unfinished_tasks > 0:
if self._shutting_down.is_set():
return False
if deadline is not None and time.monotonic() >= deadline:
if expired():
logger.debug("Prefetch: retain drain timed out after %.1fs (%d pending)",
timeout, self._retain_queue.unfinished_tasks)
return False
time.sleep(0.05)
return self._wait_for_server_retain_ops(deadline, timeout)
return self._wait_for_server_retain_ops(expired, timeout)
def _wait_for_server_retain_ops(self, deadline: float | None, timeout: float) -> bool:
"""Poll tracked async retain ops until complete or *deadline* (monotonic; None
= unbounded). Ops still pending at the deadline are DROPPED: keeping them
def _wait_for_server_retain_ops(self, _expired: Callable[[], bool], timeout: float) -> bool:
"""Poll tracked async retain ops until complete or *_expired()* (deadline
predicate). Ops still pending at the deadline are DROPPED: keeping them
would let a permanently failing status endpoint burn the full timeout on
EVERY later prefetch (a per-turn latency penalty via prefetch()'s bounded
join). Trades a possibly-stale recall for liveness; WARNING once per prefetch."""
def _expired() -> bool:
return deadline is not None and time.monotonic() >= deadline
while True:
with self._pending_retain_ops_lock:
bank_id = self._retain_ops_bank_id or self._bank_id
@@ -701,28 +615,23 @@ class HindsightMemoryProvider(MemoryProvider):
if dropped:
self._pending_retain_ops.clear()
if dropped:
logger.warning(
"Prefetch: server retain visibility timed out after %.1fs; "
"dropping %d unresolved op(s) so later prefetches stay "
"bounded (recall may miss the just-completed turn)",
timeout, dropped,
)
logger.warning("Prefetch: server retain visibility timed out after %.1fs; "
"dropping %d unresolved op(s) so later prefetches stay "
"bounded (recall may miss the just-completed turn)", timeout, dropped)
return False
time.sleep(self._RETAIN_OP_POLL_INTERVAL_S)
# -- retain target -----------------------------------------------------------
def _probe_url(self) -> str:
"""/version probe URL: the embedded client's dynamic per-profile port when running, else api_url."""
url = getattr(self._client, "url", None) if self._mode == "local_embedded" else None
return str(url) if url else (self._api_url or "")
def _resolve_retain_target(self, fallback_document_id: str) -> tuple[str, str | None]:
"""(document_id, update_mode) from live API capability: >= 0.5.0 reuses the
stable session-scoped id with ``update_mode='append'``; older APIs get
*fallback_document_id* (per-process unique) and no update_mode — the only
way the resume-overwrite fix works there."""
if self._session_id and _check_api_supports_update_mode_append(self._probe_url(), self._api_key):
way the resume-overwrite fix works there. The /version probe targets the
embedded client's dynamic per-profile port when running, else api_url."""
url = getattr(self._client, "url", None) if self._mode == "local_embedded" else None
probe_url = str(url) if url else (self._api_url or "")
if self._session_id and _check_api_supports_update_mode_append(probe_url, self._api_key):
return self._session_id, "append"
return fallback_document_id, None
@@ -753,21 +662,18 @@ class HindsightMemoryProvider(MemoryProvider):
_export_port_health_grace_timeout(cfg)
available, reason = _check_local_runtime()
if not available:
logger.warning(
"Hindsight local mode disabled because its runtime could not be imported: %s.%s",
reason, _local_runtime_hint(reason),
)
logger.warning("Hindsight local mode disabled because its runtime could not be imported: %s.%s",
reason, _local_runtime_hint(reason))
self._mode = "disabled"
return
self._apply_connection_settings(cfg)
self._apply_retain_settings(cfg)
self._apply_recall_settings(cfg)
try:
client_version = "unknown"
with contextlib.suppress(Exception):
from importlib.metadata import version as pkg_version
client_version = pkg_version("hindsight-client")
except Exception:
client_version = "unknown"
logger.info("Hindsight initialized: mode=%s, api_url=%s, bank=%s, budget=%s, memory_mode=%s, prefetch_method=%s, client=%s",
self._mode, self._api_url, self._bank_id, self._budget, self._memory_mode, self._prefetch_method, client_version)
if self._bank_id_template:
@@ -808,21 +714,23 @@ class HindsightMemoryProvider(MemoryProvider):
self._bank_retain_mission = cfg.get("bank_retain_mission") or None
def _apply_retain_settings(self, cfg: dict) -> None:
def _text(key: str, env_var: str, default: str) -> str:
return str(cfg.get(key) or os.environ.get(env_var, default)).strip()
def _cfg_or_env(key: str, env_var: str, default: str = "") -> Any:
return cfg.get(key) or os.environ.get(env_var, default)
self._retain_tags = _normalize_retain_tags(
cfg.get("retain_tags") or os.environ.get("HINDSIGHT_RETAIN_TAGS", "")
)
self._retain_tags = _normalize_retain_tags(_cfg_or_env("retain_tags", "HINDSIGHT_RETAIN_TAGS"))
self._tags = self._retain_tags or None
self._observation_scopes = _normalize_observation_scopes(
cfg.get("observation_scopes") or os.environ.get("HINDSIGHT_RETAIN_OBSERVATION_SCOPES", "")
)
self._retain_source = _text("retain_source", "HINDSIGHT_RETAIN_SOURCE", _DEFAULT_RETAIN_SOURCE)
self._retain_user_prefix = _text("retain_user_prefix", "HINDSIGHT_RETAIN_USER_PREFIX", "User") or "User"
_cfg_or_env("observation_scopes", "HINDSIGHT_RETAIN_OBSERVATION_SCOPES"))
self._retain_source = str(_cfg_or_env("retain_source", "HINDSIGHT_RETAIN_SOURCE", _DEFAULT_RETAIN_SOURCE)).strip()
self._retain_user_prefix = str(_cfg_or_env("retain_user_prefix", "HINDSIGHT_RETAIN_USER_PREFIX", "User")).strip() or "User"
self._retain_assistant_prefix = (
_text("retain_assistant_prefix", "HINDSIGHT_RETAIN_ASSISTANT_PREFIX", "Assistant") or "Assistant"
str(_cfg_or_env("retain_assistant_prefix", "HINDSIGHT_RETAIN_ASSISTANT_PREFIX", "Assistant")).strip()
or "Assistant"
)
self._apply_retain_policy(cfg)
def _apply_retain_policy(self, cfg: dict) -> None:
"""Pure-config retain knobs (no env/secret reads; ``{}`` yields the defaults)."""
self._auto_retain = cfg.get("auto_retain", True)
self._retain_every_n_turns = max(1, int(cfg.get("retain_every_n_turns", 1)))
self._retain_context = cfg.get("retain_context", _RETAIN_CONTEXT_DEFAULT)
@@ -830,25 +738,29 @@ class HindsightMemoryProvider(MemoryProvider):
# On by default so the user SEES memory working whether or not the model
# mentions it; off switch for customer-facing agents (recall_indicator too).
self._retain_indicator = bool(cfg.get("retain_indicator", True))
# The next turn's warm prefetch could read BEFORE an async retain is
# recall-visible; when True it first waits (bounded, off the reply path)
# for the queue to drain AND the server-side op(s) to complete.
self._prefetch_waits_for_retain = cfg.get("prefetch_waits_for_retain", True)
self._prefetch_retain_drain_timeout = float(cfg.get("prefetch_retain_drain_timeout", 10.0))
def _apply_recall_settings(self, cfg: dict) -> None:
"""Recall knobs are pure config too (``{}`` yields the defaults)."""
self._recall_tags = cfg.get("recall_tags") or None
self._recall_tags_match = cfg.get("recall_tags_match", "any")
self._auto_recall = cfg.get("auto_recall", True)
self._recall_sync = bool(cfg.get("recall_sync", False))
self._recall_max_tokens = int(cfg.get("recall_max_tokens", 4096))
self._recall_max_input_chars = int(cfg.get("recall_max_input_chars", 800))
# None -> observation-only; a comma-separated string is accepted for
# parity with recall_tags; an explicit list broadens or disables the filter.
# None -> observation-only (Hindsight's consolidated, deduplicated layer; raw
# world/experience facts re-ship the evidence they summarize and burn the
# recall_max_tokens budget); a comma-separated string is accepted for parity
# with recall_tags; an explicit list broadens or disables the filter.
configured_types = cfg.get("recall_types")
if configured_types is None:
self._recall_types = ["observation"]
elif isinstance(configured_types, str):
if isinstance(configured_types, str):
self._recall_types = [t.strip() for t in configured_types.split(",") if t.strip()]
else:
self._recall_types = list(configured_types) or ["observation"]
self._recall_types = list([] if configured_types is None else configured_types) or ["observation"]
self._recall_prompt_preamble = cfg.get("recall_prompt_preamble", "")
self._recall_indicator = bool(cfg.get("recall_indicator", True))
@@ -858,18 +770,14 @@ class HindsightMemoryProvider(MemoryProvider):
# retries forever, reloading embedding models (~958MB RAM, ~33% CPU)
# with no user-visible error.
if hasattr(os, "geteuid") and os.geteuid() == 0:
msg = (
"Hindsight local_embedded mode cannot run as root "
"(PostgreSQL initdb refuses root). Skipping the embedded "
"memory daemon. Run Hermes as a non-root user, or switch "
"to cloud / local_external mode via 'hermes memory setup'."
)
msg = ("Hindsight local_embedded mode cannot run as root "
"(PostgreSQL initdb refuses root). Skipping the embedded "
"memory daemon. Run Hermes as a non-root user, or switch "
"to cloud / local_external mode via 'hermes memory setup'.")
logger.warning(msg)
# Also print: otherwise the user would only see Hermes get sluggish.
try:
with contextlib.suppress(Exception):
print(f" ⚠ {msg}", file=sys.stderr, flush=True)
except Exception:
pass
self._mode = "disabled"
return
_context_thread(self._daemon_start_worker, "hindsight-daemon-start").start()
@@ -879,11 +787,9 @@ class HindsightMemoryProvider(MemoryProvider):
log_path = get_hermes_home() / "logs" / "hindsight-embed.log"
log_path.parent.mkdir(parents=True, exist_ok=True)
def _log(text: str, exc: bool = False) -> None:
def _log(text: str) -> None:
with open(log_path, "a", encoding="utf-8") as f:
f.write(text)
if exc:
traceback.print_exc(file=f)
try:
# Rich console -> our log file (redirecting global fds would capture other threads).
@@ -902,10 +808,10 @@ class HindsightMemoryProvider(MemoryProvider):
client._ensure_started()
_log("\n=== Daemon started successfully ===\n")
except Exception as e:
_log(f"\n=== Daemon startup failed: {e} ===\n", exc=True)
_log(f"\n=== Daemon startup failed: {e} ===\n" + traceback.format_exc())
def system_prompt_block(self) -> str:
mode = self._memory_mode if self._memory_mode in ("context", "tools") else "hybrid"
mode = self._memory_mode if self._memory_mode in _SYSTEM_PROMPT_TAILS else "hybrid"
label = "" if mode == "hybrid" else f" ({mode} mode)"
return f"# Hindsight Memory\nActive{label}. Bank: {self._bank_id}, budget: {self._budget}.\n{_SYSTEM_PROMPT_TAILS[mode]}"
@@ -913,24 +819,19 @@ class HindsightMemoryProvider(MemoryProvider):
def _recall_disabled(self) -> bool:
"""Guards shared by the async and synchronous recall paths."""
for skip, why in ((self._memory_mode == "tools", "tools-only mode"), (not self._auto_recall, "auto_recall disabled"),
(self._shutting_down.is_set(), "shutting down")):
if skip:
logger.debug("Prefetch: skipped (%s)", why)
return True
return False
why = ("tools-only mode" if self._memory_mode == "tools" else "auto_recall disabled" if not self._auto_recall
else "shutting down" if self._shutting_down.is_set() else None)
if why:
logger.debug("Prefetch: skipped (%s)", why)
return why is not None
def _recall_kwargs(self, query: str) -> dict:
def _recall(self, query: str) -> list:
kwargs: dict = {"bank_id": self._bank_id, "query": query, "budget": self._budget, "max_tokens": self._recall_max_tokens}
if self._recall_tags:
kwargs.update(tags=self._recall_tags, tags_match=self._recall_tags_match)
if self._recall_types:
kwargs["types"] = self._recall_types
return kwargs
def _recall(self, query: str) -> list:
recall_kwargs = self._recall_kwargs(query)
resp = self._run_hindsight_operation(lambda client: client.arecall(**recall_kwargs))
resp = self._run_hindsight_operation(lambda client: client.arecall(**kwargs))
return resp.results or []
def _reflect(self, query: str) -> str | None:
@@ -939,22 +840,23 @@ class HindsightMemoryProvider(MemoryProvider):
)
return resp.text
def _do_recall(self, query: str) -> _RecallResult:
"""One recall/reflect for *query* (background prefetch and ``recall_sync`` paths)."""
if self._recall_max_input_chars and len(query) > self._recall_max_input_chars:
def _do_recall(self, query: str) -> tuple[str, int]:
"""One recall/reflect for *query* (background prefetch and ``recall_sync`` paths)
-> (text, memory count); the count is 0 for reflect (synthesis) and on error."""
if self._recall_max_input_chars:
query = query[:self._recall_max_input_chars]
try:
if self._prefetch_method == "reflect":
logger.debug("Recall: calling reflect (bank=%s, query_len=%d)", self._bank_id, len(query))
return _RecallResult(self._reflect(query) or "", 0) # synthesis -> no discrete count
return self._reflect(query) or "", 0
logger.debug("Recall: calling recall (bank=%s, query_len=%d, budget=%s)",
self._bank_id, len(query), self._budget)
results = self._recall(query)
logger.debug("Recall: returned %d results", len(results))
return _RecallResult("\n".join(f"- {r.text}" for r in results if r.text), len(results))
return "\n".join(f"- {r.text}" for r in results if r.text), len(results)
except Exception as e:
logger.debug("Hindsight recall failed: %s", e, exc_info=True)
return _RecallResult("", 0)
return "", 0
def _finish_prefetch(self, result: str, count: int) -> str:
"""Record indicator state (cleared on empty turns, never a stale count); format the block."""
@@ -971,17 +873,17 @@ class HindsightMemoryProvider(MemoryProvider):
return f"{header}\n\n{result}"
def _join_prefetch(self, timeout: float, *, log: bool = False) -> None:
if self._prefetch_thread and self._prefetch_thread.is_alive():
if log:
logger.debug("Prefetch: waiting for background thread to complete")
self._prefetch_thread.join(timeout=timeout)
if not (self._prefetch_thread and self._prefetch_thread.is_alive()):
return
if log:
logger.debug("Prefetch: waiting for background thread to complete")
self._prefetch_thread.join(timeout=timeout)
def prefetch(self, query: str, *, session_id: str = "") -> str:
# Opt-in: recall synchronously against the *current* message so the
# injected memories match this turn's query, not the previous turn's.
if self._recall_sync:
recalled = _RecallResult("", 0) if self._recall_disabled() else self._do_recall(query)
return self._finish_prefetch(recalled.text, recalled.count)
return self._finish_prefetch(*(("", 0) if self._recall_disabled() else self._do_recall(query)))
# Default: the background worker's result for the previous turn (capped join).
self._join_prefetch(3.0, log=True)
with self._prefetch_lock:
@@ -1005,10 +907,10 @@ class HindsightMemoryProvider(MemoryProvider):
# retain to be recall-visible so the warmed context includes it.
if self._prefetch_waits_for_retain:
self._wait_for_retains_drained(self._prefetch_retain_drain_timeout)
recalled = self._do_recall(query)
if recalled.text:
text, count = self._do_recall(query)
if text:
with self._prefetch_lock:
self._prefetch_result, self._prefetch_count = recalled.text, recalled.count
self._prefetch_result, self._prefetch_count = text, count
self._prefetch_thread = _context_thread(_run, "hindsight-prefetch")
self._prefetch_thread.start()
@@ -1016,14 +918,14 @@ class HindsightMemoryProvider(MemoryProvider):
# -- retain ------------------------------------------------------------------
def _build_turn_messages(self, user_content: str, assistant_content: str) -> List[Dict[str, str]]:
# One conversation turn -> both messages share the turn-level event timestamp.
now = _event_timestamp()
now = _event_timestamp() # one turn -> both messages share the event timestamp
return [{"role": role, "content": f"{prefix}: {content}", "timestamp": now} for role, prefix, content in
(("user", self._retain_user_prefix, user_content), ("assistant", self._retain_assistant_prefix, assistant_content))]
def _build_metadata(self, *, message_count: int, turn_index: int) -> Dict[str, str]:
metadata: Dict[str, str] = {
"retained_at": _utc_timestamp(),
# UTC write/audit time (event time lives on the item timestamp).
"retained_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z"),
"message_count": str(message_count),
"turn_index": str(turn_index),
}
@@ -1044,24 +946,16 @@ class HindsightMemoryProvider(MemoryProvider):
"timestamp": (occurred_at or "").strip() or _event_timestamp(),
}
merged_tags = _normalize_retain_tags(list(self._retain_tags) + _normalize_retain_tags(tags))
for key, value in (("context", context), ("update_mode", update_mode)):
if value is not None:
item[key] = value
for key, value in (("tags", merged_tags), ("observation_scopes", self._observation_scopes)):
if value:
item[key] = value
item.update({k: v for k, v in (("context", context), ("update_mode", update_mode)) if v is not None})
item.update({k: v for k, v in (("tags", merged_tags), ("observation_scopes", self._observation_scopes)) if v})
return item
def _lineage_tags(self) -> list[str]:
pairs = (("session", self._session_id), ("parent", self._parent_session_id))
return [f"{kind}:{sid}" for kind, sid in pairs if sid]
def _retain_batch(self, item: dict, *, bank_id: str, document_id: str | None = None,
retain_async: bool | None = None):
"""Dispatch one item via aretain_batch (bank_id/document_id/retain_async are
call-level args, never item keys)."""
kwargs: Dict[str, Any] = {"bank_id": bank_id, "items": [item]}
kwargs.update({k: v for k, v in (("document_id", document_id), ("retain_async", retain_async)) if v is not None})
kwargs: Dict[str, Any] = {"bank_id": bank_id, "items": [item], "document_id": document_id, "retain_async": retain_async}
kwargs = {k: v for k, v in kwargs.items() if v is not None}
return self._run_hindsight_operation(lambda client: client.aretain_batch(**kwargs))
def _make_turn_retain_job(self, turns: list[str], *, document_id: str, update_mode: str | None,
@@ -1070,7 +964,8 @@ class HindsightMemoryProvider(MemoryProvider):
writer runs after later sync_turn() calls mutate _session_turns/_turn_index/_session_id."""
content = "[" + ",".join(turns) + "]"
metadata = self._build_metadata(message_count=len(turns) * 2, turn_index=self._turn_index)
tags = self._lineage_tags() or None
lineage = (("session", self._session_id), ("parent", self._parent_session_id))
tags = [f"{kind}:{sid}" for kind, sid in lineage if sid] or None
bank_id, retain_async, retain_context = self._bank_id, self._retain_async, self._retain_context
def _job() -> None:
@@ -1090,17 +985,16 @@ class HindsightMemoryProvider(MemoryProvider):
def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
"""Enqueue a retain for the current turn (non-blocking; writer thread). Dropped
once shutdown() fired so post-exit retains never reach aiohttp during teardown."""
for skip, why in ((not self._auto_retain, "auto_retain disabled"), (self._shutting_down.is_set(), "shutting down")):
if skip:
logger.debug("sync_turn: skipped (%s)", why)
return
why = "auto_retain disabled" if not self._auto_retain else "shutting down" if self._shutting_down.is_set() else None
if why:
logger.debug("sync_turn: skipped (%s)", why)
return
if session_id:
self._session_id = str(session_id).strip()
self._session_turns.append(json.dumps(self._build_turn_messages(user_content, assistant_content), ensure_ascii=False))
self._turn_counter = self._turn_index = self._turn_counter + 1
remainder = self._turn_counter % self._retain_every_n_turns
if remainder:
if remainder := self._turn_counter % self._retain_every_n_turns:
logger.debug("sync_turn: buffered turn %d (will retain at turn %d)",
self._turn_counter, self._turn_counter + (self._retain_every_n_turns - remainder))
return
@@ -1119,7 +1013,12 @@ class HindsightMemoryProvider(MemoryProvider):
job = self._make_turn_retain_job(turns_to_retain, document_id=document_id,
update_mode=update_mode, label="retain")
# Indicator fires only past every skip/buffer gate: solely on turns that persist.
self._emit_saving_indicator()
# Model-independent status line; no-op without retain_indicator/status channel.
if self._retain_indicator and self._status_callback is not None:
try:
self._status_callback(f"{_HINDSIGHT_GLYPH} Hindsight — saving to memory…")
except Exception:
logger.debug("Retain indicator emit failed (non-fatal)", exc_info=True)
self._enqueue_retain(job)
# Advance the watermark only after the delta is queued so a later retain
# doesn't re-ship turns already handed to the writer.
@@ -1132,16 +1031,6 @@ class HindsightMemoryProvider(MemoryProvider):
self._register_atexit()
self._retain_queue.put(job)
def _emit_saving_indicator(self) -> None:
"""Model-independent "saving to memory" status line; no-op without
``retain_indicator``/status channel; never raises."""
if not self._retain_indicator or self._status_callback is None:
return
try:
self._status_callback(f"{_HINDSIGHT_GLYPH} Hindsight — saving to memory…")
except Exception:
logger.debug("Retain indicator emit failed (non-fatal)", exc_info=True)
# -- tools -------------------------------------------------------------------
def get_tool_schemas(self) -> List[Dict[str, Any]]:
@@ -1163,9 +1052,7 @@ class HindsightMemoryProvider(MemoryProvider):
self._bank_id, len(query), self._budget)
results = self._recall(query)
logger.debug("Tool hindsight_recall: %d results", len(results))
if not results:
return "No relevant memories found."
return "\n".join(f"{i}. {r.text}" for i, r in enumerate(results, 1))
return "\n".join(f"{i}. {r.text}" for i, r in enumerate(results, 1)) or "No relevant memories found."
def _tool_reflect(self, args: dict) -> str:
query = args["query"]
@@ -1175,24 +1062,24 @@ class HindsightMemoryProvider(MemoryProvider):
logger.debug("Tool hindsight_reflect: response_len=%d", len(text))
return text or "No relevant memories found."
# tool name -> (required arg, handler)
# tool name -> (required arg, handler, user-facing failure prefix)
_TOOL_HANDLERS = {
"hindsight_retain": ("content", _tool_retain),
"hindsight_recall": ("query", _tool_recall),
"hindsight_reflect": ("query", _tool_reflect),
"hindsight_retain": ("content", _tool_retain, "Failed to store memory"),
"hindsight_recall": ("query", _tool_recall, "Failed to search memory"),
"hindsight_reflect": ("query", _tool_reflect, "Failed to reflect"),
}
def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
required, handler = self._TOOL_HANDLERS.get(tool_name, ("", None))
if handler is None:
if tool_name not in self._TOOL_HANDLERS:
return tool_error(f"Unknown tool: {tool_name}")
required, handler, failure = self._TOOL_HANDLERS[tool_name]
if not args.get(required, ""):
return tool_error(f"Missing required parameter: {required}")
try:
return json.dumps({"result": handler(self, args)})
except Exception as e:
logger.warning("%s failed: %s", tool_name, e, exc_info=True)
return tool_error(f"{_TOOL_ERRORS[tool_name]}: {e}")
return tool_error(f"{failure}: {e}")
# -- session lifecycle -------------------------------------------------------
@@ -1235,14 +1122,11 @@ class HindsightMemoryProvider(MemoryProvider):
# 3. Rotate to the new session.
if parent_session_id:
self._parent_session_id = str(parent_session_id).strip()
self._session_id = new_id
self._document_id = _mint_document_id(new_id)
self._session_id, self._document_id = new_id, _mint_document_id(new_id)
self._session_turns = []
self._turn_counter = self._turn_index = self._last_retained_turn_count = 0
logger.debug(
"Hindsight on_session_switch: new_session=%s parent=%s reset=%s doc=%s",
self._session_id, self._parent_session_id, reset, self._document_id,
)
logger.debug("Hindsight on_session_switch: new_session=%s parent=%s reset=%s doc=%s",
self._session_id, self._parent_session_id, reset, self._document_id)
def _close_client(self) -> None:
if self._mode != "local_embedded":
@@ -1254,14 +1138,10 @@ class HindsightMemoryProvider(MemoryProvider):
inner_client = getattr(self._client, "_client", None)
if inner_client is not None and hasattr(inner_client, "aclose"):
_run_sync(inner_client.aclose())
try:
with contextlib.suppress(Exception):
self._client._client = None
except Exception:
pass
try:
with contextlib.suppress(RuntimeError):
self._client.close()
except RuntimeError:
pass
def shutdown(self) -> None:
logger.debug("Hindsight shutdown: stopping writer + waiting for background threads")
@@ -1269,8 +1149,7 @@ class HindsightMemoryProvider(MemoryProvider):
self._shutting_down.set()
# The writer finishes in-flight work then exits on the sentinel; the
# bounded join keeps shutdown predictable even if the daemon is wedged.
writer = self._writer_thread
if writer is not None and writer.is_alive():
if (writer := self._writer_thread) is not None and writer.is_alive():
self._retain_queue.put(_WRITER_SENTINEL)
writer.join(timeout=10.0)
if writer.is_alive():
@@ -1278,10 +1157,8 @@ class HindsightMemoryProvider(MemoryProvider):
self._retain_queue.qsize())
self._join_prefetch(5.0)
if self._client is not None:
try:
with contextlib.suppress(Exception):
self._close_client()
except Exception:
pass
self._client = None
# The module-global loop is intentionally NOT stopped: it's shared by every
# provider in the process (one per gateway chat session); stopping it would
plugins/memory/hindsight/config_schema.py
+11 -51
View File
@@ -1,12 +1,7 @@
"""Hindsight's declared config surface — rendered by the generic desktop panel."""
from plugins.memory.config_schema import (
KIND_SECRET,
KIND_SELECT,
KIND_TEXT,
ProviderConfigSchema,
ProviderField,
ProviderFieldOption,
KIND_SECRET, KIND_SELECT, KIND_TEXT, ProviderConfigSchema, ProviderField, ProviderFieldOption,
)
CONFIG_SCHEMA = ProviderConfigSchema(
@@ -14,62 +9,27 @@ CONFIG_SCHEMA = ProviderConfigSchema(
label="Hindsight",
fields=(
ProviderField(
key="mode",
label="Mode",
kind=KIND_SELECT,
default="cloud",
key="mode", label="Mode", kind=KIND_SELECT, default="cloud",
description="How Hermes connects to Hindsight.",
options=(
ProviderFieldOption(
"cloud",
"Cloud",
"Hindsight Cloud API (lightweight, just needs an API key)",
),
ProviderFieldOption(
"local_external",
"Local External",
"Connect to an existing Hindsight instance",
),
ProviderFieldOption("cloud", "Cloud", "Hindsight Cloud API (lightweight, just needs an API key)"),
ProviderFieldOption("local_external", "Local External", "Connect to an existing Hindsight instance"),
),
inline=True,
),
ProviderField(
key="api_key",
label="API key",
kind=KIND_SECRET,
env_key="HINDSIGHT_API_KEY",
key="api_key", label="API key", kind=KIND_SECRET, env_key="HINDSIGHT_API_KEY",
description="Used to authenticate with the Hindsight API.",
placeholder="Enter Hindsight API key",
inline=True,
placeholder="Enter Hindsight API key", inline=True,
),
ProviderField(
key="api_url",
label="API URL",
kind=KIND_TEXT,
default="https://api.hindsight.vectorize.io",
aliases=("apiUrl",),
env_fallbacks=("HINDSIGHT_API_URL",),
inline=True,
key="api_url", label="API URL", kind=KIND_TEXT, default="https://api.hindsight.vectorize.io",
aliases=("apiUrl",), env_fallbacks=("HINDSIGHT_API_URL",), inline=True,
),
ProviderField(key="bank_id", label="Bank ID", kind=KIND_TEXT, default="hermes", aliases=("bankId",), inline=True),
ProviderField(
key="bank_id",
label="Bank ID",
kind=KIND_TEXT,
default="hermes",
aliases=("bankId",),
inline=True,
),
ProviderField(
key="recall_budget",
label="Recall budget",
kind=KIND_SELECT,
default="mid",
aliases=("budget",),
options=(
ProviderFieldOption("low", "low"),
ProviderFieldOption("mid", "mid"),
ProviderFieldOption("high", "high"),
),
key="recall_budget", label="Recall budget", kind=KIND_SELECT, default="mid", aliases=("budget",),
options=tuple(ProviderFieldOption(b, b) for b in ("low", "mid", "high")),
inline=True,
),
),
plugins/memory/hindsight/embedded.py
+15 -20
View File
@@ -3,6 +3,7 @@ file the standalone ``hindsight-embed`` daemon consumes, and the health-grace ex
from __future__ import annotations
import contextlib
import importlib
import logging
import os
@@ -39,11 +40,9 @@ def _export_port_health_grace_timeout(config: dict[str, Any]) -> None:
try:
seconds = float(raw)
except (TypeError, ValueError):
seconds = None
if seconds is None or seconds < 0:
logger.warning("%s Hindsight port_health_grace_timeout %r; ignoring.",
"Invalid" if seconds is None else "Negative", raw)
return
return logger.warning("Invalid Hindsight port_health_grace_timeout %r; ignoring.", raw)
if seconds < 0:
return logger.warning("Negative Hindsight port_health_grace_timeout %r; ignoring.", raw)
os.environ.setdefault(_PORT_HEALTH_GRACE_ENV, repr(seconds))
@@ -91,10 +90,14 @@ def _embedded_profile_env_path(config: dict[str, Any]) -> Path:
return Path.home() / ".hindsight" / "profiles" / f"{profile}.env"
def _embedded_llm_api_key(config: dict[str, Any]) -> str:
return config.get("llmApiKey") or config.get("llm_api_key") or get_secret("HINDSIGHT_LLM_API_KEY", "")
def _build_embedded_profile_env(config: dict[str, Any], *, llm_api_key: str | None = None) -> dict[str, str]:
"""Build the profile-scoped env that standalone hindsight-embed consumes."""
if llm_api_key is None:
llm_api_key = config.get("llmApiKey") or config.get("llm_api_key") or get_secret("HINDSIGHT_LLM_API_KEY", "")
llm_api_key = _embedded_llm_api_key(config)
env_values = {
"HINDSIGHT_API_LLM_PROVIDER": str(_daemon_llm_provider(config.get("llm_provider", ""))),
"HINDSIGHT_API_LLM_API_KEY": str(llm_api_key or ""),
@@ -104,26 +107,19 @@ def _build_embedded_profile_env(config: dict[str, Any], *, llm_api_key: str | No
base_url = config.get("llm_base_url") or os.environ.get("HINDSIGHT_API_LLM_BASE_URL", "")
if base_url:
env_values["HINDSIGHT_API_LLM_BASE_URL"] = str(base_url)
idle_timeout = config.get("idle_timeout")
if idle_timeout is None:
if (idle_timeout := config.get("idle_timeout")) is None:
idle_timeout = os.environ.get("HINDSIGHT_IDLE_TIMEOUT")
if idle_timeout is not None and idle_timeout != "":
env_values["HINDSIGHT_EMBED_DAEMON_IDLE_TIMEOUT"] = str(_parse_int_setting(idle_timeout, _DEFAULT_IDLE_TIMEOUT))
return env_values
def _chmod_owner_only(path: Path) -> None:
try:
os.chmod(path, 0o600)
except OSError:
pass
def _secure_write_profile_env(profile_env: Path, content: str) -> None:
"""Create/overwrite *profile_env* owner-only (0600); a pre-existing file is
tightened BEFORE the plaintext LLM API key is written."""
if profile_env.exists():
_chmod_owner_only(profile_env)
with contextlib.suppress(OSError):
os.chmod(profile_env, 0o600)
fd = os.open(str(profile_env), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(content)
@@ -136,7 +132,8 @@ def _validate_profile_env_permissions(profile_env: Path) -> None:
import stat
if stat.S_IMODE(profile_env.stat().st_mode) != 0o600:
_chmod_owner_only(profile_env)
with contextlib.suppress(OSError):
os.chmod(profile_env, 0o600)
if stat.S_IMODE(profile_env.stat().st_mode) != 0o600:
raise PermissionError(
f"Embedded Hindsight profile environment is not owner-only: {profile_env}"
@@ -154,9 +151,7 @@ def _materialize_embedded_profile_env(config: dict[str, Any], *, llm_api_key: st
_secure_write_profile_env(profile_env, content)
_validate_profile_env_permissions(profile_env)
except BaseException:
try:
with contextlib.suppress(OSError):
profile_env.unlink()
except OSError:
pass
raise
return profile_env
plugins/memory/hindsight/settings.py
+17 -39
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import contextlib
import json
import logging
import re
@@ -60,21 +61,14 @@ def _normalize_retain_tags(value: Any) -> List[str]:
"""Normalize tag config/tool values to a deduplicated list of strings."""
if value is None:
return []
if isinstance(value, list):
raw_items = value
elif isinstance(value, str):
raw_items = value if isinstance(value, list) else [value]
if isinstance(value, str):
text = value.strip()
if not text:
return []
parsed = None
if text.startswith("["):
try:
with contextlib.suppress(Exception):
parsed = json.loads(text)
except Exception:
pass
raw_items = parsed if isinstance(parsed, list) else text.split(",")
else:
raw_items = [value]
normalized: list[str] = []
for item in raw_items:
tag = str(item).strip()
@@ -98,37 +92,23 @@ def _normalize_observation_scopes(value: Any) -> Any:
except Exception:
return None
return None
if isinstance(value, (list, tuple)):
if all(isinstance(entry, str) for entry in value):
inner = [entry.strip() for entry in value if entry.strip()]
return [inner] if inner else None
scopes: list[list[str]] = []
for entry in value:
if isinstance(entry, (list, tuple)):
inner = [str(tag).strip() for tag in entry if str(tag).strip()]
if inner:
scopes.append(inner)
elif isinstance(entry, str) and entry.strip():
scopes.append([entry.strip()])
return scopes or None
return None
if not isinstance(value, (list, tuple)):
return None
if all(isinstance(entry, str) for entry in value): # flat tag list -> one scope
value = [value]
scopes = [
[str(tag).strip() for tag in entry if str(tag).strip()] if isinstance(entry, (list, tuple))
else [entry.strip()] if isinstance(entry, str) and entry.strip() else []
for entry in value
]
return [s for s in scopes if s] or None
def _sanitize_bank_segment(value: str) -> str:
"""URL/filesystem-safe bank_id placeholder: runs outside ``[A-Za-z0-9_-]`` (per
``str.isalnum``) become one dash; leading/trailing ``-``/``_`` are stripped."""
if not value:
return ""
out = []
prev_dash = False
for ch in str(value):
if ch.isalnum() or ch in "-_":
out.append(ch)
prev_dash = False
elif not prev_dash:
out.append("-")
prev_dash = True
return "".join(out).strip("-_")
# \w == str.isalnum() + "_" for str patterns, so this matches the per-char rule.
return re.sub(r"[^\w-]+", "-", str(value)).strip("-_") if value else ""
def _resolve_bank_id_template(template: str, fallback: str, **placeholders: str) -> str:
@@ -143,6 +123,4 @@ def _resolve_bank_id_template(template: str, fallback: str, **placeholders: str)
logger.warning("Invalid bank_id_template %r: %s — using fallback %r",
template, exc, fallback)
return fallback
rendered = re.sub(r"-{2,}", "-", rendered)
rendered = re.sub(r"_{2,}", "_", rendered)
return rendered.strip("-_") or fallback
return re.sub(r"([-_])\1+", r"\1", rendered).strip("-_") or fallback
plugins/memory/hindsight/setup.py
+44 -64
View File
@@ -2,7 +2,9 @@
from __future__ import annotations
import contextlib
import json
import os
import sys
from pathlib import Path
@@ -10,17 +12,9 @@ from agent.secret_scope import get_secret
from hermes_cli.secret_prompt import masked_secret_prompt
from . import templates as _hs_templates
from .embedded import (
_embedded_profile_env_path,
_load_simple_env,
_materialize_embedded_profile_env,
)
from .embedded import _embedded_profile_env_path, _load_simple_env, _materialize_embedded_profile_env
from .settings import (
_DEFAULT_API_URL,
_DEFAULT_IDLE_TIMEOUT,
_DEFAULT_LOCAL_URL,
_DEFAULT_TIMEOUT,
_MIN_CLIENT_VERSION,
_DEFAULT_API_URL, _DEFAULT_IDLE_TIMEOUT, _DEFAULT_LOCAL_URL, _DEFAULT_TIMEOUT, _MIN_CLIENT_VERSION,
_PROVIDER_DEFAULT_MODELS,
)
@@ -60,29 +54,26 @@ def _write_env(env_path: Path, env_writes: dict) -> None:
new_lines = []
for line in existing:
key = line.split("=", 1)[0].strip() if "=" in line and not line.startswith("#") else None
if key in env_writes:
new_lines.append(f"{key}={env_writes[key]}")
updated.add(key)
else:
new_lines.append(line)
new_lines.append(f"{key}={env_writes[key]}" if key in env_writes else line)
updated.add(key)
new_lines.extend(f"{k}={v}" for k, v in env_writes.items() if k not in updated)
env_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
def _offer_starter_template(mode: str, provider_config: dict, env_writes: dict) -> None:
"""Seed the bank with a Hermes starter template (best-effort)."""
import os
from hermes_cli.memory_setup import _CANCELLED, _curses_select
default_url = _DEFAULT_LOCAL_URL if mode == "local_external" else _DEFAULT_API_URL
_hs_templates.run_template_step(
api_url=provider_config.get("api_url") or default_url,
bank_id=provider_config.get("bank_id", "hermes"),
api_key=env_writes.get("HINDSIGHT_API_KEY") or os.environ.get("HINDSIGHT_API_KEY", "") or None,
select=_curses_select,
cancelled=_CANCELLED,
)
def _prompt_embedded_llm(llm_provider: str, provider_config: dict, env_writes: dict, hermes_env: Path) -> None:
"""local_embedded wizard step: endpoint (openai_compatible only), model, LLM key."""
if llm_provider == "openai_compatible":
existing_base_url = provider_config.get("llm_base_url", "")
prompt = " LLM endpoint URL (e.g. http://192.168.1.10:8080/v1)" + (f" [{existing_base_url}]" if existing_base_url else "")
if val := input(prompt + ": ").strip():
provider_config["llm_base_url"] = val
elif llm_provider == "openrouter":
provider_config["llm_base_url"] = "https://openrouter.ai/api/v1"
current_model = provider_config.get("llm_model") or _PROVIDER_DEFAULT_MODELS.get(llm_provider, "gpt-4o-mini")
val = input(f" LLM model [{current_model}]: ").strip()
provider_config["llm_model"] = val or current_model
llm_key = _secret_prompt(" LLM API key: ")
env_writes["HINDSIGHT_LLM_API_KEY"] = llm_key or _load_simple_env(hermes_env).get("HINDSIGHT_LLM_API_KEY", "")
def run_setup(provider, hermes_home: str, config: dict) -> None:
@@ -94,19 +85,21 @@ def run_setup(provider, hermes_home: str, config: dict) -> None:
print("\n Configuring Hindsight memory:\n")
existing_config = provider._config if isinstance(provider._config, dict) else _load_config()
existing_config = existing_config if isinstance(existing_config, dict) else {}
if not isinstance(existing_config, dict):
existing_config = {}
mode = _select(" Select mode", _MODE_ITEMS, _MODE_VALUES, existing_config.get("mode"))
if mode is None:
return
provider_config: dict = dict(existing_config, mode=mode)
env_writes: dict = {}
hermes_env = Path(hermes_home) / ".env"
llm_provider = ""
if mode == "local_embedded":
providers = list(_PROVIDER_DEFAULT_MODELS)
llm_items = [(p, f"default model: {_PROVIDER_DEFAULT_MODELS[p]}") for p in providers]
llm_provider = _select(" Select LLM provider", llm_items, providers, provider_config.get("llm_provider"))
llm_items = [(p, f"default model: {m}") for p, m in _PROVIDER_DEFAULT_MODELS.items()]
llm_provider = _select(" Select LLM provider", llm_items, list(_PROVIDER_DEFAULT_MODELS),
provider_config.get("llm_provider"))
if llm_provider is None:
return
provider_config["llm_provider"] = llm_provider
@@ -132,35 +125,15 @@ def run_setup(provider, hermes_home: str, config: dict) -> None:
api_key = _secret_prompt(f" API key (current: {masked}, blank to keep): " if existing_key else " API key: ")
if api_key:
env_writes["HINDSIGHT_API_KEY"] = api_key
val = input(f" API URL [{_DEFAULT_API_URL}]: ").strip()
if val:
if val := input(f" API URL [{_DEFAULT_API_URL}]: ").strip():
provider_config["api_url"] = val
elif mode == "local_external":
val = input(f" Hindsight API URL [{_DEFAULT_LOCAL_URL}]: ").strip()
provider_config["api_url"] = val or _DEFAULT_LOCAL_URL
api_key = _secret_prompt(" API key (optional, blank to skip): ")
if api_key:
if api_key := _secret_prompt(" API key (optional, blank to skip): "):
env_writes["HINDSIGHT_API_KEY"] = api_key
else: # local_embedded
if llm_provider == "openai_compatible":
existing_base_url = provider_config.get("llm_base_url", "")
prompt = " LLM endpoint URL (e.g. http://192.168.1.10:8080/v1)" + (f" [{existing_base_url}]" if existing_base_url else "")
val = input(prompt + ": ").strip()
if val:
provider_config["llm_base_url"] = val
elif llm_provider == "openrouter":
provider_config["llm_base_url"] = "https://openrouter.ai/api/v1"
current_model = provider_config.get("llm_model") or _PROVIDER_DEFAULT_MODELS.get(llm_provider, "gpt-4o-mini")
val = input(f" LLM model [{current_model}]: ").strip()
provider_config["llm_model"] = val or current_model
llm_key = _secret_prompt(" LLM API key: ")
env_writes["HINDSIGHT_LLM_API_KEY"] = (
llm_key or _load_simple_env(Path(hermes_home) / ".env").get("HINDSIGHT_LLM_API_KEY", "")
)
else:
_prompt_embedded_llm(llm_provider, provider_config, env_writes, hermes_env)
provider_config.setdefault("bank_id", "hermes")
provider_config.setdefault("recall_budget", "mid")
@@ -176,23 +149,30 @@ def run_setup(provider, hermes_home: str, config: dict) -> None:
save_config(config)
provider.save_config(provider_config, hermes_home)
if env_writes:
_write_env(Path(hermes_home) / ".env", env_writes)
_write_env(hermes_env, env_writes)
# Starter template only where the API is reachable now (local_embedded's daemon isn't up).
# Starter template (best-effort) only where the API is reachable now
# (local_embedded's daemon isn't up).
if _hs_templates.supported_for_mode(mode):
_offer_starter_template(mode, provider_config, env_writes)
from hermes_cli.memory_setup import _CANCELLED, _curses_select
default_url = _DEFAULT_LOCAL_URL if mode == "local_external" else _DEFAULT_API_URL
_hs_templates.run_template_step(
api_url=provider_config.get("api_url") or default_url,
bank_id=provider_config.get("bank_id", "hermes"),
api_key=env_writes.get("HINDSIGHT_API_KEY") or os.environ.get("HINDSIGHT_API_KEY", "") or None,
select=_curses_select, cancelled=_CANCELLED,
)
if mode == "local_embedded":
materialized_config = dict(provider_config)
try:
with contextlib.suppress(Exception):
materialized_config = json.loads(
(Path(hermes_home) / "hindsight" / "config.json").read_text(encoding="utf-8")
)
except Exception:
pass
llm_api_key = (
env_writes.get("HINDSIGHT_LLM_API_KEY", "")
or _load_simple_env(Path(hermes_home) / ".env").get("HINDSIGHT_LLM_API_KEY", "")
or _load_simple_env(hermes_env).get("HINDSIGHT_LLM_API_KEY", "")
or _load_simple_env(_embedded_profile_env_path(materialized_config)).get("HINDSIGHT_API_LLM_API_KEY", "")
)
_materialize_embedded_profile_env(materialized_config, llm_api_key=llm_api_key or None)
plugins/memory/hindsight/templates.py
+5 -7
View File
@@ -49,16 +49,14 @@ def fetch_hermes_templates(url: str | None = None) -> list[dict]:
def fetch_manifest(entry: dict, url: str | None = None) -> dict:
"""Fetch the BankTemplateManifest JSON for a catalog entry."""
# manifest_file is relative to the catalog (e.g. "templates/foo.json").
manifest_url = urljoin(url or catalog_url(), entry["manifest_file"])
return _get_json(manifest_url)
"""Fetch the BankTemplateManifest JSON for a catalog entry (``manifest_file`` is
relative to the catalog, e.g. "templates/foo.json")."""
return _get_json(urljoin(url or catalog_url(), entry["manifest_file"]))
def _bank_request(api_url: str, bank_id: str, api_key: str | None, action: str, **kwargs) -> urllib.request.Request:
headers = dict(kwargs.pop("headers"))
def _bank_request(api_url: str, bank_id: str, api_key: str | None, action: str, *, headers: dict, **kwargs):
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
headers = {**headers, "Authorization": f"Bearer {api_key}"}
endpoint = f"{api_url.rstrip('/')}/v1/default/banks/{bank_id}/{action}"
return urllib.request.Request(endpoint, headers=headers, **kwargs) # noqa: S310