"""Honcho client construction and ``HonchoClientConfig`` resolution. Config file resolution: $HERMES_HOME/honcho.json -> ~/.honcho/config.json -> env vars (HONCHO_API_KEY, HONCHO_ENVIRONMENT). Within a file, host-block fields win over flat/global fields, which win over defaults. """ from __future__ import annotations import atexit import contextlib import hashlib import ipaddress import json import logging import os import threading as _threading import time import weakref # --- per-identity client cache ------------------------------------------- One slot per client identity, # 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. from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any, Callable from urllib.parse import urlparse from agent.memory_provider import spawn_context_thread as _spawn_context_thread from agent.secret_scope import get_secret from hermes_cli.profiles import _get_default_hermes_home from hermes_constants import get_hermes_home from hermes_state_common import TITLE_SOURCE_DERIVED, TITLE_SOURCE_LLM from plugins.memory.honcho.client_cache import ( _DEFAULT_HTTP_TIMEOUT, _client_cache_key, _client_slots, _client_slots_lock, _honcho_json_timeout_memo, _refresh_oauth, _slot_for, ) if TYPE_CHECKING: from honcho import Honcho logger = logging.getLogger(__name__) HOST = "hermes" _AUTOMATIC_SESSION_TITLE_SOURCES = frozenset({TITLE_SOURCE_DERIVED, TITLE_SOURCE_LLM}) def _sanitize_url(url: str | None) -> str | None: """``url``, or None (with a warning) if it carries non-printable ASCII: a stray terminal escape in a pasted URL otherwise makes the SDK raise at client construction.""" if url is None or all(0x20 <= ord(c) < 0x7F for c in url): return url logger.warning("Honcho base_url contains non-printable characters and will be ignored: %r", url) return None def profile_host_key(profile: str | None) -> str: """Return the safe Honcho host key for a Hermes profile.""" if not profile or profile in {"default", "custom"}: return HOST sanitized = "".join(c if c.isalnum() or c in "_-" else "_" for c in profile).strip("_") return f"{HOST}_{sanitized or 'profile'}" def _host_block(raw: dict, host: str) -> dict: """Return host config, accepting legacy dot-form profile host keys.""" hosts = raw.get("hosts") or {} block = hosts.get(host, {}) if block or not host.startswith(f"{HOST}_"): return block return hosts.get(f"{HOST}.{host[len(HOST) + 1:]}", {}) def resolve_active_host() -> str: """Honcho host key: HERMES_HONCHO_HOST (profile-scoped .env), else the active profile. The config's ``defaultHost`` is honored only for the default profile so named profiles stay isolated — which is also why the override is read through the secret scope: from raw environ it would fold every multiplexed profile onto the default profile's host block and peer.""" explicit = (get_secret("HERMES_HONCHO_HOST", "") or "").strip() if explicit: return explicit try: from hermes_cli.profiles import get_active_profile_name profile_host = profile_host_key(get_active_profile_name()) except Exception: profile_host = HOST if profile_host == HOST: try: default_host = str(_read_config(resolve_config_path()).get("defaultHost", "")).strip() except Exception: default_host = "" if default_host: return default_host return profile_host def _read_config(path: Path) -> dict: """Parse a honcho.json; {} when absent (parse/OS errors propagate).""" return json.loads(path.read_text(encoding="utf-8-sig")) if path.exists() else {} def resolve_global_config_path() -> Path: """Return the shared Honcho config path for the current HOME.""" return Path.home() / ".honcho" / "config.json" def resolve_config_path() -> Path: """Active Honcho config path: $HERMES_HOME/honcho.json -> default profile's honcho.json (host blocks accumulate there via setup/clone) -> ~/.honcho/config.json (also the first-time-setup write target when nothing exists).""" local_path = get_hermes_home() / "honcho.json" if local_path.exists(): return local_path default_path = _get_default_hermes_home() / "honcho.json" if default_path != local_path and default_path.exists(): return default_path return resolve_global_config_path() # --- config coercion ------------------------------------------------------- _RECALL_MODES = ({"auto": "hybrid"}, {"hybrid", "context", "tools"}, "hybrid") _OBSERVATION_MODES = ( {"shared": "unified", "separate": "directional", "cross": "directional"}, {"unified", "directional"}, "directional", ) _VALID_REASONING_LEVELS = ("minimal", "low", "medium", "high", "max") # Granular observation booleans derived from the legacy string mode; explicit # per-peer config always wins over these presets. _OBSERVATION_PRESETS = { "directional": {"user_observe_me": True, "user_observe_others": True, "ai_observe_me": True, "ai_observe_others": True}, "unified": {"user_observe_me": True, "user_observe_others": False, "ai_observe_me": False, "ai_observe_others": True}, } def _normalize_choice(val: str, spec: tuple[dict, set, str]) -> str: """Map aliases, then fall back to the default for unknown values.""" aliases, valid, default = spec val = aliases.get(val, val) return val if val in valid else default def _first_set(*vals, default): """First non-None value, else default.""" return next((val for val in vals if val is not None), default) def _first_parsed(vals, caster: Callable[[Any], Any], default): """First non-None value that ``caster`` accepts, else default.""" for val in (v for v in vals if v is not None): try: return caster(val) except (ValueError, TypeError): pass return default def _positive_float(value: Any) -> float: parsed = float(value) if parsed <= 0: raise ValueError(value) return parsed def _resolve_optional_float(*values: Any) -> float | None: """First value (blank strings included) that parses as a positive float, else None.""" return _first_parsed(values, _positive_float, None) def _parse_dialectic_depth_levels(vals, depth: int) -> list[str] | None: """First list value, validated and padded/truncated to ``depth``; None if unset.""" for val in vals: if isinstance(val, list): levels = [lvl if lvl in _VALID_REASONING_LEVELS else "low" for lvl in val[:depth]] return levels + ["low"] * (depth - len(levels)) return None def _resolve_observation(mode: str, observation_obj: dict | None) -> dict: """Per-peer observation booleans: ``observation`` object fields override the mode preset.""" preset = _OBSERVATION_PRESETS.get(mode, _OBSERVATION_PRESETS["directional"]) if not isinstance(observation_obj, dict) or not observation_obj: return dict(preset) return { f"{kind}_observe_{who}": (observation_obj.get(kind) or {}).get(json_key, preset[f"{kind}_observe_{who}"]) for kind in ("user", "ai") for who, json_key in (("me", "observeMe"), ("others", "observeOthers")) } class _HostLookup: """Host-block-over-root field lookups for one honcho.json.""" def __init__(self, host_block: dict, raw: dict): self.host, self.raw = host_block, raw def pick(self, key: str, default=None): """Truthy host value, else ``raw.get(key, default)``.""" return self.host.get(key) or self.raw.get(key, default) def pick_set(self, key: str, default=None): """Non-None host value, else ``raw.get(key, default)``.""" return _first_set(self.host.get(key), default=self.raw.get(key, default)) def vals(self, *keys: str) -> list: """Host values then root values for the aliased ``keys`` (lookup order).""" return [self.host.get(k) for k in keys] + [self.raw.get(k) for k in keys] def flag(self, *keys: str, default: bool) -> bool: """First non-None value across host-then-root for the aliased ``keys``, as bool.""" return bool(_first_set(*self.vals(*keys), default=default)) def parsed(self, key: str, caster: Callable[[Any], Any], default): """First host-then-root value ``caster`` accepts, else default.""" return _first_parsed(self.vals(key), caster, default) def present(self, key: str, default=None): """Host value if the key is PRESENT there (even empty/None), else root.""" return self.host[key] if key in self.host else self.raw.get(key, default) def string(self, key: str, default: str = "") -> str: """String field where a host-level empty string can override root.""" value = self.present(key, default) return default if value is None else str(value).strip() def string_map(self, key: str) -> dict[str, str]: """String-to-string map; a host-level map replaces the root map wholesale.""" source = self.present(key) if not isinstance(source, dict): return {} pairs = ((str(k).strip(), str(v).strip() if v is not None else "") for k, v in source.items()) return {k: v for k, v in pairs if k and v} def _is_local_base_url(base_url: str | None) -> bool: """True for loopback/RFC1918/link-local/ULA/CGNAT self-hosted Honcho URLs. Local deployments can run without auth but the SDK needs a non-empty api_key, so LAN/VPN URLs get the same placeholder-key treatment as localhost.""" if not base_url: return False try: host = (urlparse(base_url).hostname or "").strip().lower() except Exception: host = "" if host in {"localhost", "127.0.0.1", "::1"}: return True try: ip = ipaddress.ip_address(host) except ValueError: return False # Tailscale/other VPN setups often sit in carrier-grade NAT space (100.64.0.0/10). cgnat = ip.version == 4 and ipaddress.ip_address("100.64.0.0") <= ip <= ipaddress.ip_address("100.127.255.255") return ip.is_loopback or ip.is_private or ip.is_link_local or cgnat def _env_base_url() -> str | None: """HONCHO_BASE_URL / HONCHO_URL (the SDK's own var). A self-hosted URL varies per profile, so it is read through the secret scope: the scoped HONCHO_API_KEY beside it must not be sent to the default profile's server.""" return (get_secret("HONCHO_BASE_URL", "") or "").strip() or (get_secret("HONCHO_URL", "") or "").strip() or None def _connection_fields(look: _HostLookup, host: str, path: Path) -> dict[str, Any]: """Resolve identity/credential/transport fields (host block -> root -> env).""" raw, host_block = look.raw, look.host api_key = look.pick("apiKey") or get_secret("HONCHO_API_KEY") # Named-profile host blocks do NOT inherit the default host's apiKey (profiles # are credential-isolated); the failure is silent 401s, so warn loudly. if not api_key and host_block and host != HOST and _host_block(raw, HOST).get("apiKey"): logger.warning("Honcho host block '%s' has no apiKey; the default '%s' host's key " "is NOT inherited (profiles are credential-isolated). Set apiKey on " "hosts.%s in %s or this profile runs unauthenticated.", host, HOST, host, path) # The SDK's native format (and Claude Desktop) nests the URL at endpoint.baseUrl; # read it before the flat Hermes spellings. endpoint_block = raw.get("endpoint") native_base_url = endpoint_block.get("baseUrl") if isinstance(endpoint_block, dict) else None base_url = _sanitize_url(host_block.get("baseUrl") or host_block.get("base_url") or native_base_url or raw.get("baseUrl") or raw.get("base_url") or _env_base_url()) return { "workspace_id": look.pick("workspace") or host, "ai_peer": look.pick("aiPeer") or host, "api_key": api_key, "environment": look.pick("environment", "production"), "base_url": base_url, "timeout": _resolve_optional_float(*look.vals("timeout", "requestTimeout"), os.environ.get("HONCHO_TIMEOUT")), # Explicit enabled (host, then root) wins; else auto-enable on key/url. "enabled": _first_set(*look.vals("enabled"), default=bool(api_key or base_url)), } def _behavior_fields(look: _HostLookup, explicitly_configured: bool) -> dict[str, Any]: """Resolve memory-behavior tuning fields (host block -> root -> defaults).""" raw_wf = look.pick("writeFrequency") or "async" write_frequency: str | int = _first_parsed([raw_wf], int, str(raw_wf)) depth = look.parsed("dialecticDepth", lambda v: max(1, min(int(v), 3)), 1) # Migration guard: configs that predate observationMode keep the old # "unified" default; fresh installs get "directional" (all observations on). observation_mode = _normalize_choice( look.pick("observationMode") or ("unified" if explicitly_configured else "directional"), _OBSERVATION_MODES) return { "peer_name": look.pick("peerName"), # pinUserPeer is the clearer name; the original pinPeerName stays accepted. "pin_peer_name": look.flag("pinUserPeer", "pinPeerName", default=False), "user_peer_aliases": look.string_map("userPeerAliases"), "runtime_peer_prefix": look.string("runtimePeerPrefix"), "save_messages": look.pick_set("saveMessages", True), "write_frequency": write_frequency, "context_tokens": look.parsed("contextTokens", int, None), "dialectic_reasoning_level": look.pick("dialecticReasoningLevel") or "low", "dialectic_dynamic": look.flag("dialecticDynamic", default=True), "dialectic_max_chars": look.parsed("dialecticMaxChars", int, 600), "dialectic_depth": depth, "dialectic_depth_levels": _parse_dialectic_depth_levels(look.vals("dialecticDepthLevels"), depth), "reasoning_heuristic": look.flag("reasoningHeuristic", default=True), "reasoning_level_cap": look.pick("reasoningLevelCap") or "high", "message_max_chars": look.parsed("messageMaxChars", int, 25000), "dialectic_max_input_chars": look.parsed("dialecticMaxInputChars", int, 10000), "recall_mode": _normalize_choice(look.pick("recallMode") or "hybrid", _RECALL_MODES), "recall_sync": look.flag("recallSync", default=False), "init_on_session_start": look.flag("initOnSessionStart", default=False), "injection_frequency": look.pick("injectionFrequency", "every-turn"), "context_cadence": look.parsed("contextCadence", int, 1), "dialectic_cadence": look.parsed("dialecticCadence", int, 1), "query_rewrite": look.flag("queryRewrite", default=False), "first_turn_base_wait": look.parsed("firstTurnBaseWait", lambda v: max(0.0, float(v)), 3.0), "first_turn_dialectic_wait": look.parsed("firstTurnDialecticWait", lambda v: max(0.0, float(v)), 2.0), "observation_mode": observation_mode, **_resolve_observation(observation_mode, look.pick("observation")), "session_strategy": look.pick("sessionStrategy", "per-directory"), "session_peer_prefix": look.pick_set("sessionPeerPrefix", False), "a2a_sessions": look.flag("a2aSessions", default=True), "session_ai_peer_prefix": look.pick_set("sessionAiPeerPrefix", False), } @dataclass class HonchoClientConfig: """Configuration for Honcho client, resolved for a specific host.""" host: str = HOST workspace_id: str = "hermes" api_key: str | None = None environment: str = "production" base_url: str | None = None # self-hosted override of the environment mapping timeout: float | None = None # SDK HTTP timeout, seconds # Identity peer_name: str | None = None ai_peer: str = "hermes" # True: peer_name wins over gateway runtime identity (Telegram UID, ...), so a # single-user deployment keeps one memory across platforms. # This keeps memory unified across platforms for single-user deployments where Honcho's one peer-name is # an unambiguous identity — otherwise each platform would fork memory into its own peer (#14984). # Default ``False`` preserves existing multi-user behaviour. pin_peer_name: bool = False # Gateway runtime user id -> stable Honcho peer; host map replaces root map. user_peer_aliases: dict[str, str] = field(default_factory=dict) runtime_peer_prefix: str = "" # prefix for unknown runtime user ids, e.g. "telegram_" # Toggles enabled: bool = False save_messages: bool = True write_frequency: str | int = "async" # "async" | "turn" | "session" | every-N-turns int context_tokens: int | None = None # prefetch budget; None = uncapped # Dialectic (peer.chat) settings dialectic_reasoning_level: str = "low" # minimal | low | medium | high | max dialectic_dynamic: bool = True # model may override the level via honcho_reasoning dialectic_max_chars: int = 600 # auto-injection cap; explicit calls bypass it dialectic_depth: int = 1 # .chat() passes per cycle (1-3): audit / synthesis / reconcile dialectic_depth_levels: list[str] | None = None # per-pass levels; None = proportional defaults reasoning_heuristic: bool = True # scale auto-injected level up on longer queries reasoning_level_cap: str = "high" # Honcho API limits (Honcho cloud: 25000 / 10000) — configurable for self-hosts message_max_chars: int = 25000 dialectic_max_input_chars: int = 10000 # "hybrid" (context + tools) | "context" (no tools) | "tools" (no auto context) recall_mode: str = "hybrid" recall_sync: bool = False # bounded current-query automatic recall init_on_session_start: bool = False # tools mode: init eagerly instead of on first tool call injection_frequency: str = "every-turn" # or "first-turn" context_cadence: int = 1 # min turns between peer.context() calls dialectic_cadence: int = 1 # min turns between dialectic prefetches query_rewrite: bool = False # rewrite the user message into a retrieval query (one aux LLM call) # Bounded synchronous waits on turn 1, seconds; 0 = fully async first turn. first_turn_base_wait: float = 3.0 first_turn_dialectic_wait: float = 2.0 # Legacy string shorthand; the granular per-peer booleans below are preferred # and map 1:1 to Honcho's SessionPeerConfig. observation_mode: str = "directional" user_observe_me: bool = True user_observe_others: bool = True ai_observe_me: bool = True ai_observe_others: bool = True # Session resolution session_strategy: str = "per-directory" session_peer_prefix: bool = False # Bot-authored DMs write into their own session per sender bot. a2a_sessions: bool = True # Prefixes every resolved session name with ``{ai_peer}-``. The gateway_session_key branch is # AI-peer-agnostic, so several AI peers sharing one workspace + peerName + chat key would collide. session_ai_peer_prefix: bool = False sessions: dict[str, str] = field(default_factory=dict) raw: dict[str, Any] = field(default_factory=dict) # A hosts. block or explicit enabled flag, vs auto-enabled from a stray env key. explicitly_configured: bool = False # Provenance captured at resolution time; bound consumers use these instead of # re-resolving (the resolvers read a ContextVar background threads can't see). # Provenance: WHERE this config was resolved from, captured at resolution time (inside the caller's # profile scope). Bound consumers (session manager, OAuth refresh paths) use these instead of # re-resolving resolve_config_path()/get_hermes_home() later — those resolvers read a ContextVar that # background threads cannot see, so re-resolution from a daemon thread silently lands on the DEFAULT # profile (#69123, #74065). config_path: Path | None = None hermes_home: Path | None = None def bound_config_path(self) -> Path: """Config path this was resolved from; ambient fallback for hand-built configs.""" return self.config_path if self.config_path is not None else resolve_config_path() @classmethod def from_env(cls, workspace_id: str = "hermes", host: str | None = None) -> HonchoClientConfig: """Create config from environment variables (fallback).""" resolved_host = host or resolve_active_host() api_key = get_secret("HONCHO_API_KEY") base_url = _sanitize_url(_env_base_url()) return cls( host=resolved_host, workspace_id=workspace_id, api_key=api_key, base_url=base_url, environment=get_secret("HONCHO_ENVIRONMENT", "") or "production", timeout=_resolve_optional_float(os.environ.get("HONCHO_TIMEOUT")), ai_peer=resolved_host, enabled=bool(api_key or base_url), config_path=resolve_config_path(), hermes_home=get_hermes_home(), ) @classmethod def from_global_config(cls, host: str | None = None, config_path: Path | None = None) -> HonchoClientConfig: """Config from the resolved Honcho config path, falling back to env. ``host=None`` derives it from the active Hermes profile.""" resolved_host = host or resolve_active_host() path = config_path or resolve_config_path() if not path.exists(): logger.debug("No global Honcho config at %s, falling back to env", path) return cls.from_env(host=resolved_host) try: raw = _read_config(path) except (json.JSONDecodeError, OSError) as e: logger.warning("Failed to read %s: %s, falling back to env", path, e) return cls.from_env(host=resolved_host) host_block = _host_block(raw, resolved_host) explicitly_configured = bool(host_block) or raw.get("enabled") is True look = _HostLookup(host_block, raw) return cls( host=resolved_host, **_connection_fields(look, resolved_host, path), **_behavior_fields(look, explicitly_configured), sessions=raw.get("sessions", {}), raw=raw, explicitly_configured=explicitly_configured, config_path=path, hermes_home=get_hermes_home(), ) @staticmethod def _git_repo_name(cwd: str) -> str | None: """Return the git repo root directory name, or None if not in a repo.""" import subprocess try: root = subprocess.run(["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, encoding='utf-8', errors='replace', cwd=cwd, timeout=5, stdin=subprocess.DEVNULL) except (OSError, subprocess.TimeoutExpired): return None return Path(root.stdout.strip()).name if root.returncode == 0 else None # Honcho rejects session IDs over 100 chars; long gateway keys (Matrix # rooms + thread ids, Slack threads) overflow after sanitization. _HONCHO_SESSION_ID_MAX_LEN, _HONCHO_SESSION_ID_HASH_LEN = 100, 8 @classmethod def _enforce_session_id_limit(cls, sanitized: str, original: str) -> str: """Truncate to the limit with a ``-`` suffix hashed over the ORIGINAL key: two long keys sharing a prefix stay distinct; keys that sanitize identically still collide intentionally (same logical session).""" max_len, hash_len = cls._HONCHO_SESSION_ID_MAX_LEN, cls._HONCHO_SESSION_ID_HASH_LEN if len(sanitized) <= max_len: return sanitized digest = hashlib.sha256(original.encode("utf-8")).hexdigest()[:hash_len] prefix = sanitized[: max_len - hash_len - 1].rstrip("-") return f"{prefix}-{digest}" def _with_peer_prefix(self, name: str) -> str: return f"{self.peer_name}-{name}" if self.session_peer_prefix and self.peer_name else name def resolve_session_name( self, cwd: str | None = None, session_title: str | None = None, session_id: str | None = None, gateway_session_key: str | None = None, session_title_source: str | None = None, ) -> str | None: """Resolve the Honcho session name; with ``session_ai_peer_prefix`` the result is prefixed ``{ai_peer}-`` on every path, including the AI-peer-agnostic gateway session key.""" import re result = self._resolve_session_name_base(cwd=cwd, session_title=session_title, session_id=session_id, gateway_session_key=gateway_session_key, session_title_source=session_title_source) if result and self.session_ai_peer_prefix and self.ai_peer: ai = re.sub(r'[^a-zA-Z0-9_-]+', '-', self.ai_peer).strip('-') if ai: prefixed = f"{ai}-{result}" return self._enforce_session_id_limit(prefixed, prefixed) return result def _resolve_session_name_base( self, cwd: str | None = None, session_title: str | None = None, session_id: str | None = None, gateway_session_key: str | None = None, session_title_source: str | None = None, ) -> str | None: """Order: gateway session key (per-chat isolation no cwd/strategy gives) -> per-session strategy's session_id (authoritative, so a generated title never remaps a live conversation) -> sessions map override -> /title -> per-repo (git root name) -> per-directory (basename) -> global (workspace).""" import re def _slug(text: str) -> str: return re.sub(r'[^a-zA-Z0-9_-]+', '-', text).strip('-') cwd = cwd or os.getcwd() if gateway_session_key and _slug(gateway_session_key): return self._enforce_session_id_limit(_slug(gateway_session_key), gateway_session_key) if self.session_strategy == "per-session" and session_id: return self._with_peer_prefix(session_id) manual = self.sessions.get(cwd) if manual: return manual # Absent provenance retains the legacy explicit-title override. Generated # display titles must not change a strategy-selected memory identity. if session_title and session_title_source not in _AUTOMATIC_SESSION_TITLE_SOURCES and _slug(session_title): return self._with_peer_prefix(_slug(session_title)) if self.session_strategy == "per-repo": return self._with_peer_prefix(self._git_repo_name(cwd) or Path(cwd).name) if self.session_strategy in {"per-directory", "per-session"}: return self._with_peer_prefix(Path(cwd).name) return self.workspace_id # Threads keyed by the provider or manager that owns them, so shutdown never waits on another agent's work. _plugin_threads: "weakref.WeakKeyDictionary[Any, weakref.WeakSet]" = weakref.WeakKeyDictionary() _plugin_threads_lock = _threading.Lock() def spawn_context_thread( target, *, name: str, daemon: bool = True, args: tuple = (), owner: Any = None, ) -> "_threading.Thread": """agent.memory_provider.spawn_context_thread plus an ``owner``: the thread is registered so join_plugin_threads can wait on exactly the threads this provider or manager spawned.""" thread = _spawn_context_thread(target, name=name, daemon=daemon, args=args) if owner is not None: with _plugin_threads_lock: _plugin_threads.setdefault(owner, weakref.WeakSet()).add(thread) return thread def join_plugin_threads(owners, timeout: float) -> list[str]: """Join every live thread spawned for ``owners`` within one shared ``timeout``. Returns the names of the threads still running when the budget ran out.""" deadline = time.monotonic() + max(0.0, timeout) me = _threading.current_thread() with _plugin_threads_lock: threads = [t for owner in owners if owner is not None for t in list(_plugin_threads.get(owner, ())) if t.is_alive() and t is not me] for thread in threads: thread.join(timeout=max(0.0, deadline - time.monotonic())) return [t.name for t in threads if t.is_alive()] def close_honcho_clients() -> None: """Close the HTTP pool of every cached client and drop the slots. Process exit only: one client serves every manager with the same identity, so a per-agent shutdown must never call this.""" with _client_slots_lock: slots = list(_client_slots.values()) _client_slots.clear() for slot in slots: close = getattr(getattr(slot.peek(), "_http", None), "close", None) if callable(close): with contextlib.suppress(Exception): close() _exit_close_registered = False def _register_exit_close() -> None: global _exit_close_registered with _client_slots_lock: if _exit_close_registered: return _exit_close_registered = True atexit.register(close_honcho_clients) 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 multi-profile processes don't share a first-config-wins client. With no config the active honcho.json is resolved — correct only on threads that see the profile ContextVar; pass a bound config elsewhere. Each identity's client is built once under concurrent first calls. See #69123, #74065. """ key = _client_cache_key(config) slot = _slot_for(key) cached = slot.peek() if cached is not None: _refresh_oauth(config, cached, slot) refreshed = slot.peek() if refreshed is not None: return refreshed # Slot was reset by a failed in-place rotation — rebuild below. if config is None: config = HonchoClientConfig.from_global_config() # Start with a live access token rather than 401ing an hour in. _refresh_oauth(config) if not config.api_key and not config.base_url: raise ValueError("Honcho API key not found. Get your API key at https://app.honcho.dev, " "then run 'hermes honcho setup' or set HONCHO_API_KEY. " "For local instances, set HONCHO_BASE_URL instead.") # Build inside the singleton factory so racing callers share one client. return slot.get(lambda: _build_client(config)) def _build_client(config: HonchoClientConfig) -> "Honcho": """Construct the SDK client (runs inside the slot factory so racing callers share one).""" with contextlib.suppress(Exception): # lazy-dep failures fall through to the canonical import error below from pm import ensure_import as _lazy_ensure _lazy_ensure("honcho") try: from honcho import Honcho except ImportError: raise ImportError("honcho-ai is required for Honcho integration. Install it with: pip install honcho-ai " "(or run `hermes honcho setup` to configure).") # config.yaml honcho.base_url / timeout fill whatever honcho.json left unset. base_url, timeout = config.base_url, config.timeout if not base_url or timeout is None: with contextlib.suppress(Exception): from hermes_cli.config import load_config honcho_cfg = load_config().get("honcho", {}) if isinstance(honcho_cfg, dict): base_url = base_url or _sanitize_url(honcho_cfg.get("base_url", "").strip() or None) if timeout is None: timeout = _resolve_optional_float(honcho_cfg.get("timeout"), honcho_cfg.get("request_timeout")) if timeout is None: timeout = _DEFAULT_HTTP_TIMEOUT # an unconfigured install must not hang on a stalled request if base_url: logger.info("Initializing Honcho client (base_url: %s, workspace: %s)", base_url, config.workspace_id) else: # Name the SDK's environment fallback at INFO so a self-hosted user whose # config wasn't picked up notices they're talking to the public cloud. logger.info("Initializing Honcho client (host: %s, workspace: %s, base_url unset — SDK will resolve from environment=%s)", config.host, config.workspace_id, config.environment) # Local instances need no key but the SDK wants a non-empty string: honor a # key set EXPLICITLY in honcho.json (host block or root) and treat an # env-sourced key as likely-cloud, substituting the placeholder. raw = config.raw or {} explicit_key = _host_block(raw, config.host).get("apiKey") or raw.get("apiKey") api_key = "local" if _is_local_base_url(base_url) and not explicit_key else config.api_key kwargs: dict = {"workspace_id": config.workspace_id, "api_key": api_key, "environment": config.environment, "timeout": timeout} if base_url: # The SDK's route builders already carry the version prefix ("/v3/..."), so # strip a trailing version segment from any base_url to avoid "/v3/v3/...". import re kwargs["base_url"] = re.sub(r"/v\d+/*$", "", base_url).rstrip("/") _register_exit_close() return Honcho(**kwargs) def reset_honcho_client() -> None: """Reset all cached Honcho clients (tests, OAuth re-login).""" with _client_slots_lock: _client_slots.clear() _honcho_json_timeout_memo.clear()