diff --git a/agent/memory_manager.py b/agent/memory_manager.py index 850a9bb410..d7f9caa632 100644 --- a/agent/memory_manager.py +++ b/agent/memory_manager.py @@ -16,7 +16,7 @@ from concurrent.futures import Future, ThreadPoolExecutor, wait from functools import partial from typing import Any, Callable, Dict, List, Optional -from agent.memory_provider import MemoryProvider, PRE_COMPRESS_CHECKPOINT_API_VERSION +from agent.memory_provider import MemoryProvider, PRE_COMPRESS_CHECKPOINT_API_VERSION, ctx_bound, spawn_context_thread from agent.skill_commands import extract_user_instruction_from_skill_message from tools.hook_output_spill import get_spill_config, spill_if_oversized from tools.registry import tool_error @@ -59,12 +59,6 @@ def _accepts_require_checkpoint(fn: Callable[..., Any]) -> bool: return _has_var_kwargs(params) or kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) -def _ctx_bound(fn: Callable[[], Any]) -> Callable[[], Any]: - """Bind ``fn`` to the CALLER's contextvars for another thread: profile isolation is a - ContextVar-scoped HERMES_HOME override, and an unbound worker would silently use the default profile.""" - return partial(contextvars.copy_context().run, fn) - - # -- Tool-schema plumbing ----------------------------------------------------- def normalize_tool_schema(schema: Any) -> Optional[Dict[str, Any]]: @@ -421,7 +415,7 @@ class MemoryManager: except Exception as exc: # pragma: no cover - re-raised by caller result_box["error"] = exc - thread = threading.Thread(target=_ctx_bound(_run), daemon=True, name=f"memory-prefetch-{provider.name}") + thread = spawn_context_thread(_run, name=f"memory-prefetch-{provider.name}") with self._external_prefetch_lock: existing = self._external_prefetch_threads.get(provider.name) if existing is not None and existing.is_alive(): @@ -511,9 +505,9 @@ class MemoryManager: def _submit_background(self, fn, *, kind: str = "write") -> None: """Queue ``fn`` on the serialized worker (created lazily; None once shutting down) and track its - durability class. Runs under the caller's contextvars (``_ctx_bound``). If the executor is + durability class. Runs under the caller's contextvars (``ctx_bound``). If the executor is unavailable outside shutdown, run inline — the historical fail-safe.""" - fn = _ctx_bound(fn) + fn = ctx_bound(fn) executor = None if self._shutting_down else self._sync_executor if executor is None and not self._shutting_down: with self._sync_executor_lock: diff --git a/agent/memory_provider.py b/agent/memory_provider.py index 9d698ada7a..b5a9d2edc5 100644 --- a/agent/memory_provider.py +++ b/agent/memory_provider.py @@ -7,14 +7,31 @@ prefetch / sync_turn per turn -> tool dispatch -> shutdown, plus optional ``on_* 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, Dict, List, Optional +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 = ()) -> 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, 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 diff --git a/plugins/memory/byterover/__init__.py b/plugins/memory/byterover/__init__.py index 6d81225bf5..42c2684b48 100644 --- a/plugins/memory/byterover/__init__.py +++ b/plugins/memory/byterover/__init__.py @@ -17,7 +17,7 @@ import threading from pathlib import Path from typing import Any, Dict, List, Optional -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, spawn_context_thread from tools.registry import tool_error logger = logging.getLogger(__name__) @@ -171,7 +171,7 @@ class ByteRoverMemoryProvider(MemoryProvider): except Exception as e: logger.debug("ByteRover %s failed: %s", what, e) - t = threading.Thread(target=_work, daemon=True, name=name) + t = spawn_context_thread(_work, name=name) t.start() return t diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py index 25e663e761..1cc6d708bc 100644 --- a/plugins/memory/hindsight/__init__.py +++ b/plugins/memory/hindsight/__init__.py @@ -13,7 +13,6 @@ from __future__ import annotations import asyncio import atexit import contextlib -import contextvars import json import logging import os @@ -25,12 +24,13 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional -from agent.memory_provider import MemoryProvider, RecallStatus +from agent.memory_provider import MemoryProvider, RecallStatus, spawn_context_thread from agent.secret_scope import get_secret from hermes_cli.config import cfg_get from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now from tools.registry import tool_error +from utils import read_json_or_empty from .embedded import ( _RETRIABLE_CONNECTION_MARKERS, _build_embedded_profile_env, @@ -185,14 +185,6 @@ def _run_sync(coro, timeout: float = _DEFAULT_TIMEOUT): return future.result(timeout=timeout) -def _context_thread(target, name: str) -> threading.Thread: - """Daemon thread running *target* in a snapshot of the spawner's contextvars. - 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) - - RETAIN_SCHEMA = { "name": "hindsight_retain", "description": ( @@ -244,8 +236,7 @@ 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(): - with contextlib.suppress(Exception): - return json.loads(path.read_text(encoding="utf-8")) + return read_json_or_empty(path) # Mode, bank (the data partition), endpoint and retain shaping are per-profile .env values like # the key beside them: read through the secret scope so a multiplexed secondary never inherits # the default profile's bank/mode. Tuning knobs (timeouts, budget) stay process-global. @@ -389,13 +380,7 @@ class HindsightMemoryProvider(MemoryProvider): """Merge *values* into $HERMES_HOME/hindsight/config.json.""" from utils import atomic_json_write config_path = Path(hermes_home) / "hindsight" / "config.json" - config_path.parent.mkdir(parents=True, exist_ok=True) - 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) + atomic_json_write(config_path, {**read_json_or_empty(config_path), **values}, mode=0o600) def post_setup(self, hermes_home: str, config: dict) -> None: """Custom setup wizard — installs only the deps needed for the selected mode.""" @@ -519,7 +504,7 @@ class HindsightMemoryProvider(MemoryProvider): return # A previous writer may have exited after shutdown(); allow the fresh one to drain. self._shutting_down.clear() - thread = _context_thread(self._writer_loop, "hindsight-writer") + thread = spawn_context_thread(self._writer_loop, name="hindsight-writer") self._writer_thread = self._sync_thread = thread thread.start() @@ -805,7 +790,7 @@ class HindsightMemoryProvider(MemoryProvider): print(f" ⚠ {msg}", file=sys.stderr, flush=True) self._mode = "disabled" return - _context_thread(self._daemon_start_worker, "hindsight-daemon-start").start() + spawn_context_thread(self._daemon_start_worker, name="hindsight-daemon-start").start() def _daemon_start_worker(self) -> None: import traceback @@ -951,7 +936,7 @@ class HindsightMemoryProvider(MemoryProvider): with self._prefetch_lock: self._prefetch_result, self._prefetch_count = text, count - self._prefetch_thread = _context_thread(_run, "hindsight-prefetch") + self._prefetch_thread = spawn_context_thread(_run, name="hindsight-prefetch") self._prefetch_thread.start() # -- retain ------------------------------------------------------------------ diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index 8ff1d0ce9a..aede5a1323 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -113,16 +113,10 @@ class HolographicMemoryProvider(MemoryProvider): def save_config(self, values, hermes_home): """Write config to config.yaml under plugins.hermes-memory-store.""" - config_path = Path(hermes_home) / "config.yaml" - try: - import yaml - from hermes_cli.config import read_user_config_raw # raw read: merged defaults must not be persisted - existing = read_user_config_raw(config_path) - existing.setdefault("plugins", {})["hermes-memory-store"] = values - with open(config_path, "w", encoding="utf-8") as f: - yaml.dump(existing, f, default_flow_style=False) - except Exception: - pass + # The canonical writer: config lock, managed-mode refusal, default stripping, atomic replace. + # ``merge_existing`` keeps every other section; *hermes_home* is the active profile already. + from hermes_cli.config import save_config + save_config({"plugins": {"hermes-memory-store": dict(values)}}, merge_existing=True) def get_config_schema(self): from hermes_constants import display_hermes_home diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index 834cc0e48e..261018b7c3 100644 --- a/plugins/memory/honcho/__init__.py +++ b/plugins/memory/honcho/__init__.py @@ -17,9 +17,9 @@ import time from typing import Any, Callable, Dict, List, Optional from agent.memory_manager import sanitize_context -from agent.memory_provider import MemoryProvider, is_trivial_prompt +from agent.memory_provider import MemoryProvider, is_trivial_prompt, spawn_context_thread from agent.turn_author import a2a_key -from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path, spawn_context_thread +from plugins.memory.honcho.client import HonchoClientConfig, resolve_config_path from plugins.memory.honcho.dialectic import DialecticMixin from plugins.memory.honcho.session_peers import assistant_peer_id_for, sanitize_peer_id from plugins.memory.honcho.tool_schemas import ALL_TOOL_SCHEMAS @@ -168,14 +168,9 @@ class HonchoMemoryProvider(DialecticMixin, MemoryProvider): def save_config(self, values, hermes_home): """Merge ``values`` into $HERMES_HOME/honcho.json (Honcho SDK native format).""" from pathlib import Path - from utils import atomic_json_write - from plugins.memory.honcho.client import _read_config + from utils import atomic_json_write, read_json_or_empty config_path = Path(hermes_home) / "honcho.json" - try: - existing = _read_config(config_path) - except Exception: - existing = {} - atomic_json_write(config_path, {**existing, **values}, mode=0o600) + atomic_json_write(config_path, {**read_json_or_empty(config_path), **values}, mode=0o600) def get_config_schema(self): return [ diff --git a/plugins/memory/honcho/cli.py b/plugins/memory/honcho/cli.py index eda74badd7..722390f83e 100644 --- a/plugins/memory/honcho/cli.py +++ b/plugins/memory/honcho/cli.py @@ -10,6 +10,7 @@ from pathlib import Path from hermes_constants import get_hermes_home from plugins.memory.honcho.client import _first_parsed, _host_block, profile_host_key, resolve_active_host, resolve_config_path, HOST from hermes_cli.config import cfg_get +from utils import read_json_or_empty RULE = "─" * 40 REASONING_LEVELS = ("minimal", "low", "medium", "high", "max") @@ -66,10 +67,7 @@ def _local_config_path() -> Path: def _read_config() -> dict: - try: - return json.loads(_config_path().read_text(encoding="utf-8")) - except Exception: - return {} + return read_json_or_empty(_config_path()) def _write_config(cfg: dict, path: Path | None = None) -> None: diff --git a/plugins/memory/honcho/client.py b/plugins/memory/honcho/client.py index 035173688a..2b18e870aa 100644 --- a/plugins/memory/honcho/client.py +++ b/plugins/memory/honcho/client.py @@ -17,7 +17,6 @@ import os # replacing the single process-wide slot that pinned the first profile's workspace and bearer for every # later profile in multi-profile processes (#69123 multiplexed gateway, #74065 dashboard). The legacy names # above are retained only for reset bookkeeping. -import threading as _threading from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, Callable @@ -507,16 +506,6 @@ class HonchoClientConfig: return self.workspace_id -def spawn_context_thread(target, *, name: str, daemon: bool = True, args: tuple = ()) -> "_threading.Thread": - """Thread that inherits the caller's contextvars: profile isolation is a ContextVar - (set_hermes_home_override) and a plain Thread starts EMPTY, so ambient resolution on it - would silently land on the default profile.""" - import contextvars - - ctx = contextvars.copy_context() - return _threading.Thread(target=lambda: ctx.run(target, *args), name=name, daemon=daemon) - - def get_honcho_client(config: HonchoClientConfig | None = None) -> Honcho: """Get or create the Honcho client for this config's identity. Clients are cached PER IDENTITY (host, workspace, provenance paths, credential fingerprint, timeout) so diff --git a/plugins/memory/honcho/dialectic.py b/plugins/memory/honcho/dialectic.py index eaaca19fd1..1a49353339 100644 --- a/plugins/memory/honcho/dialectic.py +++ b/plugins/memory/honcho/dialectic.py @@ -7,7 +7,7 @@ import re import threading import time -from plugins.memory.honcho.client import spawn_context_thread +from agent.memory_provider import spawn_context_thread logger = logging.getLogger("plugins.memory.honcho") diff --git a/plugins/memory/honcho/oauth.py b/plugins/memory/honcho/oauth.py index 7de658ca34..b78227a340 100644 --- a/plugins/memory/honcho/oauth.py +++ b/plugins/memory/honcho/oauth.py @@ -21,6 +21,7 @@ from pathlib import Path from typing import Any from agent.redact import redact_sensitive_text, register_redaction_patterns +from utils import read_json_or_empty logger = logging.getLogger(__name__) @@ -106,15 +107,9 @@ def _mark_grant_dead(key: tuple[str, str], cred: OAuthCredential) -> None: _dead_grants[key] = hashlib.sha256(cred.refresh_token.encode("utf-8")).hexdigest() _reauth_check_cache.pop(key, None) # verdict changed without a config rewrite -def _read_config(path: Path) -> dict[str, Any]: - try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - def _load_cred(path: Path, host: str, raw: dict[str, Any] | None = None) -> OAuthCredential | None: """Credential from ``host``'s block in ``raw`` (or the file at ``path``).""" - source = raw if raw is not None else _read_config(path) + source = raw if raw is not None else read_json_or_empty(path) return OAuthCredential.from_host_block((source.get("hosts") or {}).get(host) or {}) def reauth_required(path: Path, host: str) -> bool: @@ -284,7 +279,7 @@ def _persist_credential(path: Path, host: str, cred: OAuthCredential, raw: dict[ the file's current content), leaving the rest intact; marks the grant live.""" from utils import atomic_json_write - raw = _read_config(path) if raw is None else raw + raw = read_json_or_empty(path) if raw is None else raw block = raw.setdefault("hosts", {}).setdefault(host, {}) block["apiKey"], block["oauth"] = cred.access_token, cred.oauth_block() atomic_json_write(path, raw, mode=0o600) @@ -360,7 +355,7 @@ def install_grant( ``apiKey`` and ``oauth`` block. ``apply_config=False`` stores tokens only.""" now = time.time() if now is None else now cred = OAuthCredential.from_token_response(grant, now=now, client_id=client_id, token_endpoint=token_endpoint) - raw = _read_config(path) + raw = read_json_or_empty(path) granted_config = grant.get("config") if isinstance(granted_config, dict): cred.consent_peer_name = granted_config.get("peerName") diff --git a/plugins/memory/honcho/recall_sync.py b/plugins/memory/honcho/recall_sync.py index 9c25d1a1e6..6c93f51b8d 100644 --- a/plugins/memory/honcho/recall_sync.py +++ b/plugins/memory/honcho/recall_sync.py @@ -7,7 +7,7 @@ import math import threading import time -from plugins.memory.honcho.client import spawn_context_thread +from agent.memory_provider import spawn_context_thread logger = logging.getLogger("plugins.memory.honcho") diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index 3c04e19b45..2c8b6e6280 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -10,7 +10,8 @@ from dataclasses import dataclass, field from datetime import datetime from typing import Any, TYPE_CHECKING -from plugins.memory.honcho.client import get_honcho_client, spawn_context_thread +from agent.memory_provider import spawn_context_thread +from plugins.memory.honcho.client import get_honcho_client from plugins.memory.honcho.session_auth import HonchoAuthError, SessionAuthMixin from plugins.memory.honcho.session_context import SessionContextMixin from plugins.memory.honcho.session_migration import SessionMigrationMixin diff --git a/plugins/memory/mem0/__init__.py b/plugins/memory/mem0/__init__.py index d873167e0d..f5ec6770f5 100644 --- a/plugins/memory/mem0/__init__.py +++ b/plugins/memory/mem0/__init__.py @@ -18,9 +18,10 @@ from contextlib import suppress from pathlib import Path from typing import Any, Dict, List -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, spawn_context_thread from agent.secret_scope import UnscopedSecretError, get_secret from tools.registry import tool_error +from utils import atomic_json_write, read_json_or_empty logger = logging.getLogger(__name__) @@ -72,14 +73,6 @@ def _is_client_error(exc: Exception) -> bool: return type(exc).__name__ in _CLIENT_ERROR_TYPES or any(s in err_str for s in ("404", "not found", "valid uuid")) -def _read_mem0_json(config_path: Path) -> dict: - """Best-effort read of mem0.json; missing/corrupt file -> {}.""" - if config_path.exists(): - with suppress(Exception): - return json.loads(config_path.read_text(encoding="utf-8")) - return {} - - def _scoped_env(name: str) -> str: """Profile-scoped read of a non-secret mem0 setting; no scope under multiplex = unset (never ``os.environ``). Only the API key may fail closed — OSS mode has none to read (#99121).""" @@ -100,7 +93,7 @@ def _load_config() -> dict: "agent_id": _scoped_env("MEM0_AGENT_ID") or "hermes", "oss": {}} if user_id := _scoped_env("MEM0_USER_ID"): # only when explicitly configured, so initialize() can fall back to the gateway-native id config["user_id"] = user_id - file_cfg = _read_mem0_json(get_hermes_home() / "mem0.json") + file_cfg = read_json_or_empty(get_hermes_home() / "mem0.json") config.update({k: v for k, v in file_cfg.items() if v is not None and v != ""}) # MEM0_API_KEY authenticates the Platform and self-hosted HTTP backends; pure OSS mode builds its # backend from the local ``oss`` config and has no platform credential to resolve. Decide after @@ -161,9 +154,8 @@ class Mem0MemoryProvider(MemoryProvider): def save_config(self, values, hermes_home): """Merge-write config to $HERMES_HOME/mem0.json.""" - from utils import atomic_json_write config_path = Path(hermes_home) / "mem0.json" - atomic_json_write(config_path, {**_read_mem0_json(config_path), **values}, mode=0o600) + atomic_json_write(config_path, {**read_json_or_empty(config_path), **values}, mode=0o600) def get_config_schema(self): api_key_required = _load_config().get("mode", "platform") != "oss" @@ -299,7 +291,7 @@ class Mem0MemoryProvider(MemoryProvider): if self._prefetch_query == query and (self._prefetch_done or (self._prefetch_thread and self._prefetch_thread.is_alive())): return self._prefetch_query, self._prefetch_result, self._prefetch_done = query, "", False - self._prefetch_thread = t = threading.Thread(target=_run, daemon=True, name="mem0-prefetch") + self._prefetch_thread = t = spawn_context_thread(_run, name="mem0-prefetch") t.start() def prefetch(self, query: str, *, session_id: str = "") -> str: @@ -332,7 +324,7 @@ class Mem0MemoryProvider(MemoryProvider): prev.join(timeout=5.0) if prev.is_alive(): # still busy after the wait: skip to avoid duplicate ingestion return - self._sync_thread = threading.Thread(target=_sync, daemon=True, name="mem0-sync") + self._sync_thread = spawn_context_thread(_sync, name="mem0-sync") self._sync_thread.start() def get_tool_schemas(self) -> List[Dict[str, Any]]: diff --git a/plugins/memory/mem0/_setup.py b/plugins/memory/mem0/_setup.py index 60bcbfc3ce..9db123667c 100644 --- a/plugins/memory/mem0/_setup.py +++ b/plugins/memory/mem0/_setup.py @@ -169,8 +169,8 @@ def _persist_provider_config(hermes_home: str, config: dict, provider_config: di def _setup_platform(hermes_home: str, config: dict, flags: dict[str, str]) -> None: """Platform mode setup — prompts for API key (secret -> .env), user/agent ids and rerank (-> mem0.json).""" - from . import _read_mem0_json - provider_config = _read_mem0_json(Path(hermes_home) / "mem0.json") + from utils import read_json_or_empty + provider_config = read_json_or_empty(Path(hermes_home) / "mem0.json") print("\n Configuring mem0:\n") env_writes = _api_key_writes(flags, "Mem0 Platform API key", url="https://app.mem0.ai") for key, desc, default in (("user_id", "User identifier", "hermes-user"), ("agent_id", "Agent identifier", "hermes")): @@ -205,8 +205,8 @@ def _check_selfhosted_server(host: str) -> None: def _setup_selfhosted(hermes_home: str, config: dict, flags: dict[str, str]) -> None: """Self-hosted mode — point at an existing Mem0 server: URL -> mem0.json, key -> .env (MEM0_API_KEY).""" - from . import _read_mem0_json - provider_config = _read_mem0_json(Path(hermes_home) / "mem0.json") + from utils import read_json_or_empty + provider_config = read_json_or_empty(Path(hermes_home) / "mem0.json") print("\n Configuring mem0 (self-hosted server):\n") host = flags.get("host") or _prompt("Mem0 server URL (e.g. http://localhost:8888)", default=provider_config.get("host") or None) if not host: @@ -238,11 +238,11 @@ def _print_oss_summary(oss_config: dict, env_writes: dict, dry_run: bool = False def _finish_oss(hermes_home: str, config: dict, oss_config: dict, env_writes: dict[str, str], user_id: str, agent_id: str, pgvector_config: dict | None = None) -> None: """Shared OSS tail: write secrets + mem0.json, install deps, activate, check, summarize.""" - from . import _read_mem0_json + from utils import read_json_or_empty if env_writes: _write_env(Path(hermes_home) / ".env", env_writes) config_path = Path(hermes_home) / "mem0.json" # merge-write, plain text (platform path uses save_config's 0600 atomic write) - config_path.write_text(json.dumps({**_read_mem0_json(config_path), "mode": "oss", "user_id": user_id, "agent_id": agent_id, "oss": oss_config}, indent=2) + "\n", encoding="utf-8") + config_path.write_text(json.dumps({**read_json_or_empty(config_path), "mode": "oss", "user_id": user_id, "agent_id": agent_id, "oss": oss_config}, indent=2) + "\n", encoding="utf-8") _install_provider_deps(oss_config["llm"]["provider"], oss_config["embedder"]["provider"], oss_config["vector_store"]["provider"]) if pgvector_config: _ensure_pgvector_extension(pgvector_config) diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py index f29341be05..97a3c255d6 100644 --- a/plugins/memory/openviking/__init__.py +++ b/plugins/memory/openviking/__init__.py @@ -35,10 +35,11 @@ from urllib.parse import quote, unquote, urlparse from urllib.request import url2pathname from agent.message_content import flatten_message_text -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, spawn_context_thread from agent.secret_scope import get_secret from agent.skill_commands import extract_user_instruction_from_skill_message from hermes_cli import __version__ as _HERMES_VERSION +from hermes_constants import get_hermes_home from tools.registry import tool_error from utils import atomic_json_write, env_var_enabled @@ -904,15 +905,6 @@ def _local_openviking_bind(endpoint: str) -> tuple[str, int]: return parsed.hostname or "127.0.0.1", parsed.port or 1933 -def _hermes_home_path() -> Path: - try: - from hermes_constants import get_hermes_home - return get_hermes_home() - except Exception: - env_home = os.environ.get("HERMES_HOME") - return Path(env_home).expanduser() if env_home else Path.home() / ".hermes" - - def _local_openviking_port_is_open(host: str, port: int) -> bool: """Pre-spawn guard: a successful connect proves a listener owns the port (so a second openviking-server would lose the data-dir lock); says nothing about health.""" @@ -976,7 +968,7 @@ def _start_local_openviking_server(endpoint: str) -> tuple[str, str]: server_cmd = shutil.which("openviking-server") if not server_cmd: return _LOCAL_SERVER_FAILED, "openviking-server was not found on PATH. Start it manually, then retry." - log_path = _hermes_home_path() / _OPENVIKING_SERVER_LOG_RELATIVE_PATH + log_path = get_hermes_home() / _OPENVIKING_SERVER_LOG_RELATIVE_PATH try: log_path.parent.mkdir(parents=True, exist_ok=True) # Strip PYTHONPATH: the Desktop backend puts the Hermes venv on it, which @@ -1316,9 +1308,9 @@ class OpenVikingMemoryProvider(MemoryProvider): # Caller holds _runtime_start_lock and reserved ownership via _runtime_start_pending. if self._runtime_start_thread and self._runtime_start_thread.is_alive(): return - self._runtime_start_thread = threading.Thread( - target=self._finish_runtime_openviking_start, daemon=True, name="openviking-runtime-start", - kwargs={"endpoint": endpoint, "status_callback": status_callback, "warning_callback": warning_callback}) + self._runtime_start_thread = spawn_context_thread( + lambda: self._finish_runtime_openviking_start(endpoint=endpoint, status_callback=status_callback, warning_callback=warning_callback), + name="openviking-runtime-start") self._runtime_start_thread.start() def _settings_tuple(self, endpoint: Optional[str] = None) -> tuple: @@ -1411,7 +1403,7 @@ class OpenVikingMemoryProvider(MemoryProvider): self._env_refresh_enabled = True self._session_id = session_id self._turn_count = 0 - self._hermes_home = str(kwargs.get("hermes_home") or "").strip() or str(_hermes_home_path()) + self._hermes_home = str(kwargs.get("hermes_home") or "").strip() or str(get_hermes_home()) self._acquire_run_lock() self._profile_prefetched_sessions.clear() @@ -2078,7 +2070,7 @@ class OpenVikingMemoryProvider(MemoryProvider): if after_discard is not None: after_discard() - thread = threading.Thread(target=_run, daemon=True, name=name) + thread = spawn_context_thread(_run, name=name) with lock: if skip_if is not None and skip_if(): return diff --git a/plugins/memory/retaindb/__init__.py b/plugins/memory/retaindb/__init__.py index 1e421b3659..b5b236107c 100644 --- a/plugins/memory/retaindb/__init__.py +++ b/plugins/memory/retaindb/__init__.py @@ -22,7 +22,7 @@ from pathlib import Path from typing import Any, Callable from urllib.parse import quote -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, spawn_context_thread from agent.secret_scope import get_secret from agent.file_safety import raise_if_read_blocked from tools.registry import tool_error @@ -186,7 +186,7 @@ class _WriteQueue: def __init__(self, client: _Client, db_path: Path): self._client, self._db_path, self._q = client, db_path, queue.Queue() - self._thread = threading.Thread(target=self._loop, name="retaindb-writer", daemon=True) + self._thread = spawn_context_thread(self._loop, name="retaindb-writer") db_path.parent.mkdir(parents=True, exist_ok=True) self._local = threading.local() # one cached connection per thread, all tracked in _connections self._connections: set[sqlite3.Connection] = set() @@ -346,7 +346,7 @@ class RetainDBMemoryProvider(MemoryProvider): soul = (home / "SOUL.md").read_text(encoding="utf-8", errors="replace").strip() if (home / "SOUL.md").exists() else "" if soul: # seed agent identity from SOUL.md in background seed = lambda: self._client.seed_agent_identity(self._agent_id, soul, source="soul_md") # noqa: E731 - threading.Thread(target=_quiet, args=("soul seed", seed), name="retaindb-soul-seed", daemon=True).start() + spawn_context_thread(_quiet, args=("soul seed", seed), name="retaindb-soul-seed").start() def system_prompt_block(self) -> str: project = self._client.project if self._client else "retaindb" @@ -368,7 +368,7 @@ class RetainDBMemoryProvider(MemoryProvider): self._client.ask_user(self._user_id, query, reasoning_level=self._reasoning_level(query)).get("answer") or "") or None), ("retaindb-agent-model", "agent model", "_agent_model", lambda: self._agent_model_or_none(self._client.get_agent_model(self._agent_id))), ) - self._prefetch_threads = [threading.Thread(target=self._store, args=(label, attr, fetch), name=name, daemon=True) + self._prefetch_threads = [spawn_context_thread(self._store, args=(label, attr, fetch), name=name) for name, label, attr, fetch in jobs] for t in self._prefetch_threads: t.start() diff --git a/plugins/memory/supermemory/__init__.py b/plugins/memory/supermemory/__init__.py index 49e8362e92..8b4ac247cc 100644 --- a/plugins/memory/supermemory/__init__.py +++ b/plugins/memory/supermemory/__init__.py @@ -13,7 +13,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Dict, List, Optional -from agent.memory_provider import MemoryProvider +from agent.memory_provider import MemoryProvider, spawn_context_thread from agent.secret_scope import get_secret, is_multiplex_active from tools.registry import tool_error @@ -449,8 +449,8 @@ class SupermemoryMemoryProvider(MemoryProvider): return if self._write_thread and self._write_thread.is_alive(): self._write_thread.join(timeout=2.0) - self._write_thread = threading.Thread( - target=_quietly, daemon=False, name="supermemory-memory-write", + self._write_thread = spawn_context_thread( + _quietly, daemon=False, name="supermemory-memory-write", args=(lambda: self._client.add_memory(content.strip(), metadata={"target": target, "type": "explicit_memory"}, entity_context=self._entity_context), "Supermemory on_memory_write failed")) self._write_thread.start() diff --git a/tests/honcho_plugin/test_client_identity_isolation.py b/tests/honcho_plugin/test_client_identity_isolation.py index ae278c2105..cc0cf07df7 100644 --- a/tests/honcho_plugin/test_client_identity_isolation.py +++ b/tests/honcho_plugin/test_client_identity_isolation.py @@ -155,7 +155,7 @@ class TestBackgroundThreadIsolation: def test_spawn_context_thread_sees_profile_override(self, tmp_path): """spawn_context_thread must carry the caller's HERMES_HOME override.""" from hermes_constants import get_hermes_home - from plugins.memory.honcho.client import spawn_context_thread + from agent.memory_provider import spawn_context_thread home_b = tmp_path / "profiles" / "b" home_b.mkdir(parents=True) diff --git a/tests/plugins/memory/test_holographic_save_config.py b/tests/plugins/memory/test_holographic_save_config.py new file mode 100644 index 0000000000..cf6e473551 --- /dev/null +++ b/tests/plugins/memory/test_holographic_save_config.py @@ -0,0 +1,39 @@ +"""``save_config`` of the holographic provider writes config.yaml through the canonical writer. + +The provider used to ``yaml.dump`` straight over config.yaml, bypassing the config lock, the +managed-mode refusal and the atomic replace. Two contracts pin the canonical path: unrelated +sections survive a provider save, and a managed install refuses the write. +""" +from __future__ import annotations + +import yaml + +from plugins.memory.holographic import HolographicMemoryProvider + + +def _provider(): + return HolographicMemoryProvider(config={}) + + +def test_save_config_merges_into_existing_yaml(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text("model:\n default: keep-me\nmemory:\n provider: holographic\n") + + _provider().save_config({"db_path": "custom.db", "hrr_dim": "512"}, str(tmp_path)) + + raw = yaml.safe_load((tmp_path / "config.yaml").read_text()) + assert raw["plugins"]["hermes-memory-store"] == {"db_path": "custom.db", "hrr_dim": "512"} + assert raw["model"]["default"] == "keep-me" + assert raw["memory"]["provider"] == "holographic" + + +def test_save_config_respects_managed_mode(tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + before = "model:\n default: managed\n" + (tmp_path / "config.yaml").write_text(before) + monkeypatch.setattr("hermes_cli.config.is_managed", lambda: True) + monkeypatch.setattr("hermes_cli.config.managed_error", lambda *_a, **_k: None) + + _provider().save_config({"db_path": "custom.db"}, str(tmp_path)) + + assert (tmp_path / "config.yaml").read_text() == before diff --git a/tests/plugins/memory/test_provider_threads_inherit_profile.py b/tests/plugins/memory/test_provider_threads_inherit_profile.py new file mode 100644 index 0000000000..5f2cfda9aa --- /dev/null +++ b/tests/plugins/memory/test_provider_threads_inherit_profile.py @@ -0,0 +1,111 @@ +"""Multiplex invariant: every memory provider's background thread runs under the spawner's profile. + +Profile isolation is a ContextVar-scoped HERMES_HOME override; a plain ``threading.Thread`` starts +with an EMPTY context, so a provider's prefetch/sync/writer thread would silently resolve the DEFAULT +profile's home (and fail closed on scoped secrets). Each case drives the provider's real spawn path +with a fake backend and asserts the thread saw the parent's home. +""" +from __future__ import annotations + +import threading +from unittest.mock import MagicMock + +import pytest + +from hermes_constants import get_hermes_home, reset_hermes_home_override, set_hermes_home_override + + +def _probe_home(seen: dict, key: str = "home"): + def _record(*_args, **_kwargs): + seen[key] = get_hermes_home() + return _record + + +def _mem0(seen, tmp_path): + from plugins.memory.mem0 import Mem0MemoryProvider + + p = Mem0MemoryProvider() + p._backend = MagicMock() + p._config = {"mode": "platform"} + p._add = _probe_home(seen) + p.sync_turn("a long enough user message", "assistant reply") + return [p._sync_thread] + + +def _retaindb(seen, tmp_path): + import plugins.memory.retaindb as retaindb + + p = retaindb.RetainDBMemoryProvider() + p._client = MagicMock() + p._context_overlay = lambda query: {"context": seen.setdefault("home", get_hermes_home()) and "ctx"} + p._client.ask_user.return_value = {"answer": ""} + p._client.get_agent_model.return_value = {} + p.queue_prefetch("what do you know") + return list(p._prefetch_threads) + + +def _byterover(seen, tmp_path): + import plugins.memory.byterover as byterover + + p = byterover.ByteRoverMemoryProvider() + p._curate = _probe_home(seen) + return [p._curate_in_background("content", name="brv-test", what="test")] + + +def _supermemory(seen, tmp_path): + import plugins.memory.supermemory as supermemory + + p = supermemory.SupermemoryMemoryProvider() + p._active = p._write_enabled = True + p._client = MagicMock() + p._client.add_memory = _probe_home(seen) + p.on_memory_write("add", "user", "a fact") + return [p._write_thread] + + +def _openviking(seen, tmp_path): + import plugins.memory.openviking as openviking + + p = openviking.OpenVikingMemoryProvider() + workers: set = set() + p._spawn_tracked("ov-test", _probe_home(seen), threading.Lock(), lambda: workers) + return list(workers) + + +def _hindsight(seen, tmp_path): + import plugins.memory.hindsight as hindsight + + p = hindsight.HindsightMemoryProvider() + p._recall_sync, p._memory_mode, p._auto_recall, p._prefetch_waits_for_retain = False, "hybrid", True, False + p._do_recall = lambda query: (seen.setdefault("home", get_hermes_home()) and "text", 1) + p.queue_prefetch("remember this") + return [p._prefetch_thread] + + +def _honcho(seen, tmp_path): + from plugins.memory.honcho import HonchoMemoryProvider + + return [HonchoMemoryProvider._spawn_write(_probe_home(seen), "honcho-test", "failed %s")] + + +_PROVIDERS = { + "mem0": _mem0, "retaindb": _retaindb, "byterover": _byterover, "supermemory": _supermemory, + "openviking": _openviking, "hindsight": _hindsight, "honcho": _honcho, +} + + +@pytest.mark.parametrize("name", sorted(_PROVIDERS)) +def test_provider_background_thread_sees_spawner_profile_home(name, tmp_path, monkeypatch): + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "default")) + profile_home = tmp_path / "profiles" / "b" + profile_home.mkdir(parents=True) + seen: dict = {} + token = set_hermes_home_override(profile_home) + try: + threads = _PROVIDERS[name](seen, tmp_path) + finally: + reset_hermes_home_override(token) + for t in threads: + if t is not None: + t.join(timeout=10) + assert seen.get("home") == profile_home, f"{name}: background thread resolved {seen.get('home')}" diff --git a/utils.py b/utils.py index dabf197d07..d6d46489b0 100644 --- a/utils.py +++ b/utils.py @@ -313,6 +313,18 @@ def atomic_json_write( fsync_dir=fsync_dir) +def read_json_or_empty(path: Union[str, Path]) -> dict: + """The JSON object at *path*, or ``{}`` when the file is missing, unreadable, malformed or + not an object. The read half of every ``read → merge → atomic_json_write`` config store + (memory-provider ``save_config``), so a corrupt sidecar degrades to defaults instead of + taking the provider down.""" + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + def warn_if_credential_file_broadly_readable(path: Union[str, Path], *, label: str = "", log: logging.Logger | None = None) -> bool: """Warn when a credential file is group/world-readable; True when a warning was emitted.