refactor(memory-plugins): background threads inherit the profile context; JSON sidecar reads and the holographic config.yaml write use core primitives

Six of eight memory providers spawned plain threading.Thread for prefetch/sync/
writer work. A plain thread starts with an EMPTY contextvars.Context, so under
multiplex profiles the worker resolved the DEFAULT profile's HERMES_HOME (and
fails closed on scoped secrets). honcho and hindsight had each noticed and
written their own copy_context() wrapper; core had a third in memory_manager.
One canonical pair now lives on the ABC module every provider already imports:
agent/memory_provider.py::ctx_bound / spawn_context_thread. memory_manager,
honcho, hindsight, mem0, retaindb, byterover, supermemory and openviking all use
it; the honcho and hindsight wrappers and memory_manager._ctx_bound are deleted.

Five "json.loads(path.read_text()) or {}" readers (mem0._read_mem0_json,
honcho client/oauth/cli _read_config, hindsight save_config/_load_config) fold
into utils.read_json_or_empty, the read half of every read-merge-atomic_json_write
sidecar store.

holographic.save_config was the only config.yaml writer in the tree that
bypassed hermes_cli.config.save_config: raw open("w") + yaml.dump with no config
lock, no managed-mode refusal, no atomic replace, and a swallowed exception. It
now calls save_config(..., merge_existing=True). Behavior change: a managed
install refuses the write (previously silently rewrote config.yaml); other
sections are deep-merged instead of round-tripped through a raw dump.

openviking._hermes_home_path guarded an impossible ImportError of
hermes_constants (the module already imports agent.*) with a ~/.hermes fallback
that is wrong on Windows and under profile overrides; it is replaced by
get_hermes_home() directly.

Tests: tests/plugins/memory/test_provider_threads_inherit_profile.py drives each
provider's real spawn path with a fake backend and asserts the thread sees the
spawner's HERMES_HOME override (sabotage: retaindb back on threading.Thread ->
red). tests/plugins/memory/test_holographic_save_config.py pins merge-with-
existing-sections and managed-mode refusal (sabotage: raw yaml.dump -> red).
This commit is contained in:
teknium1
2026-09-12 20:26:20 -07:00
committed by Teknium
parent 3bef6b6a5c
commit 9ba5850e95
21 changed files with 239 additions and 125 deletions
+4 -10
View File
@@ -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:
+18 -1
View File
@@ -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
+2 -2
View File
@@ -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
+7 -22
View File
@@ -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 ------------------------------------------------------------------
+4 -10
View File
@@ -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
+4 -9
View File
@@ -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 [
+2 -4
View File
@@ -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:
-11
View File
@@ -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
+1 -1
View File
@@ -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")
+4 -9
View File
@@ -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")
+1 -1
View File
@@ -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")
+2 -1
View File
@@ -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
+6 -14
View File
@@ -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]]:
+6 -6
View File
@@ -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)
+8 -16
View File
@@ -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
+4 -4
View File
@@ -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()
+3 -3
View File
@@ -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()
@@ -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)
@@ -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
@@ -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')}"
+12
View File
@@ -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.