diff --git a/plugins/memory/__init__.py b/plugins/memory/__init__.py index 7bdf88abae..f7d2eff298 100644 --- a/plugins/memory/__init__.py +++ b/plugins/memory/__init__.py @@ -98,9 +98,7 @@ def _iter_entry_points(): eps = importlib.metadata.entry_points() if hasattr(eps, "select"): return list(eps.select(group=ENTRY_POINTS_GROUP)) - if isinstance(eps, dict): - return list(eps.get(ENTRY_POINTS_GROUP, [])) - return [ep for ep in eps if ep.group == ENTRY_POINTS_GROUP] + return list(eps.get(ENTRY_POINTS_GROUP, [])) if isinstance(eps, dict) else [ep for ep in eps if ep.group == ENTRY_POINTS_GROUP] except Exception as exc: logger.debug("Memory provider entry-point scan failed: %s", exc) return [] @@ -174,11 +172,7 @@ def discover_memory_providers() -> List[Tuple[str, str, bool]]: return results -def load_memory_provider( - name: str, - *, - register_skills: Optional[bool] = None, -) -> Optional["MemoryProvider"]: +def load_memory_provider(name: str, *, register_skills: Optional[bool] = None) -> Optional["MemoryProvider"]: """Load a MemoryProvider by name (bundled, user, project, then pip entry point); None if not found or failing to load. Skills register only for the configured active provider unless ``register_skills`` is explicit, so inspecting inactive @@ -214,31 +208,25 @@ def _instantiate_subclass(namespace) -> Optional["MemoryProvider"]: return None -def _load_provider_from_entry_point( - entry_point, - *, - register_skills: bool = True, -) -> Optional["MemoryProvider"]: - """Import a provider entry point and extract the MemoryProvider instance.""" +def _load_provider_from_entry_point(entry_point, *, register_skills: bool = True) -> Optional["MemoryProvider"]: + """Import a provider entry point and extract the MemoryProvider instance: an + instance, a subclass, a module with ``register(ctx)``, a factory / ``register`` + callable, or a namespace holding a subclass — in that order.""" from agent.memory_provider import MemoryProvider loaded = entry_point.load() - if isinstance(loaded, MemoryProvider): return loaded - if isinstance(loaded, type) and issubclass(loaded, MemoryProvider): try: return loaded() except Exception: pass - if hasattr(loaded, "register"): collector = _ProviderCollector(entry_point.name, register_skills=register_skills) loaded.register(collector) if collector.provider: return collector.provider - if callable(loaded): try: provider = loaded() @@ -246,7 +234,6 @@ def _load_provider_from_entry_point( return provider except TypeError: pass - collector = _ProviderCollector(entry_point.name, register_skills=register_skills) loaded(collector) return collector.provider @@ -257,11 +244,7 @@ def _load_provider_from_entry_point( return provider -def _load_provider_from_dir( - provider_dir: Path, - *, - register_skills: bool = True, -) -> Optional["MemoryProvider"]: +def _load_provider_from_dir(provider_dir: Path, *, register_skills: bool = True) -> Optional["MemoryProvider"]: """Import a provider module; ``register(ctx)`` first, else a top-level subclass.""" name = provider_dir.name mod = _loader.load_plugin_module( @@ -368,9 +351,7 @@ def _get_active_memory_provider() -> Optional[str]: return None -def _prune_inactive_memory_provider_skills( - active_provider: Optional[str] = None, -) -> None: +def _prune_inactive_memory_provider_skills(active_provider: Optional[str] = None) -> None: """Remove tracked skills that no longer belong to the active provider.""" if active_provider is None: active_provider = _get_active_memory_provider() diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py index 11fcd2f7c3..6a0d7f84ae 100644 --- a/plugins/memory/hindsight/__init__.py +++ b/plugins/memory/hindsight/__init__.py @@ -885,9 +885,8 @@ class HindsightMemoryProvider(MemoryProvider): def _daemon_start_worker(self) -> None: import traceback - log_dir = get_hermes_home() / "logs" - log_dir.mkdir(parents=True, exist_ok=True) - log_path = log_dir / "hindsight-embed.log" + log_path = get_hermes_home() / "logs" / "hindsight-embed.log" + log_path.parent.mkdir(parents=True, exist_ok=True) def _log(text: str, exc: bool = False) -> None: with open(log_path, "a", encoding="utf-8") as f: @@ -1046,16 +1045,9 @@ class HindsightMemoryProvider(MemoryProvider): metadata.update({name: value for name in _METADATA_ATTRS if (value := getattr(self, f"_{name}"))}) return metadata - def _build_retain_kwargs( - self, - content: str, - *, - context: str | None = None, - metadata: Dict[str, str] | None = None, - tags: List[str] | None = None, - occurred_at: str | None = None, - update_mode: str | None = None, - ) -> Dict[str, Any]: + def _build_retain_kwargs(self, content: str, *, context: str | None = None, + metadata: Dict[str, str] | None = None, tags: List[str] | None = None, + occurred_at: str | None = None, update_mode: str | None = None) -> Dict[str, Any]: """Build one aretain_batch item. The server resolves occurred_start/end (incl. relative phrases in content) from the item timestamp: explicit occurred_at wins, else the configured event clock.""" @@ -1220,14 +1212,8 @@ class HindsightMemoryProvider(MemoryProvider): # -- session lifecycle ------------------------------------------------------- - def on_session_switch( - self, - new_session_id: str, - *, - parent_session_id: str = "", - reset: bool = False, - **kwargs, - ) -> None: + def on_session_switch(self, new_session_id: str, *, parent_session_id: str = "", + reset: bool = False, **kwargs) -> None: """Rotate per-session state (/resume, /branch, /reset, /new, compression) so writes don't land in the previous session's document. Always: flush buffered turns under the OLD ids first (``retain_every_n_turns > 1`` would silently @@ -1301,10 +1287,7 @@ class HindsightMemoryProvider(MemoryProvider): # bounded join keeps shutdown predictable even if the daemon is wedged. writer = self._writer_thread if writer is not None and writer.is_alive(): - try: - self._retain_queue.put(_WRITER_SENTINEL) - except Exception: - pass + self._retain_queue.put(_WRITER_SENTINEL) writer.join(timeout=10.0) if writer.is_alive(): logger.warning("Hindsight writer did not stop within 10s; abandoning %d pending retain(s)", diff --git a/plugins/memory/hindsight/settings.py b/plugins/memory/hindsight/settings.py index a23f1f9ea0..9f7c65e834 100644 --- a/plugins/memory/hindsight/settings.py +++ b/plugins/memory/hindsight/settings.py @@ -137,9 +137,8 @@ def _resolve_bank_id_template(template: str, fallback: str, **placeholders: str) collapsed (``hermes-{user}`` -> ``hermes``). Empty/invalid template -> *fallback*.""" if not template: return fallback - sanitized = {k: _sanitize_bank_segment(v) for k, v in placeholders.items()} try: - rendered = template.format(**sanitized) + rendered = template.format(**{k: _sanitize_bank_segment(v) for k, v in placeholders.items()}) except (KeyError, IndexError) as exc: logger.warning("Invalid bank_id_template %r: %s — using fallback %r", template, exc, fallback) diff --git a/plugins/memory/hindsight/templates.py b/plugins/memory/hindsight/templates.py index 594a550341..289a9f9469 100644 --- a/plugins/memory/hindsight/templates.py +++ b/plugins/memory/hindsight/templates.py @@ -85,15 +85,7 @@ def probe_existing_customization(api_url: str, bank_id: str, api_key: str | None return bool(data.get("bank") or data.get("mental_models") or data.get("directives")) -def run_template_step( - *, - api_url: str, - bank_id: str, - api_key: str | None, - select, - cancelled, - log=print, -) -> str | None: +def run_template_step(*, api_url: str, bank_id: str, api_key: str | None, select, cancelled, log=print) -> str | None: """Wizard starter-template step. ``select(title, items, default, cancel_returns)`` is the picker (injected: testable without curses). Returns the applied template id, or None if skipped/blank/failed. Never raises — a template is a nice-to-have.""" diff --git a/plugins/observability/langfuse/__init__.py b/plugins/observability/langfuse/__init__.py index be08eef659..977fac1f64 100644 --- a/plugins/observability/langfuse/__init__.py +++ b/plugins/observability/langfuse/__init__.py @@ -404,10 +404,8 @@ def _safe_value(value: Any, *, max_chars: Optional[int] = None, depth: int = 0, def _extract_last_user_message(messages: Any) -> Any: if not isinstance(messages, list): return None - for message in reversed(messages): - if isinstance(message, dict) and message.get("role") == "user": - return {"role": "user", "content": _capture_content(message.get("content"))} - return None + last = next((m for m in reversed(messages) if isinstance(m, dict) and m.get("role") == "user"), None) + return None if last is None else {"role": "user", "content": _capture_content(last.get("content"))} def _coerce_request_messages(*, request_messages: Any = None, messages: Any = None, @@ -415,9 +413,7 @@ def _coerce_request_messages(*, request_messages: Any = None, messages: Any = No for candidate in (request_messages, messages, conversation_history): if isinstance(candidate, list): return candidate - if user_message is None: - return [] - return [{"role": "user", "content": user_message}] + return [] if user_message is None else [{"role": "user", "content": user_message}] def _serialize_system_prompt(system_prompt: Any) -> Optional[dict[str, Any]]: @@ -435,9 +431,7 @@ def _serialize_system_prompt(system_prompt: Any) -> Optional[dict[str, Any]]: text = "\n\n".join(parts) else: return None - if not text: - return None - return {"role": "system", "content": _capture_content(text)} + return {"role": "system", "content": _capture_content(text)} if text else None def _messages_for_langfuse_input(*, request_messages: Any = None, messages: Any = None, @@ -707,10 +701,10 @@ def _finalize_all_traces() -> None: for key, state in states: try: _end_children(state, include_subagents=True) - state.root_span.end() - _exit_root_ctx(state) except Exception as exc: # pragma: no cover - fail-open _debug(f"atexit finalize failed for {key}: {exc}") + else: + _end_root(state, f"atexit finalize for {key}") if states: _flush(_get_langfuse()) @@ -744,11 +738,7 @@ def _finish_trace(task_key: str, *, output: Any = None) -> None: getattr(state.root_span, method)(output=final_output) except Exception as exc: _debug(f"{label} failed: {exc}") - try: - state.root_span.end() - except Exception as exc: - _debug(f"root end() failed: {exc}") - _exit_root_ctx(state) + _end_root(state, "root end()") except Exception as exc: # pragma: no cover - fail-open _debug(f"finish trace failed: {exc}") # Last-chance end so an unexpected error still exports the root. @@ -991,12 +981,11 @@ def on_post_tool_call(*, tool_name: str = "", args: Any = None, result: Any = No if state is None: return observation = state.tools.pop(tool_call_id, None) if tool_call_id else None - if observation is None: - queue = state.pending_tools_by_name.get(tool_name) - if queue: - observation = queue.pop(0) - if not queue: - state.pending_tools_by_name.pop(tool_name, None) + queue = state.pending_tools_by_name.get(tool_name) if observation is None else None + if queue: + observation = queue.pop(0) + if not queue: + state.pending_tools_by_name.pop(tool_name, None) if observation is None: return @@ -1006,14 +995,12 @@ def on_post_tool_call(*, tool_name: str = "", args: Any = None, result: Any = No if tool_call_id: with _STATE_LOCK: state = _TRACE_STATE.get(task_key) - if state is not None: - for tool_call in reversed(state.turn_tool_calls): - if tool_call.get("id") == tool_call_id: - tool_call["output"] = safe_result_value - function_payload = tool_call.get("function") - if isinstance(function_payload, dict): - function_payload["output"] = safe_result_value - break + calls = state.turn_tool_calls if state is not None else [] + tool_call = next((tc for tc in reversed(calls) if tc.get("id") == tool_call_id), None) + if tool_call is not None: + tool_call["output"] = safe_result_value + if isinstance(tool_call.get("function"), dict): + tool_call["function"]["output"] = safe_result_value _end_observation( observation, output=safe_result_value, @@ -1137,14 +1124,13 @@ def on_subagent_stop(*, parent_turn_id: str = "", child_session_id: Any = None, def register(ctx) -> None: # Both hook-name variants so the plugin works across Hermes versions: # *_api_request fire per API call (preferred); *_llm_call once per turn. - ctx.register_hook("pre_api_request", on_pre_llm_request) - ctx.register_hook("post_api_request", on_post_llm_call) - ctx.register_hook("api_request_error", on_api_request_error) - ctx.register_hook("pre_llm_call", on_pre_llm_call) - ctx.register_hook("post_llm_call", on_post_llm_call) - ctx.register_hook("pre_tool_call", on_pre_tool_call) - ctx.register_hook("post_tool_call", on_post_tool_call) - ctx.register_hook("on_session_finalize", on_session_finalize) - ctx.register_hook("on_session_end", on_session_finalize) - ctx.register_hook("subagent_start", on_subagent_start) - ctx.register_hook("subagent_stop", on_subagent_stop) + hooks = ( + ("pre_api_request", on_pre_llm_request), ("post_api_request", on_post_llm_call), + ("api_request_error", on_api_request_error), ("pre_llm_call", on_pre_llm_call), + ("post_llm_call", on_post_llm_call), ("pre_tool_call", on_pre_tool_call), + ("post_tool_call", on_post_tool_call), ("on_session_finalize", on_session_finalize), + ("on_session_end", on_session_finalize), ("subagent_start", on_subagent_start), + ("subagent_stop", on_subagent_stop), + ) + for name, fn in hooks: + ctx.register_hook(name, fn)