"""langfuse — Hermes plugin for Langfuse observability. Traces Hermes conversations, LLM calls, and tool usage to Langfuse. Activation is handled by the Hermes plugin system — standalone plugins only load when listed in ``plugins.enabled`` (via ``hermes plugins enable observability/langfuse`` or ``hermes tools → Langfuse Observability``). At runtime the plugin also requires the ``langfuse`` SDK and credentials; if either is missing the hooks are inert. Required env vars (set via ``hermes tools`` or ~/.hermes/.env): HERMES_LANGFUSE_PUBLIC_KEY - Langfuse project public key (pk-lf-...) HERMES_LANGFUSE_SECRET_KEY - Langfuse project secret key (sk-lf-...) HERMES_LANGFUSE_BASE_URL - Langfuse server URL (default: https://cloud.langfuse.com) Optional env vars: HERMES_LANGFUSE_ENV - environment tag (e.g. "production", "local") HERMES_LANGFUSE_RELEASE - release/version tag HERMES_LANGFUSE_SAMPLE_RATE - sampling rate 0.0–1.0 (default: 1.0) HERMES_LANGFUSE_MAX_CHARS - max chars per field (default: 12000) HERMES_LANGFUSE_CAPTURE - content capture mode (default: "sanitized") metadata - no content: sizes, roles, tool names, IDs, usage, cost only sanitized - content with secret-pattern redaction + truncation full - raw content (truncated only); explicit opt-in HERMES_LANGFUSE_DEBUG - set to "true" for verbose logging """ from __future__ import annotations import json import logging import os import re import threading import time from dataclasses import dataclass, field from typing import Any, Dict, Optional logger = logging.getLogger(__name__) try: from langfuse import Langfuse, propagate_attributes except Exception: # pragma: no cover - fail-open when optional dep is missing Langfuse = None propagate_attributes = None @dataclass class TraceState: trace_id: str root_ctx: Any root_span: Any generations: Dict[str, Any] = field(default_factory=dict) tools: Dict[str, Any] = field(default_factory=dict) pending_tools_by_name: Dict[str, list] = field(default_factory=dict) turn_tool_calls: list[dict[str, Any]] = field(default_factory=list) # Keyed by child_session_id: subagent_stop carries no child_subagent_id. subagents: Dict[str, Any] = field(default_factory=dict) # Fingerprints of MoA fan-outs already recorded. The client holds its last # fan-out until the next one, so a tool-loop turn would re-emit the same # advisors on every API call without this. moa_emitted: set = field(default_factory=set) last_updated_at: float = field(default_factory=time.time) _STATE_LOCK = threading.Lock() _TRACE_STATE: Dict[str, TraceState] = {} # Hard cap on live trace state. Each turn keys _TRACE_STATE by a unique # turn_id, and an entry is normally reclaimed by _finish_trace when a turn # ends cleanly (final response has content and no tool calls). A turn that # never reaches that state — interrupted, a tool-only final step, or empty # final content — would otherwise linger forever, so over the cap we evict # the least-recently-updated entries (ending their root span first). The cap # is far above any realistic concurrent-live-turn working set; it exists only # to bound the leak from non-finalizing turns, not to limit concurrency. _MAX_TRACE_STATE = 256 _LANGFUSE_CLIENT = None # Guards _LANGFUSE_CLIENT initialization against the TOCTOU race: two # concurrent first callers both pass the ``is not None`` guard, both # construct a Langfuse(**kwargs) client, and the loser's client leaks an # open HTTPS connection + background flush thread. _STATE_LOCK is not # reused here because it guards _TRACE_STATE (hot path) and nesting the # two locks would risk deadlock with future callers. _LANGFUSE_CLIENT_LOCK = threading.Lock() _READ_FILE_LINE_RE = re.compile(r"^\s*(\d+)\|(.*)$") _READ_FILE_HEAD_LINES = 25 _READ_FILE_TAIL_LINES = 15 # Langfuse-issued keys always carry these prefixes (cloud or self-hosted — # the prefix is baked into the server-side issuance flow, not a UI hint). # Anything else (`placeholder`, `test-key`, `your-langfuse-key`, etc.) is a # leftover template value and would cause the SDK to silently accept the # credentials at construction time but drop every trace at flush time. # See #23823 — the silent-failure bug this guard fixes. _LANGFUSE_KEY_PREFIXES: Dict[str, str] = { "HERMES_LANGFUSE_PUBLIC_KEY": "pk-lf-", "HERMES_LANGFUSE_SECRET_KEY": "sk-lf-", } def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() def _env_bool(*names: str) -> bool: for name in names: value = _env(name).lower() if value: return value in {"1", "true", "yes", "on"} return False def _debug_enabled() -> bool: return _env_bool("HERMES_LANGFUSE_DEBUG") def _debug(message: str) -> None: if _debug_enabled(): logger.info("Langfuse tracing: %s", message) # --------------------------------------------------------------------------- # Capture modes # --------------------------------------------------------------------------- _CAPTURE_MODES = ("metadata", "sanitized", "full") _DEFAULT_CAPTURE_MODE = "sanitized" _warned_invalid_capture = False def _capture_mode() -> str: """Resolve the content-capture mode: ``metadata | sanitized | full``. Read per call (cheap env lookup) so tests and long-lived processes can flip modes without a client reset. Invalid values warn once per process and fall back to the default rather than silently capturing more than the operator intended. """ global _warned_invalid_capture value = _env("HERMES_LANGFUSE_CAPTURE").lower() if not value: return _DEFAULT_CAPTURE_MODE if value in _CAPTURE_MODES: return value if not _warned_invalid_capture: _warned_invalid_capture = True logger.warning( "Langfuse plugin: invalid HERMES_LANGFUSE_CAPTURE=%r, falling back " "to %r (valid: %s)", value, _DEFAULT_CAPTURE_MODE, ", ".join(_CAPTURE_MODES), ) return _DEFAULT_CAPTURE_MODE # Secret redaction in ``sanitized`` mode reuses the project-wide # ``agent.redact.redact_sensitive_text(force=True)`` — which covers 50+ credential # patterns, private keys, JWTs, auth headers, DB connection strings, and env # assignments with pre-check-gated regex. The ``force=True`` flag ensures # redaction runs even if the user has ``security.redact_secrets: false`` set — # appropriate for an observability plugin exporting to an external service. def _redact_secrets(value: str) -> str: try: from agent.redact import redact_sensitive_text return redact_sensitive_text(value, force=True) except Exception: return value def _describe_content(value: Any, *, depth: int = 0) -> Any: """Metadata-mode stand-in for content: shape and size, never payload.""" if value is None or isinstance(value, bool): return value if isinstance(value, (int, float)): return {"omitted": True, "type": "number"} if isinstance(value, bytes): return {"omitted": True, "type": "bytes", "length": len(value)} if isinstance(value, str): return {"omitted": True, "type": "text", "chars": len(value)} if isinstance(value, dict): return { "omitted": True, "type": "object", "keys": [str(k) for k in list(value.keys())[:20]], } if isinstance(value, (list, tuple, set)): return {"omitted": True, "type": "array", "items": len(value)} return {"omitted": True, "type": type(value).__name__} def _capture_content(value: Any, *, parse_json_strings: bool = False, tool_name: str = "", args: Any = None) -> Any: """Apply the active capture mode to a CONTENT value. Metadata fields (provider, model, IDs, counts) should NOT go through this — they stay as-is in every mode. Only prompt/response text, tool arguments, and tool results are content. """ mode = _capture_mode() if mode == "metadata": return _describe_content(value) if tool_name or args is not None: value = _normalize_payload(value, tool_name=tool_name, args=args) return _safe_value(value, parse_json_strings=parse_json_strings) # Sentinel: "_get_langfuse() has tried and failed". Lets us short-circuit # every subsequent hook call without re-checking env vars or re-attempting # SDK init. Tests clear this by reloading the module via # ``sys.modules.pop(...) + importlib.import_module(...)`` rather than via a # dedicated reset function. Runtime callers cannot reset the cache; if an # operator fixes a misconfigured credential they must restart the process. _INIT_FAILED = object() def _redact_key_preview(value: str) -> str: """Return a brief, log-safe preview of a credential value. Keeps enough characters to disambiguate common placeholders (``placeholder``, ``test-key``, ``your-key``) without echoing a real secret in full if an operator pasted one into the wrong env var. Used only for the once-per-process placeholder-detection warning in :func:`_get_langfuse`. """ if not value: return "" if len(value) <= 12: return repr(value) return repr(value[:6] + "...") def _validate_langfuse_key(env_name: str, value: str) -> Optional[str]: """Return an error message if ``value`` is not a real Langfuse key. Returns ``None`` when the value matches the documented Langfuse prefix for ``env_name``, or when no prefix is registered for the name (in which case we trust the operator). When validation fails the returned string is suitable for direct inclusion in a single log line — it names the env var and shows a safe preview. """ expected = _LANGFUSE_KEY_PREFIXES.get(env_name, "") if not expected: return None if value.startswith(expected): return None return ( f"{env_name}={_redact_key_preview(value)} " f"(expected {expected!r} prefix)" ) def _get_langfuse() -> Optional[Langfuse]: """Return a cached Langfuse client, or ``None`` if unavailable. Activation of this plugin is controlled by the Hermes plugin system — this function only handles the runtime-availability gate (SDK installed + credentials present). The result is cached: on the first call we try to construct a client, and every subsequent call returns that client (or fast-returns ``None`` if init failed). Thread-safe: ``_LANGFUSE_CLIENT_LOCK`` serializes the first build so concurrent callers can't both pass the ``is not None`` guard, both construct a ``Langfuse(**kwargs)`` client, and leak the loser's open HTTP connection and background flush thread (same TOCTOU class fixed for the Honcho and FAL clients in ``plugins/plugin_utils.py``). """ global _LANGFUSE_CLIENT # Fast path — already settled (success or _INIT_FAILED); no lock needed. if _LANGFUSE_CLIENT is _INIT_FAILED: return None if _LANGFUSE_CLIENT is not None: return _LANGFUSE_CLIENT with _LANGFUSE_CLIENT_LOCK: # Re-check inside the lock: a racing thread may have completed init # while we were waiting. if _LANGFUSE_CLIENT is _INIT_FAILED: return None if _LANGFUSE_CLIENT is not None: return _LANGFUSE_CLIENT if Langfuse is None: logger.warning( "Langfuse plugin is enabled but the langfuse SDK is unavailable; " "tracing is disabled. Run `hermes tools` and configure Langfuse " "Observability to reinstall it." ) _LANGFUSE_CLIENT = _INIT_FAILED return None public_key = _env("HERMES_LANGFUSE_PUBLIC_KEY") or _env("LANGFUSE_PUBLIC_KEY") secret_key = _env("HERMES_LANGFUSE_SECRET_KEY") or _env("LANGFUSE_SECRET_KEY") if not (public_key and secret_key): _LANGFUSE_CLIENT = _INIT_FAILED return None # Reject placeholder credentials with a one-shot warning so the # operator sees the misconfiguration instead of silently shipping a # broken observability stack (#23823). The SDK does not validate # keys at construction time — it queues traces in memory and only # discovers the auth failure when the background flush thread tries # to post them, by which point the warning is buried under whatever # else the process is logging. Catch it here, surface it once, and # short-circuit via the same _INIT_FAILED path as the empty case. placeholder_issues = [ msg for msg in ( _validate_langfuse_key("HERMES_LANGFUSE_PUBLIC_KEY", public_key), _validate_langfuse_key("HERMES_LANGFUSE_SECRET_KEY", secret_key), ) if msg ] if placeholder_issues: logger.warning( "Langfuse plugin: credentials look like placeholders, traces will " "NOT be emitted (%s). Set real Langfuse keys (pk-lf-... / sk-lf-...) " "or unset HERMES_LANGFUSE_PUBLIC_KEY / HERMES_LANGFUSE_SECRET_KEY to " "silence this warning.", "; ".join(placeholder_issues), ) _LANGFUSE_CLIENT = _INIT_FAILED return None base_url = _env("HERMES_LANGFUSE_BASE_URL") or _env("LANGFUSE_BASE_URL") or "https://cloud.langfuse.com" environment = _env("HERMES_LANGFUSE_ENV") or _env("LANGFUSE_ENV") release = _env("HERMES_LANGFUSE_RELEASE") or _env("LANGFUSE_RELEASE") sample_rate = _env("HERMES_LANGFUSE_SAMPLE_RATE") kwargs: Dict[str, Any] = { "public_key": public_key, "secret_key": secret_key, "base_url": base_url, } if environment: kwargs["environment"] = environment if release: kwargs["release"] = release if sample_rate: try: kwargs["sample_rate"] = float(sample_rate) except ValueError: logger.warning("Invalid HERMES_LANGFUSE_SAMPLE_RATE=%r", sample_rate) try: _LANGFUSE_CLIENT = Langfuse(**kwargs) except Exception as exc: # pragma: no cover - fail-open logger.warning("Could not initialize Langfuse client: %s", exc) _LANGFUSE_CLIENT = _INIT_FAILED return None # atexit is LIFO: registering AFTER the SDK's constructor (which installs # its own shutdown flush) means our finalizer runs FIRST at exit — root # spans ended there are still picked up by the SDK's exporter. Closes the # short-lived-process gap (kanban workers / hermes chat -q / cron): exit # with tool calls still queued left the root span un-ended → anonymous # trace with no name/session/metadata on the backend. try: import atexit atexit.register(_finalize_all_traces) except Exception: # pragma: no cover - fail-open pass return _LANGFUSE_CLIENT def _scope_prefix(task_id: str, session_id: str) -> str: """The task/session/thread prefix shared by every trace-key shape.""" if task_id: return f"task:{task_id}" if session_id: return f"session:{session_id}" return f"thread:{threading.get_ident()}" def _trace_key( task_id: str, session_id: str, *, turn_id: str = "", api_request_id: str = "", ) -> str: """Build a stable in-process trace scope key for one agent turn. Older Hermes paths only expose ``task_id``/``session_id``. Newer paths pass ``turn_id`` and ``api_request_id`` in LLM/tool hooks; when present, they must scope trace state so concurrent requests sharing one task/session never collide. ``turn_id`` is preferred over ``api_request_id`` so the turn-level ``post_llm_call`` hook (which carries ``turn_id`` but no ``api_request_id``) resolves to the same key as the request-level hooks. """ if turn_id: return f"{_scope_prefix(task_id, session_id)}:turn:{turn_id}" if api_request_id: return f"{_scope_prefix(task_id, session_id)}:api:{api_request_id}" # Legacy shape: a bare ``task_id`` (NOT the ``task:`` prefix) when present, # otherwise the session/thread prefix. Kept distinct for backward # compatibility with keys minted before turn/request scoping existed. if task_id: return task_id return _scope_prefix(task_id, session_id) def _state_for_turn(turn_id: str) -> Optional[str]: """Resolve a live trace key from a turn id alone. The subagent hooks carry ``parent_turn_id`` but no ``task_id``, and ``_scope_prefix`` prefers ``task_id`` when the LLM hooks minted the key — so rebuilding the key here would miss whenever a task id is in play. ``turn_id`` is already unique per turn, so match on its suffix instead. Caller must hold ``_STATE_LOCK``. """ if not turn_id: return None suffix = f":turn:{turn_id}" for key in _TRACE_STATE: if key.endswith(suffix): return key return None def _is_base64_data_uri(value: str) -> bool: prefix = value[:200].lower() return prefix.startswith("data:") and ";base64," in prefix def _redact_data_uri(value: str) -> dict[str, Any]: header = value.split(",", 1)[0] if "," in value else "data:" media_type = header[5:].split(";", 1)[0] if header.startswith("data:") else "" return { "type": "data_uri", "media_type": media_type or None, "omitted": True, "length": len(value), } def _truncate_text(value: str, max_chars: int) -> Any: # Langfuse SDK treats data:*;base64 strings as media and attempts to # decode them. Truncating those strings produces invalid base64 and noisy # "Error parsing base64 data URI" logs. Observability only needs metadata, # not raw image/audio payloads, so redact the whole data URI before it # reaches the SDK. if _is_base64_data_uri(value): return _redact_data_uri(value) # Redact BEFORE truncating so a secret straddling the cut point cannot # leak its prefix. Truncation is a size control, not redaction. if _capture_mode() == "sanitized": value = _redact_secrets(value) if len(value) <= max_chars: return value return value[:max_chars] + f"... [truncated {len(value) - max_chars} chars]" def _maybe_parse_json_string(value: str) -> Any: stripped = value.strip() if len(stripped) < 2 or stripped[0] not in "{[" or stripped[-1] not in "}]": if len(stripped) < 2 or stripped[0] not in "{[": return value try: parsed, idx = json.JSONDecoder().raw_decode(stripped) except Exception: return value if not isinstance(parsed, (dict, list)): return value trailing = stripped[idx:].strip() if not trailing: return parsed hint_key = "_hint" if trailing.startswith("[Hint:") else "_trailing_text" if isinstance(parsed, dict): merged = dict(parsed) key = hint_key if hint_key not in merged else "_trailing_text" merged[key] = trailing return merged return {"data": parsed, hint_key: trailing} def _looks_like_read_file_payload(value: Any) -> bool: if not isinstance(value, dict): return False content = value.get("content") return ( isinstance(content, str) and "total_lines" in value and "file_size" in value and "is_binary" in value and "is_image" in value and not value.get("error") ) def _parse_read_file_lines(content: str) -> list[dict[str, Any]]: if not isinstance(content, str) or not content: return [] lines = [] for raw_line in content.splitlines(): match = _READ_FILE_LINE_RE.match(raw_line) if not match: return [] lines.append({ "line": int(match.group(1)), "text": match.group(2), }) return lines def _build_read_file_preview(lines: list[dict[str, Any]]) -> dict[str, Any]: if len(lines) <= (_READ_FILE_HEAD_LINES + _READ_FILE_TAIL_LINES): return {"lines": lines} return { "head": lines[:_READ_FILE_HEAD_LINES], "tail": lines[-_READ_FILE_TAIL_LINES:], "omitted_line_count": len(lines) - _READ_FILE_HEAD_LINES - _READ_FILE_TAIL_LINES, } def _normalize_read_file_payload(value: dict[str, Any], *, args: Any = None) -> dict[str, Any]: normalized: dict[str, Any] = {} if isinstance(args, dict): path = args.get("path") offset = args.get("offset") limit = args.get("limit") if isinstance(path, str) and path: normalized["path"] = path if isinstance(offset, int): normalized["offset"] = offset if isinstance(limit, int): normalized["limit"] = limit lines = _parse_read_file_lines(value.get("content", "")) if lines: normalized["returned_lines"] = { "start": lines[0]["line"], "end": lines[-1]["line"], "count": len(lines), } normalized["content_preview"] = _build_read_file_preview(lines) elif value.get("content"): normalized["content_preview"] = { "text": value.get("content", ""), } for key in ( "total_lines", "file_size", "truncated", "is_binary", "is_image", "hint", "_warning", "mime_type", "dimensions", "similar_files", "error", ): if key in value: normalized[key] = value[key] base64_content = value.get("base64_content") if isinstance(base64_content, str) and base64_content: normalized["base64_content"] = { "omitted": True, "length": len(base64_content), } return normalized def _normalize_payload(value: Any, *, tool_name: str = "", args: Any = None) -> Any: if _looks_like_read_file_payload(value): return _normalize_read_file_payload( value, args=args if tool_name == "read_file" else None, ) return value def _safe_value(value: Any, *, max_chars: Optional[int] = None, depth: int = 0, parse_json_strings: bool = False) -> Any: max_chars = max_chars if max_chars is not None else int(_env("HERMES_LANGFUSE_MAX_CHARS", "12000") or "12000") if depth > 4: return "" if value is None or isinstance(value, (int, float, bool)): return value if isinstance(value, bytes): return {"type": "bytes", "len": len(value)} if isinstance(value, str): if parse_json_strings: parsed = _maybe_parse_json_string(value) if parsed is not value: return _safe_value(parsed, max_chars=max_chars, depth=depth, parse_json_strings=True) return _truncate_text(value, max_chars) if isinstance(value, dict): normalized = _normalize_payload(value) if normalized is not value: return _safe_value(normalized, max_chars=max_chars, depth=depth, parse_json_strings=parse_json_strings) return { str(k): _safe_value(v, max_chars=max_chars, depth=depth + 1, parse_json_strings=parse_json_strings) for k, v in list(value.items())[:50] } if isinstance(value, (list, tuple, set)): return [ _safe_value(v, max_chars=max_chars, depth=depth + 1, parse_json_strings=parse_json_strings) for v in list(value)[:50] ] if hasattr(value, "__dict__"): return _safe_value(vars(value), max_chars=max_chars, depth=depth + 1, parse_json_strings=parse_json_strings) return _truncate_text(repr(value), max_chars) 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 def _coerce_request_messages( *, request_messages: Any = None, messages: Any = None, conversation_history: Any = None, user_message: Any = None, ) -> list[dict[str, Any]]: 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}] def _serialize_system_prompt(system_prompt: Any) -> Optional[dict[str, Any]]: """Normalize Anthropic/Bedrock ``system`` param or OpenAI-style system content for Langfuse.""" if system_prompt is None: return None if isinstance(system_prompt, str): text = system_prompt.strip() if not text: return None return {"role": "system", "content": _capture_content(text)} if isinstance(system_prompt, list): parts: list[str] = [] for block in system_prompt: if isinstance(block, dict): # Anthropic blocks carry {"type": "text", "text": ...}; Bedrock # Converse system blocks are {"text": ...} with no "type" key. block_type = block.get("type") if block_type == "text" or (block_type is None and "text" in block): piece = block.get("text", "") if isinstance(piece, str) and piece: parts.append(piece) elif isinstance(block, str) and block: parts.append(block) if not parts: return None return {"role": "system", "content": _capture_content("\n\n".join(parts))} return None def _messages_for_langfuse_input( *, request_messages: Any = None, messages: Any = None, conversation_history: Any = None, user_message: Any = None, system_prompt: Any = None, pre_coerced: Any = None, ) -> list[dict[str, Any]]: """Build generation input: include Anthropic ``system`` when split out of ``messages``. Pass ``pre_coerced`` to skip the internal ``_coerce_request_messages`` call when the caller already has the result — avoids double-coercion per hook. """ raw = pre_coerced if pre_coerced is not None else _coerce_request_messages( request_messages=request_messages, messages=messages, conversation_history=conversation_history, user_message=user_message, ) if raw and raw[0].get("role") == "system": return _serialize_messages(raw) system_msg = _serialize_system_prompt(system_prompt) if system_msg is None: return _serialize_messages(raw) return [system_msg, *_serialize_messages(raw)] def _serialize_messages(messages: Any) -> list[dict[str, Any]]: if not isinstance(messages, list): return [] serialized = [] for message in messages[-12:]: if not isinstance(message, dict): continue role = message.get("role") item = { "role": role, "content": _capture_content( message.get("content"), parse_json_strings=(role == "tool"), ), } if role == "tool": if message.get("tool_call_id"): item["tool_call_id"] = message.get("tool_call_id") if message.get("name"): item["name"] = _safe_value(message.get("name")) if message.get("tool_calls"): item["tool_calls"] = _capture_content(message.get("tool_calls"), parse_json_strings=True) serialized.append(item) return serialized def _serialize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]: if not tool_calls: return [] serialized = [] for tool_call in tool_calls: fn = getattr(tool_call, "function", None) name = getattr(fn, "name", None) if fn else None arguments = getattr(fn, "arguments", None) if fn else None safe_arguments = _capture_content(arguments, parse_json_strings=False) serialized.append({ "id": getattr(tool_call, "id", None), "type": getattr(tool_call, "type", None) or "function", "name": name, "arguments": safe_arguments, "function": { "name": name, "arguments": safe_arguments, }, }) return serialized def _extract_assistant_reasoning(message: Any) -> Any: for field in ("reasoning", "reasoning_content", "reasoning_details"): value = getattr(message, field, None) if value is not None: return _capture_content(value) return None def _serialize_assistant_message(message: Any) -> dict[str, Any]: return { "content": _capture_content(getattr(message, "content", None)), "reasoning": _extract_assistant_reasoning(message), "tool_calls": _serialize_tool_calls(getattr(message, "tool_calls", None)), } def _canonical_usage_and_cost( canonical: Any, *, provider: str, model: str, base_url: str, ) -> tuple[dict[str, int], dict[str, float]]: """Translate canonical Hermes usage into Langfuse usage and cost maps.""" usage_details: Dict[str, int] = { "input": canonical.input_tokens, "output": canonical.output_tokens, } if canonical.cache_read_tokens: usage_details["cache_read_input_tokens"] = canonical.cache_read_tokens if canonical.cache_write_tokens: usage_details["cache_creation_input_tokens"] = canonical.cache_write_tokens if canonical.reasoning_tokens: usage_details["reasoning_tokens"] = canonical.reasoning_tokens cost_details: Dict[str, float] = {} try: from agent.usage_pricing import estimate_usage_cost, resolve_billing_route # Subscription-included routes (e.g. openai-codex): Langfuse treats # provided cost_details as authoritative and will not recalculate # from model pricing when explicit zeros are sent. Omit cost_details # entirely so Langfuse falls back to its own estimation. route = resolve_billing_route(model, provider=provider, base_url=base_url) if getattr(route, "billing_mode", "") == "subscription_included": return usage_details, cost_details cost = estimate_usage_cost( model, canonical, provider=provider, base_url=base_url, api_key="", ) except Exception as exc: # pragma: no cover - fail-open _debug(f"usage pricing failed: {exc}") return usage_details, cost_details # A partial component breakdown is not a valid request total. In # particular, cache pricing may be unavailable even when input/output # rates are known. Leave all costs absent so Langfuse cannot mistake a # partial subtotal for the full generation cost. if cost.amount_usd is None: return usage_details, cost_details # Langfuse only derives a total for the built-in input/output cost keys. # Cache/custom keys therefore need an explicit canonical total. Use the # Hermes estimate rather than summing components because it also includes # request-level pricing. Preserve the existing component-only payload for # subscription-included routes; their export policy is handled separately. # A zero estimate is not exported either: a priced model that billed no # tokens writes no per-type keys, and an explicit 0.0 total would be # treated as authoritative by Langfuse instead of left for it to derive. if cost.status != "included" and float(cost.amount_usd) > 0: cost_details["total"] = float(cost.amount_usd) # Langfuse cost_details keys match usage_details keys. Keep the per-type # breakdown for dashboards in addition to the authoritative request total. try: from decimal import Decimal from agent.usage_pricing import get_pricing_entry one_million = Decimal("1000000") entry = get_pricing_entry(model, provider=provider, base_url=base_url) if entry: if entry.input_cost_per_million is not None and canonical.input_tokens: cost_details["input"] = float( Decimal(canonical.input_tokens) * entry.input_cost_per_million / one_million ) if entry.output_cost_per_million is not None and canonical.output_tokens: cost_details["output"] = float( Decimal(canonical.output_tokens) * entry.output_cost_per_million / one_million ) if entry.cache_read_cost_per_million is not None and canonical.cache_read_tokens: cost_details["cache_read_input_tokens"] = float( Decimal(canonical.cache_read_tokens) * entry.cache_read_cost_per_million / one_million ) if entry.cache_write_cost_per_million is not None and canonical.cache_write_tokens: cost_details["cache_creation_input_tokens"] = float( Decimal(canonical.cache_write_tokens) * entry.cache_write_cost_per_million / one_million ) except Exception: # pragma: no cover - canonical total remains usable pass return usage_details, cost_details def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str, base_url: str) -> tuple[dict[str, int], dict[str, float]]: raw_usage = getattr(response, "usage", None) if not raw_usage: return {}, {} try: from agent.usage_pricing import normalize_usage canonical = normalize_usage(raw_usage, provider=provider, api_mode=api_mode) return _canonical_usage_and_cost( canonical, provider=provider, model=model, base_url=base_url, ) except Exception as exc: # pragma: no cover - fail-open _debug(f"usage normalization failed: {exc}") return {}, {} def _start_root_trace(task_key: str, *, task_id: str, session_id: str, platform: str, provider: str, model: str, api_mode: str, messages: Any, client: Langfuse, turn_id: str = "", api_request_id: str = "") -> TraceState: trace_id = client.create_trace_id(seed=f"{session_id or 'sessionless'}::{task_id or task_key}") trace_input = _extract_last_user_message(messages) metadata = { "source": "hermes", "task_id": task_id, "turn_id": turn_id, "api_request_id": api_request_id, "platform": platform, "provider": provider, "model": model, "api_mode": api_mode, "capture_mode": _capture_mode(), } # session_id must be passed in trace_context for Langfuse session grouping. trace_ctx: Dict[str, Any] = {"trace_id": trace_id} if session_id: trace_ctx["session_id"] = session_id if propagate_attributes is not None: try: with propagate_attributes( session_id=session_id or task_key, trace_name="Hermes turn", tags=["hermes", "langfuse"], ): root_ctx = client.start_as_current_observation( trace_context=trace_ctx, name="Hermes turn", as_type="chain", input=trace_input, metadata=metadata, end_on_exit=False, ) root_span = root_ctx.__enter__() except Exception: root_ctx = client.start_as_current_observation( trace_context=trace_ctx, name="Hermes turn", as_type="chain", input=trace_input, metadata=metadata, end_on_exit=False, ) root_span = root_ctx.__enter__() else: root_ctx = client.start_as_current_observation( trace_context=trace_ctx, name="Hermes turn", as_type="chain", input=trace_input, metadata=metadata, end_on_exit=False, ) root_span = root_ctx.__enter__() # SDK v3 uses update_trace() (not set_trace_io). Failures must never block # the rest of the turn — the observation still carries input from start. try: root_span.update_trace(input=trace_input) except Exception as exc: _debug(f"update_trace(input) failed: {exc}") _debug(f"started trace {trace_id} for {task_key}") return TraceState(trace_id=trace_id, root_ctx=root_ctx, root_span=root_span) def _start_child_observation(state: TraceState, *, client: Langfuse, name: str, as_type: str, input_value: Any, metadata: Optional[dict] = None, model: Optional[str] = None, model_parameters: Optional[dict] = None) -> Any: return state.root_span.start_observation( name=name, as_type=as_type, input=input_value, metadata=metadata or {}, model=model, model_parameters=model_parameters, ) def _end_observation(observation: Any, *, output: Any = None, metadata: Optional[dict] = None, usage_details: Optional[dict] = None, cost_details: Optional[dict] = None) -> None: if observation is None: return try: update_kwargs: Dict[str, Any] = {} if output is not None: update_kwargs["output"] = output if metadata: update_kwargs["metadata"] = metadata if usage_details: update_kwargs["usage_details"] = usage_details if cost_details: update_kwargs["cost_details"] = cost_details if update_kwargs: observation.update(**update_kwargs) observation.end() except Exception as exc: # pragma: no cover - fail-open _debug(f"end observation failed: {exc}") def _merge_trace_output(output: Any, state: TraceState) -> Any: if not state.turn_tool_calls: return output merged = dict(output) if isinstance(output, dict) else {"content": output} merged["tool_calls"] = list(state.turn_tool_calls) return merged def _evict_stale_locked() -> None: """Drop least-recently-updated trace state to make room for a new entry. Caller MUST hold ``_STATE_LOCK`` and call this immediately before inserting one new entry. Bounds the leak from turns that never reach ``_finish_trace`` (interrupted / tool-only final step / empty final content), whose unique per-turn key would otherwise linger forever. We evict down to ``_MAX_TRACE_STATE - 1`` so that the about-to-be-added entry leaves the dict at ``_MAX_TRACE_STATE`` — a true ceiling. The evicted entry's root span is ended so it is not left dangling on the Langfuse side. """ over = len(_TRACE_STATE) - (_MAX_TRACE_STATE - 1) if over <= 0: return # Oldest-first by last_updated_at; evict just enough to make room. stale = sorted(_TRACE_STATE.items(), key=lambda kv: kv[1].last_updated_at)[:over] for key, state in stale: _TRACE_STATE.pop(key, None) try: state.root_span.end() if state.root_ctx is not None: try: state.root_ctx.__exit__(None, None, None) except Exception: # pragma: no cover - fail-open pass except Exception as exc: # pragma: no cover - fail-open _debug(f"evict stale trace failed: {exc}") def _finalize_all_traces() -> None: """End every open root span so short-lived processes export complete traces. Gateway turns normally end their root span via ``_finish_trace`` (final assistant message with no tool calls). But short-lived CLI processes — kanban workers, ``hermes chat -q`` one-shots, cron jobs — can exit while the last LLM call still has tool calls queued, leaving the root span un-ended. Ended children DO export via the SDK's own atexit flush, so the backend shows an anonymous trace (no name/session/metadata) whose observations all point at a root that never arrived (observed live 2026-08-08: kanban workers produced 4 such traces in one tick). Registered with ``atexit`` AFTER the Langfuse client is constructed: atexit is LIFO, so this runs BEFORE the SDK's own shutdown hook — spans ended here still get flushed by the SDK's exporter. """ with _STATE_LOCK: states = list(_TRACE_STATE.items()) _TRACE_STATE.clear() for _key, state in states: try: for observation in state.generations.values(): _end_observation(observation) for observation in state.tools.values(): _end_observation(observation) for queue in state.pending_tools_by_name.values(): for observation in queue: _end_observation(observation) for observation in state.subagents.values(): _end_observation(observation) state.root_span.end() # Exit the root observation's context manager so its generator # unwinds now, while opentelemetry.trace.Span is still a real # type — otherwise GC closes it during interpreter teardown and # use_span's isinstance(span, Span) raises TypeError. if state.root_ctx is not None: try: state.root_ctx.__exit__(None, None, None) except Exception: # pragma: no cover - fail-open pass except Exception as exc: # pragma: no cover - fail-open _debug(f"atexit finalize failed for {_key}: {exc}") if states: client = _get_langfuse() if client is not None: try: client.flush() except Exception: pass def _finish_trace(task_key: str, *, output: Any = None) -> None: client = _get_langfuse() if client is None: return with _STATE_LOCK: state = _TRACE_STATE.pop(task_key, None) if state is None: return try: for observation in state.generations.values(): _end_observation(observation) for observation in state.tools.values(): _end_observation(observation) for queue in state.pending_tools_by_name.values(): for observation in queue: _end_observation(observation) final_output = _merge_trace_output(output, state) if final_output is not None: # update_trace sets TRACE-level Input/Output columns in the UI # (SDK v3; set_trace_io was the pre-v3 spelling). Root observation # I/O is set via update(); never let either call prevent end() — # otherwise generations/tools export without a CHAIN root and the # list view looks half-empty. try: state.root_span.update_trace(output=final_output) except Exception as exc: _debug(f"update_trace(output) failed: {exc}") try: state.root_span.update(output=final_output) except Exception as exc: _debug(f"root update(output) failed: {exc}") try: state.root_span.end() except Exception as exc: _debug(f"root end() failed: {exc}") # Properly exit the root context manager so the generator unwinds # now, while opentelemetry.trace.Span is still a real type. Without # this the generator is left suspended; at interpreter teardown the # GC calls .close(), which throws GeneratorExit through use_span # __exit__ -> isinstance(span, Span) -- but Span has been torn down # to None by then, producing the TypeError traceback on quit. if state.root_ctx is not None: try: state.root_ctx.__exit__(None, None, None) except Exception: # pragma: no cover - fail-open pass except Exception as exc: # pragma: no cover - fail-open _debug(f"finish trace failed: {exc}") # Last-chance end so an earlier unexpected error still exports the root. try: state.root_span.end() except Exception: pass finally: try: client.flush() except Exception: pass def _assistant_has_tool_calls(message: Any) -> bool: return bool(getattr(message, "tool_calls", None)) def _request_key(api_call_count: Any) -> str: return str(api_call_count or 0) def on_pre_llm_call(*, task_id: str = "", session_id: str = "", platform: str = "", model: str = "", provider: str = "", base_url: str = "", api_mode: str = "", api_call_count: int = 0, messages: Any = None, turn_type: str = "user", conversation_history: Any = None, user_message: Any = None, turn_id: str = "", api_request_id: str = "", **_: Any) -> None: # Older Hermes branches used pre_llm_call for request-scoped tracing and # passed the actual API messages. Current Hermes also has a turn-scoped # pre_llm_call used for context injection; tracing that hook creates an # extra orphan/root trace before the real request trace. Only trace the # legacy request-shaped call here. if not isinstance(messages, list): return client = _get_langfuse() if client is None: return # messages is a list only for legacy Hermes branches that fired # pre_llm_call with API messages directly. Current Hermes fires # pre_llm_call for context injection (conversation_history/user_message, # no messages list) — tracing that would create orphan traces. task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: state = _start_root_trace( task_key, task_id=task_id, session_id=session_id, platform=platform, provider=provider, model=model, api_mode=api_mode, messages=messages, client=client, turn_id=turn_id, api_request_id=api_request_id, ) _evict_stale_locked() _TRACE_STATE[task_key] = state state.last_updated_at = time.time() def _emit_moa_reference_generations(state: TraceState, *, client: Langfuse, references: Any) -> None: """Record each MoA advisor as its own generation under the turn. MoA returns only the aggregator's response, so without this the whole fan-out collapses into one generation priced at the aggregator's model. Each advisor carries its own model, usage, and dollars — advisors routinely run on a different provider than the aggregator, so their spend cannot be priced at the aggregator's rate. """ if not isinstance(references, list) or not references: return fingerprint = json.dumps( [ [r.get("label"), r.get("model"), (r.get("usage") or {}).get("output_tokens")] for r in references if isinstance(r, dict) ], sort_keys=True, default=str, ) with _STATE_LOCK: if fingerprint in state.moa_emitted: return state.moa_emitted.add(fingerprint) for ref in references: if not isinstance(ref, dict): continue usage = ref.get("usage") or {} usage_details = {} if isinstance(usage, dict): if usage.get("input_tokens"): usage_details["input"] = usage["input_tokens"] if usage.get("output_tokens"): usage_details["output"] = usage["output_tokens"] if usage.get("cache_read_tokens"): usage_details["cache_read_input_tokens"] = usage["cache_read_tokens"] if usage.get("cache_write_tokens"): usage_details["cache_creation_input_tokens"] = usage["cache_write_tokens"] if usage.get("reasoning_tokens"): usage_details["reasoning_tokens"] = usage["reasoning_tokens"] cost_details = {} cost_usd = ref.get("cost_usd") if isinstance(cost_usd, (int, float)): cost_details["total"] = float(cost_usd) label = ref.get("label") or "advisor" metadata = {"moa_role": "reference", "label": label} for key in ("provider", "cost_status", "cost_source", "temperature"): if ref.get(key) is not None: metadata[key] = ref[key] observation = _start_child_observation( state, client=client, name=f"MoA advisor: {label}", as_type="generation", input_value=None, metadata=metadata, model=ref.get("model"), ) _end_observation( observation, output=_capture_content(ref.get("output")), usage_details=usage_details, cost_details=cost_details, metadata=metadata, ) def on_pre_llm_request( *, task_id: str = "", session_id: str = "", platform: str = "", model: str = "", provider: str = "", base_url: str = "", api_mode: str = "", api_call_count: int = 0, request_messages: Any = None, messages: Any = None, turn_type: str = "user", message_count: int = 0, tool_count: int = 0, approx_input_tokens: int = 0, request_char_count: int = 0, max_tokens: Any = None, conversation_history: Any = None, user_message: Any = None, turn_id: str = "", api_request_id: str = "", request: Any = None, system_prompt: Any = None, **_: Any, ) -> None: client = _get_langfuse() if client is None: return # ``model`` is the agent's current attribute at hook time; the request # body carries the model actually being dispatched. They can diverge # (mid-session /model switch propagation, provider fallback, middleware # rewrites) — prefer the wire truth for generation attribution. if isinstance(request, dict): body = request.get("body") if isinstance(body, dict): body_model = body.get("model") if isinstance(body_model, str) and body_model: model = body_model input_messages = _coerce_request_messages( request_messages=request_messages, messages=messages, conversation_history=conversation_history, user_message=user_message, ) langfuse_input = _messages_for_langfuse_input( request_messages=request_messages, messages=messages, conversation_history=conversation_history, user_message=user_message, system_prompt=system_prompt, pre_coerced=input_messages, ) system_chars = 0 if langfuse_input and langfuse_input[0].get("role") == "system": system_chars = len(str(langfuse_input[0].get("content") or "")) task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) req_key = _request_key(api_call_count) with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: state = _start_root_trace( task_key, task_id=task_id, session_id=session_id, platform=platform, provider=provider, model=model, api_mode=api_mode, messages=input_messages, client=client, turn_id=turn_id, api_request_id=api_request_id, ) _evict_stale_locked() _TRACE_STATE[task_key] = state state.last_updated_at = time.time() previous = state.generations.pop(req_key, None) if previous is not None: _end_observation(previous) gen_metadata = { "provider": provider, "platform": platform, "api_mode": api_mode, "base_url": base_url, "message_count": message_count, "approx_input_tokens": approx_input_tokens, } if system_chars: gen_metadata["system_prompt_chars"] = system_chars state.generations[req_key] = _start_child_observation( state, client=client, name=f"LLM call {api_call_count}", as_type="generation", input_value=langfuse_input, metadata=gen_metadata, model=model, model_parameters={"api_mode": api_mode, "provider": provider}, ) def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str = "", base_url: str = "", api_mode: str = "", model: str = "", api_call_count: int = 0, assistant_message: Any = None, response: Any = None, api_duration: float = 0.0, finish_reason: str = "", usage: Any = None, assistant_content_chars: int = 0, assistant_tool_call_count: int = 0, assistant_response: Any = None, turn_id: str = "", api_request_id: str = "", response_model: Any = None, moa_references: Any = None, **_: Any) -> None: client = _get_langfuse() if client is None: return # The provider's response echoes the model that actually served the # request. Prefer it over the agent attribute, which can be stale after # a mid-session model switch or fallback. if isinstance(response_model, str) and response_model: model = response_model task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) req_key = _request_key(api_call_count) with _STATE_LOCK: state = _TRACE_STATE.get(task_key) generation = state.generations.pop(req_key, None) if state else None if state is None or generation is None: return if moa_references: _emit_moa_reference_generations(state, client=client, references=moa_references) # Handle both call patterns: # 1. post_api_request: passes usage (dict), assistant_content_chars, assistant_tool_call_count # 2. post_llm_call: passes assistant_message (object), response (object), assistant_response (str) if assistant_message is not None: output = _serialize_assistant_message(assistant_message) elif assistant_response is not None: # post_llm_call passes assistant_response as a plain string output = {"content": _capture_content(assistant_response), "reasoning": None, "tool_calls": []} else: # post_api_request path — reconstruct from summary kwargs output = { "content": f"[{assistant_content_chars} chars]" if assistant_content_chars else None, "reasoning": None, "tool_calls": [{"id": f"tc_{i}"} for i in range(assistant_tool_call_count)] if assistant_tool_call_count else [], } if output.get("tool_calls"): state.turn_tool_calls.extend(output["tool_calls"]) # Extract usage: prefer a real response object that carries usage, else # fall back to the usage summary dict from post_api_request. # # post_api_request passes `response` as a SANITIZED dict (no ``.usage`` # attribute) alongside a separate `usage` summary dict. Gating on # ``response is not None`` here took the response-object path on that dict, # where ``getattr(response, "usage", None)`` is always None — so usage and # cost were silently dropped for every gateway turn. Gate on a real # ``.usage`` attribute instead so the usage-dict fallback below is reached. if getattr(response, "usage", None) is not None: usage_details, cost_details = _usage_and_cost( response, provider=provider, api_mode=api_mode, model=model, base_url=base_url, ) elif isinstance(usage, dict) and usage: # post_api_request passes a pre-built CanonicalUsage summary dict. _input = usage.get("input_tokens", 0) _output = usage.get("output_tokens", 0) or usage.get("completion_tokens", 0) _cache_read = usage.get("cache_read_tokens", 0) _cache_write = usage.get("cache_write_tokens", 0) _reasoning = usage.get("reasoning_tokens", 0) try: from agent.usage_pricing import CanonicalUsage _cu = CanonicalUsage( input_tokens=_input, output_tokens=_output, cache_read_tokens=_cache_read, cache_write_tokens=_cache_write, reasoning_tokens=_reasoning, request_count=usage.get("request_count", 1), ) usage_details, cost_details = _canonical_usage_and_cost( _cu, provider=provider, model=model, base_url=base_url, ) except Exception: usage_details, cost_details = {}, {} else: usage_details, cost_details = {}, {} tool_count = len(output.get("tool_calls", [])) or assistant_tool_call_count gen_metadata: Dict[str, Any] = {"tool_call_count": tool_count} if api_duration and api_duration > 0: gen_metadata["api_duration_s"] = round(api_duration, 3) if finish_reason: gen_metadata["finish_reason"] = finish_reason _end_observation( generation, output=output, usage_details=usage_details, cost_details=cost_details, metadata=gen_metadata, ) has_tools = _assistant_has_tool_calls(assistant_message) if assistant_message else (assistant_tool_call_count > 0) has_content = bool(output.get("content")) if not has_tools and has_content: _finish_trace(task_key, output=output) def on_pre_tool_call(*, tool_name: str = "", args: Any = None, task_id: str = "", session_id: str = "", tool_call_id: str = "", turn_id: str = "", api_request_id: str = "", **_: Any) -> None: client = _get_langfuse() if client is None: return task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: return observation = _start_child_observation( state, client=client, name=f"Tool: {tool_name}", as_type="tool", input_value=_capture_content(args), metadata={"tool_name": tool_name, "tool_call_id": tool_call_id}, ) if tool_call_id: state.tools[tool_call_id] = observation else: state.pending_tools_by_name.setdefault(tool_name, []).append(observation) def on_post_tool_call(*, tool_name: str = "", args: Any = None, result: Any = None, task_id: str = "", session_id: str = "", tool_call_id: str = "", turn_id: str = "", api_request_id: str = "", **_: Any) -> None: task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) observation = None with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: return if tool_call_id: observation = state.tools.pop(tool_call_id, 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) if observation is None: return if _capture_mode() == "metadata": safe_result_value = _describe_content(result) else: if isinstance(result, str): result_value = _maybe_parse_json_string(result) else: result_value = result result_value = _normalize_payload(result_value, tool_name=tool_name, args=args) safe_result_value = _safe_value(result_value, parse_json_strings=True) # Backfill so the generation's tool_call record carries the result alongside arguments. 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 _end_observation( observation, output=safe_result_value, metadata={"tool_name": tool_name, "args": _capture_content(args, parse_json_strings=True)}, ) def on_api_request_error(*, task_id: str = "", session_id: str = "", provider: str = "", model: str = "", api_mode: str = "", api_call_count: int = 0, api_duration: float = 0.0, status_code: Any = None, retry_count: Any = None, max_retries: Any = None, retryable: Any = None, reason: Any = None, error: Any = None, turn_id: str = "", api_request_id: str = "", **_: Any) -> None: """Close the open generation for a failed API request. Without this, a failed request leaves its generation open until trace eviction — the failure is invisible in Langfuse and the turn appears to hang. Marks the generation with ERROR level and the error summary. If the request was not retryable (or retries are exhausted), the turn is finished too, since the agent loop is about to unwind. """ client = _get_langfuse() if client is None: return task_key = _trace_key( task_id, session_id, turn_id=turn_id, api_request_id=api_request_id, ) req_key = _request_key(api_call_count) with _STATE_LOCK: state = _TRACE_STATE.get(task_key) generation = state.generations.pop(req_key, None) if state else None if state is None: return error_type = "" error_message = "" if isinstance(error, dict): error_type = str(error.get("type") or "") error_message = str(error.get("message") or "") error_metadata: Dict[str, Any] = { "error": True, "error_type": error_type, # Error messages can embed request fragments (URLs w/ keys, prompt # echoes) — run them through the capture pipeline like content. "error_message": _capture_content(error_message), } if status_code is not None: error_metadata["status_code"] = status_code if retry_count is not None: error_metadata["retry_count"] = retry_count if max_retries is not None: error_metadata["max_retries"] = max_retries if retryable is not None: error_metadata["retryable"] = retryable if reason: error_metadata["reason"] = str(reason) if api_duration and api_duration > 0: error_metadata["api_duration_s"] = round(api_duration, 3) if generation is not None: try: generation.update( level="ERROR", status_message=(error_type or "api_request_error")[:200], ) except Exception as exc: # pragma: no cover - fail-open _debug(f"error-level update failed: {exc}") _end_observation(generation, metadata=error_metadata) # A retryable failure will be followed by another pre_api_request on the # same trace; keep the turn open. A terminal failure ends the turn. if retryable is False: _finish_trace(task_key, output={"error": error_metadata}) else: state.last_updated_at = time.time() def on_session_finalize(*, session_id: str = "", reason: str = "", **_: Any) -> None: """True session-end boundary: close any traces still open and flush. A turn that ended on a tool-only or empty final response never reaches ``_finish_trace``; without this hook its root span dangles until state eviction and queued events can be lost on process exit. """ # Only act on an already-constructed client — do NOT lazily initialize # one at finalize time; if init never happened there are no traces. client = _LANGFUSE_CLIENT if client is None or client is _INIT_FAILED or not hasattr(client, "flush"): return # Close every trace belonging to this session (or all, when no # session_id is provided — process-level finalization). Trace keys carry # the session as either "session:" (no task) or "task:" (gateway # sets task_id == session_id), plus the legacy bare-task_id shape — match # on the id in any segment. if session_id: fragments = (f"session:{session_id}", f"task:{session_id}") with _STATE_LOCK: keys = [ k for k in _TRACE_STATE if k == session_id or any(f in k for f in fragments) ] else: with _STATE_LOCK: keys = list(_TRACE_STATE) for key in keys: _finish_trace(key) try: client.flush() except Exception as exc: # pragma: no cover - fail-open _debug(f"finalize flush failed: {exc}") # Explicitly shut down the Langfuse client — but only at a true # process-exit boundary. on_session_finalize also fires on /new, /reset # (reason "session_boundary"/"new_session") and gateway session expiry # ("session_expired"), where the process lives on and the cached client # must keep exporting for subsequent sessions. The shutdown matters at # exit because the Langfuse SDK's own atexit handler runs during # interpreter finalization — by then module globals (notably # opentelemetry.trace.Span) may already be torn down to None, and the # SDK's span-finalization path (use_span -> isinstance(span, Span)) # raises "TypeError: isinstance() arg 2 must be a type" which surfaces # as a noisy "Exception ignored in: " traceback on quit. # Calling shutdown() here flushes pending spans and joins the background # export threads while all modules are intact; the SDK's atexit handler # then becomes a no-op. if reason == "shutdown": shutdown = getattr(client, "shutdown", None) if callable(shutdown): try: shutdown() except Exception as exc: # pragma: no cover - fail-open _debug(f"langfuse shutdown failed: {exc}") def on_subagent_start(*, parent_session_id: Any = None, parent_turn_id: str = "", parent_subagent_id: Any = None, child_session_id: Any = None, child_subagent_id: Any = None, child_role: str = "", child_goal: Any = None, **_: Any) -> None: client = _get_langfuse() if client is None or not child_session_id: return with _STATE_LOCK: key = _state_for_turn(parent_turn_id) state = _TRACE_STATE.get(key) if key else None if state is None: return metadata = { "child_session_id": child_session_id, "child_subagent_id": child_subagent_id, "child_role": child_role, } if parent_subagent_id: metadata["parent_subagent_id"] = parent_subagent_id state.subagents[str(child_session_id)] = _start_child_observation( state, client=client, name=f"Subagent: {child_role or 'delegate'}", as_type="span", input_value=_capture_content(child_goal), metadata=metadata, ) def on_subagent_stop(*, parent_session_id: Any = None, parent_turn_id: str = "", child_session_id: Any = None, child_role: str = "", child_summary: Any = None, child_status: Any = None, tool_call_history: Any = None, duration_ms: Any = None, **_: Any) -> None: if not child_session_id: return with _STATE_LOCK: key = _state_for_turn(parent_turn_id) state = _TRACE_STATE.get(key) if key else None if state is None: return observation = state.subagents.pop(str(child_session_id), None) if observation is None: return metadata: Dict[str, Any] = {"child_role": child_role} if child_status: metadata["status"] = child_status if duration_ms: metadata["duration_ms"] = duration_ms if isinstance(tool_call_history, list): metadata["tool_call_count"] = len(tool_call_history) metadata["tool_calls"] = _capture_content(tool_call_history) _end_observation( observation, output=_capture_content(child_summary), metadata=metadata, ) def register(ctx) -> None: # Register for both hook name variants so the plugin works across # Hermes versions. pre_api_request / post_api_request fire per API # call (preferred); pre_llm_call / post_llm_call fire 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)