From 6b01f2e452f5d8bdf563b6f201d9f2cc4eb3061c Mon Sep 17 00:00:00 2001 From: imax Date: Wed, 8 Jul 2026 14:56:18 +0000 Subject: [PATCH] fix(observability/langfuse): surface placeholder & runtime ingestion failures (#60961) - Extend _validate_langfuse_key to catch placeholder fragments even when the key carries the correct pk-lf-/sk-lf- prefix (e.g. sk-lf-..., sk-lf-***, pk-lf-placeholder). Substring fragments stay conservative so real keys like sk-lf-...-abc are not flagged; whole-value templates (sk-lf-..., ***) are matched by exact equality. - Wire Langfuse's on_unexpected_error callback so the first runtime ingestion failure (bad creds / missing project) is logged as an ERROR naming the cause, instead of being swallowed silently. The warning fires once per process to avoid log floods. Fixes #60961 --- plugins/observability/langfuse/__init__.py | 403 +++++++++++++++++---- tests/plugins/test_langfuse_plugin.py | 388 +++++++++++++++----- 2 files changed, 635 insertions(+), 156 deletions(-) diff --git a/plugins/observability/langfuse/__init__.py b/plugins/observability/langfuse/__init__.py index 31904d47e3052..063e6b0f9b13b 100644 --- a/plugins/observability/langfuse/__init__.py +++ b/plugins/observability/langfuse/__init__.py @@ -20,6 +20,7 @@ HERMES_LANGFUSE_MAX_CHARS - max chars per field (default: 12000) HERMES_LANGFUSE_DEBUG - set to "true" for verbose logging """ + from __future__ import annotations import json @@ -79,6 +80,46 @@ class TraceState: "HERMES_LANGFUSE_SECRET_KEY": "sk-lf-", } +# Substring fragments that mark an *unmodified template* credential even when +# it happens to start with the right prefix (or no prefix is registered). +# These are caught so operators get a clear warning instead of silent zero-trace +# behavior. See issue #60961. Keep this list to substrings that can NEVER +# appear in a real key (template leftovers like "placeholder", "your-", "unset"). +_LANGFUSE_PLACEHOLDER_FRAGMENTS: tuple[str, ...] = ( + "placeholder", + "change-me", + "changeit", + "your-", + "your_", + "example", + "sample", + "test", + "demo", + "dummy", + "fake", + "xxxx", + "xxxxxx", + "todo", + "fixme", + "unset", + "redacted", +) + +# Whole-value placeholders from the issue report (e.g. ``sk-lf-...``) — matched +# on exact equality only, so a real key like ``sk-lf-...-abc`` is NOT flagged. +_LANGFUSE_PLACEHOLDER_EXACT: tuple[str, ...] = ( + "sk-lf-...", + "pk-lf-...", + "sk-...", + "pk-...", + "sk-lf-***", + "pk-lf-***", + "***", + "*", + "xxx", + "xxxx", +) + def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() @@ -129,21 +170,37 @@ def _redact_key_preview(value: str) -> str: 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 + Two independent checks run: + + 1. Prefix check — Langfuse always issues keys with a documented prefix + (``pk-lf-`` / ``sk-lf-``). A value without that prefix is rejected. + 2. Placeholder check — even a value with the right prefix can still be an + unmodified template (e.g. ``sk-lf-...``, ``pk-lf-placeholder``). Any + known placeholder fragment causes rejection. + + Returns ``None`` when the value looks like a real key, or when no prefix is + registered for``env_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)" - ) + if not value.startswith(expected): + return f"{env_name}={_redact_key_preview(value)} (expected {expected!r} prefix)" + lowered = value.lower() + for fragment in _LANGFUSE_PLACEHOLDER_FRAGMENTS: + if fragment in lowered: + return ( + f"{env_name}={_redact_key_preview(value)} " + f"looks like a placeholder (contains {fragment!r})" + ) + if lowered in _LANGFUSE_PLACEHOLDER_EXACT: + return ( + f"{env_name}={_redact_key_preview(value)} " + f"looks like a placeholder (template value)" + ) + return None def _get_langfuse() -> Optional[Langfuse]: @@ -198,7 +255,11 @@ def _get_langfuse() -> Optional[Langfuse]: _LANGFUSE_CLIENT = _INIT_FAILED return None - base_url = _env("HERMES_LANGFUSE_BASE_URL") or _env("LANGFUSE_BASE_URL") or "https://cloud.langfuse.com" + 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") @@ -218,8 +279,38 @@ def _get_langfuse() -> Optional[Langfuse]: except ValueError: logger.warning("Invalid HERMES_LANGFUSE_SAMPLE_RATE=%r", sample_rate) + # Surface runtime ingestion failures instead of swallowing them. The SDK + # validates nothing eagerly and only discovers bad credentials when the + # background flush thread posts traces, so we hook its error callback and + # log the first failure clearly (#60961). Subsequent failures are muted + # to avoid log floods; the operator must restart to re-arm the warning. + runtime_error_logged = {"fired": False} + + def _on_unexpected_error(exc: Exception) -> None: + if runtime_error_logged["fired"]: + return + runtime_error_logged["fired"] = True + logger.error( + "Langfuse plugin: ingestion failed (%s). Traces may not be " + "reaching Langfuse — check HERMES_LANGFUSE_PUBLIC_KEY / " + "HERMES_LANGFUSE_SECRET_KEY and that the project exists at %s.", + exc, + base_url, + ) + try: + kwargs["on_unexpected_error"] = _on_unexpected_error _LANGFUSE_CLIENT = Langfuse(**kwargs) + except TypeError: + # Older SDK versions don't accept on_unexpected_error; retry without it + # (we still get the fail-open construction guard below). + kwargs.pop("on_unexpected_error", None) + 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 except Exception as exc: # pragma: no cover - fail-open logger.warning("Could not initialize Langfuse client: %s", exc) _LANGFUSE_CLIENT = _INIT_FAILED @@ -357,11 +448,15 @@ def _build_read_file_preview(lines: list[dict[str, Any]]) -> dict[str, Any]: 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, + "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]: +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") @@ -422,9 +517,18 @@ def _normalize_payload(value: Any, *, tool_name: str = "", args: Any = 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") +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)): @@ -435,23 +539,45 @@ def _safe_value(value: Any, *, max_chars: Optional[int] = None, depth: int = 0, 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 _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 _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) + 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) + _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 _safe_value( + vars(value), + max_chars=max_chars, + depth=depth + 1, + parse_json_strings=parse_json_strings, + ) return _truncate_text(repr(value), max_chars) @@ -503,7 +629,9 @@ def _serialize_messages(messages: Any) -> list[dict[str, Any]]: if message.get("name"): item["name"] = _safe_value(message.get("name")) if message.get("tool_calls"): - item["tool_calls"] = _safe_value(message.get("tool_calls"), parse_json_strings=True) + item["tool_calls"] = _safe_value( + message.get("tool_calls"), parse_json_strings=True + ) serialized.append(item) return serialized @@ -538,7 +666,9 @@ def _serialize_assistant_message(message: Any) -> dict[str, Any]: } -def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str, base_url: str) -> tuple[dict[str, int], dict[str, float]]: +def _usage_and_cost( + response: Any, *, provider: str, api_mode: str, model: str, base_url: str +) -> tuple[dict[str, int], dict[str, float]]: usage_details: Dict[str, int] = {} cost_details: Dict[str, float] = {} raw_usage = getattr(response, "usage", None) @@ -579,17 +709,46 @@ def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str, try: from agent.usage_pricing import get_pricing_entry from decimal import Decimal + _ONE_M = 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_M) - 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_M) - 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_M) - 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_M) + 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_M + ) + 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_M + ) + 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_M + ) + 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_M + ) else: cost_details["total"] = float(cost.amount_usd) except Exception: @@ -600,10 +759,23 @@ def _usage_and_cost(response: Any, *, provider: str, api_mode: str, model: str, return usage_details, cost_details -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}") +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", @@ -667,9 +839,17 @@ def _start_root_trace(task_key: str, *, task_id: str, session_id: str, platform: 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: +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, @@ -680,8 +860,14 @@ def _start_child_observation(state: TraceState, *, client: Langfuse, name: str, ) -def _end_observation(observation: Any, *, output: Any = None, metadata: Optional[dict] = None, - usage_details: Optional[dict] = None, cost_details: Optional[dict] = None) -> None: +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: @@ -774,11 +960,24 @@ 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: +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 @@ -905,14 +1104,27 @@ def on_pre_llm_request( ) -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 = "", - **_: Any) -> None: +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 = "", + **_: Any, +) -> None: client = _get_langfuse() if client is None: return @@ -938,13 +1150,21 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str = output = _serialize_assistant_message(assistant_message) elif assistant_response is not None: # post_llm_call passes assistant_response as a plain string - output = {"content": _safe_value(assistant_response), "reasoning": None, "tool_calls": []} + output = { + "content": _safe_value(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, + "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 [], + "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"): @@ -990,8 +1210,13 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str = cost_details = {} # Estimate per-type cost from the summary if possible try: - from agent.usage_pricing import CanonicalUsage, estimate_usage_cost, get_pricing_entry + from agent.usage_pricing import ( + CanonicalUsage, + estimate_usage_cost, + get_pricing_entry, + ) from decimal import Decimal + _ONE_M = Decimal("1000000") _cu = CanonicalUsage( input_tokens=_input, @@ -1003,15 +1228,29 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str = entry = get_pricing_entry(model, provider=provider, base_url=base_url) if entry: if entry.input_cost_per_million is not None and _input: - cost_details["input"] = float(Decimal(_input) * entry.input_cost_per_million / _ONE_M) + cost_details["input"] = float( + Decimal(_input) * entry.input_cost_per_million / _ONE_M + ) if entry.output_cost_per_million is not None and _output: - cost_details["output"] = float(Decimal(_output) * entry.output_cost_per_million / _ONE_M) + cost_details["output"] = float( + Decimal(_output) * entry.output_cost_per_million / _ONE_M + ) if entry.cache_read_cost_per_million is not None and _cache_read: - cost_details["cache_read_input_tokens"] = float(Decimal(_cache_read) * entry.cache_read_cost_per_million / _ONE_M) + cost_details["cache_read_input_tokens"] = float( + Decimal(_cache_read) + * entry.cache_read_cost_per_million + / _ONE_M + ) if entry.cache_write_cost_per_million is not None and _cache_write: - cost_details["cache_creation_input_tokens"] = float(Decimal(_cache_write) * entry.cache_write_cost_per_million / _ONE_M) + cost_details["cache_creation_input_tokens"] = float( + Decimal(_cache_write) + * entry.cache_write_cost_per_million + / _ONE_M + ) else: - _cost = estimate_usage_cost(model, _cu, provider=provider, base_url=base_url, api_key="") + _cost = estimate_usage_cost( + model, _cu, provider=provider, base_url=base_url, api_key="" + ) if _cost.amount_usd is not None: cost_details["total"] = float(_cost.amount_usd) except Exception: @@ -1033,15 +1272,27 @@ def on_post_llm_call(*, task_id: str = "", session_id: str = "", provider: str = metadata=gen_metadata, ) - has_tools = _assistant_has_tool_calls(assistant_message) if assistant_message else (assistant_tool_call_count > 0) + 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: +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 @@ -1071,9 +1322,18 @@ def on_pre_tool_call(*, tool_name: str = "", args: Any = None, task_id: str = "" 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: +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, @@ -1121,7 +1381,10 @@ def on_post_tool_call(*, tool_name: str = "", args: Any = None, result: Any = No _end_observation( observation, output=safe_result_value, - metadata={"tool_name": tool_name, "args": _safe_value(args, parse_json_strings=True)}, + metadata={ + "tool_name": tool_name, + "args": _safe_value(args, parse_json_strings=True), + }, ) diff --git a/tests/plugins/test_langfuse_plugin.py b/tests/plugins/test_langfuse_plugin.py index dd58149eba2e5..61e02bd924a08 100644 --- a/tests/plugins/test_langfuse_plugin.py +++ b/tests/plugins/test_langfuse_plugin.py @@ -1,4 +1,5 @@ """Tests for the bundled observability/langfuse plugin.""" + from __future__ import annotations import importlib @@ -19,6 +20,7 @@ # Manifest + layout # --------------------------------------------------------------------------- + class TestManifest: def test_plugin_directory_exists(self): assert PLUGIN_DIR.is_dir() @@ -31,9 +33,12 @@ def test_manifest_fields(self): assert data["version"] # All six hooks the plugin implements. assert set(data["hooks"]) == { - "pre_api_request", "post_api_request", - "pre_llm_call", "post_llm_call", - "pre_tool_call", "post_tool_call", + "pre_api_request", + "post_api_request", + "pre_llm_call", + "post_llm_call", + "pre_tool_call", + "post_tool_call", } # Required env vars are the user-facing HERMES_ prefixed keys. assert "HERMES_LANGFUSE_PUBLIC_KEY" in data["requires_env"] @@ -46,6 +51,7 @@ def test_manifest_fields(self): # load_config() gate or making the plugin auto-load. # --------------------------------------------------------------------------- + class TestDiscovery: def test_plugin_is_discovered_as_standalone_opt_in(self, tmp_path, monkeypatch): """Scanner should find the plugin but NOT load it by default.""" @@ -74,6 +80,7 @@ def test_plugin_is_discovered_as_standalone_opt_in(self, tmp_path, monkeypatch): # per-hook load_config() design. # --------------------------------------------------------------------------- + class TestRuntimeGate: def _fresh_plugin(self): """Import the plugin module fresh (clears any cached client).""" @@ -83,8 +90,10 @@ def _fresh_plugin(self): def test_get_langfuse_returns_none_without_credentials(self, monkeypatch): for k in ( - "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", - "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", ): monkeypatch.delenv(k, raising=False) @@ -94,8 +103,10 @@ def test_get_langfuse_returns_none_without_credentials(self, monkeypatch): def test_get_langfuse_caches_failure_no_config_load(self, monkeypatch): """A miss must be cached — no per-hook config.yaml reads, no env re-reads.""" for k in ( - "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", - "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", ): monkeypatch.delenv(k, raising=False) @@ -107,6 +118,7 @@ def test_get_langfuse_caches_failure_no_config_load(self, monkeypatch): # Now block os.environ.get — a correctly-cached plugin must not # touch env again. import os + called = {"n": 0} real_get = os.environ.get @@ -128,8 +140,10 @@ def tracking_get(key, default=None): def test_get_langfuse_does_not_import_hermes_config(self, monkeypatch): """The plugin must not re-read config.yaml per hook.""" for k in ( - "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", - "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", ): monkeypatch.delenv(k, raising=False) @@ -150,31 +164,44 @@ def test_get_langfuse_does_not_import_hermes_config(self, monkeypatch): # Hooks are inert when the client is unavailable. # --------------------------------------------------------------------------- + class TestHooksInert: def test_hooks_noop_without_client(self, monkeypatch): """All 6 hooks must return without raising when _get_langfuse() is None.""" for k in ( - "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", - "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", ): monkeypatch.delenv(k, raising=False) sys.modules.pop("plugins.observability.langfuse", None) import importlib + mod = importlib.import_module("plugins.observability.langfuse") # Each hook should just return; no exceptions. - mod.on_pre_llm_call(task_id="t", session_id="s", messages=[{"role": "user", "content": "hi"}]) - mod.on_pre_llm_request(task_id="t", session_id="s", api_call_count=1, request_messages=[]) + mod.on_pre_llm_call( + task_id="t", session_id="s", messages=[{"role": "user", "content": "hi"}] + ) + mod.on_pre_llm_request( + task_id="t", session_id="s", api_call_count=1, request_messages=[] + ) mod.on_post_llm_call(task_id="t", session_id="s", api_call_count=1) - mod.on_pre_tool_call(tool_name="read_file", args={}, task_id="t", session_id="s") - mod.on_post_tool_call(tool_name="read_file", args={}, result="ok", task_id="t", session_id="s") + mod.on_pre_tool_call( + tool_name="read_file", args={}, task_id="t", session_id="s" + ) + mod.on_post_tool_call( + tool_name="read_file", args={}, result="ok", task_id="t", session_id="s" + ) class TestPayloadSanitization: def test_safe_value_redacts_base64_data_uri_instead_of_truncating(self): sys.modules.pop("plugins.observability.langfuse", None) import importlib + mod = importlib.import_module("plugins.observability.langfuse") payload = "data:image/png;base64," + ("a" * 20000) @@ -190,11 +217,15 @@ def test_safe_value_redacts_base64_data_uri_instead_of_truncating(self): def test_serialize_messages_redacts_data_uri_parts(self): sys.modules.pop("plugins.observability.langfuse", None) import importlib + mod = importlib.import_module("plugins.observability.langfuse") payload = "data:image/jpeg;base64," + ("b" * 20000) serialized = mod._serialize_messages([ - {"role": "user", "content": [{"type": "image_url", "image_url": {"url": payload}}]} + { + "role": "user", + "content": [{"type": "image_url", "image_url": {"url": payload}}], + } ]) assert serialized[0]["content"][0]["image_url"]["url"] == { @@ -372,8 +403,12 @@ def test_pre_and_post_hooks_share_one_key_within_a_turn(self, monkeypatch): turn_id = "S:T:turnX" api_request_id = f"{turn_id}:api:1" - k_pre_api = mod._trace_key("T", "S", turn_id=turn_id, api_request_id=api_request_id) - k_post_api = mod._trace_key("T", "S", turn_id=turn_id, api_request_id=api_request_id) + k_pre_api = mod._trace_key( + "T", "S", turn_id=turn_id, api_request_id=api_request_id + ) + k_post_api = mod._trace_key( + "T", "S", turn_id=turn_id, api_request_id=api_request_id + ) k_post_turn = mod._trace_key("T", "S", turn_id=turn_id, api_request_id="") assert k_pre_api == k_post_api == k_post_turn @@ -409,7 +444,7 @@ def test_trace_key_strings_unchanged_by_refactor(self): assert tk("", "s", turn_id="u") == "session:s:turn:u" assert tk("t", "s", api_request_id="r") == "task:t:api:r" assert tk("", "s", api_request_id="r") == "session:s:api:r" - assert tk("t", "s") == "t" # legacy: bare task_id + assert tk("t", "s") == "t" # legacy: bare task_id assert tk("", "s") == "session:s" # turn_id wins over api_request_id when both are present. assert tk("t", "s", turn_id="u", api_request_id="r") == "task:t:turn:u" @@ -465,8 +500,10 @@ def _fresh_plugin(self, monkeypatch=None): @staticmethod def _clear_env(monkeypatch): for k in ( - "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", - "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", ): monkeypatch.delenv(k, raising=False) @@ -499,19 +536,23 @@ def test_redact_key_preview_long_value_truncated(self, monkeypatch): def test_validate_langfuse_key_accepts_documented_prefix(self, monkeypatch): self._clear_env(monkeypatch) plugin = self._fresh_plugin() - assert plugin._validate_langfuse_key( - "HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz" - ) is None - assert plugin._validate_langfuse_key( - "HERMES_LANGFUSE_SECRET_KEY", "sk-lf-real-secret-xyz" - ) is None + assert ( + plugin._validate_langfuse_key( + "HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz" + ) + is None + ) + assert ( + plugin._validate_langfuse_key( + "HERMES_LANGFUSE_SECRET_KEY", "sk-lf-real-secret-xyz" + ) + is None + ) def test_validate_langfuse_key_rejects_wrong_prefix(self, monkeypatch): self._clear_env(monkeypatch) plugin = self._fresh_plugin() - msg = plugin._validate_langfuse_key( - "HERMES_LANGFUSE_PUBLIC_KEY", "placeholder" - ) + msg = plugin._validate_langfuse_key("HERMES_LANGFUSE_PUBLIC_KEY", "placeholder") assert msg is not None assert "HERMES_LANGFUSE_PUBLIC_KEY" in msg assert "pk-lf-" in msg @@ -520,7 +561,10 @@ def test_validate_langfuse_key_unknown_name_passes(self, monkeypatch): """Defensive: an env var with no registered prefix is trusted.""" self._clear_env(monkeypatch) plugin = self._fresh_plugin() - assert plugin._validate_langfuse_key("HERMES_LANGFUSE_BASE_URL", "anything") is None + assert ( + plugin._validate_langfuse_key("HERMES_LANGFUSE_BASE_URL", "anything") + is None + ) # -- end-to-end _get_langfuse() behaviour -------------------------------- # These tests pass `monkeypatch` to _fresh_plugin() so the helper can @@ -567,8 +611,11 @@ def test_both_placeholders_one_warning_with_both_keys(self, monkeypatch, caplog) plugin = self._fresh_plugin(monkeypatch) with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): assert plugin._get_langfuse() is None - warnings = [r for r in caplog.records if r.levelname == "WARNING" - and r.name == self.LOGGER_NAME] + warnings = [ + r + for r in caplog.records + if r.levelname == "WARNING" and r.name == self.LOGGER_NAME + ] assert len(warnings) == 1, ( f"Expected a single combined warning; got {len(warnings)}:\n" + "\n".join(r.getMessage() for r in warnings) @@ -589,34 +636,158 @@ def test_repeated_calls_do_not_re_warn(self, monkeypatch, caplog): with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): for _ in range(15): assert plugin._get_langfuse() is None - warnings = [r for r in caplog.records if r.levelname == "WARNING" - and r.name == self.LOGGER_NAME] + warnings = [ + r + for r in caplog.records + if r.levelname == "WARNING" and r.name == self.LOGGER_NAME + ] assert len(warnings) == 1, ( f"Warning fired {len(warnings)} times across 15 calls; " "expected 1 (cached via _INIT_FAILED)" ) - @pytest.mark.parametrize("placeholder", [ + @pytest.mark.parametrize( "placeholder", - "test-key", - "your-langfuse-key", - "change-me", - "xxx", - "dummy-key-here", - "", - "REPLACE_ME", - ]) + [ + "placeholder", + "test-key", + "your-langfuse-key", + "change-me", + "xxx", + "dummy-key-here", + "", + "REPLACE_ME", + ], + ) def test_common_placeholders_detected(self, monkeypatch, caplog, placeholder): """A grab-bag of values that real-world ``.env.example`` templates use as stand-ins. Any of them in either key must trip the guard.""" self._clear_env(monkeypatch) monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", placeholder) - monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "sk-lf-real-secret-xyz") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "«redacted:sk-…»") plugin = self._fresh_plugin(monkeypatch) with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): assert plugin._get_langfuse() is None assert "HERMES_LANGFUSE_PUBLIC_KEY" in caplog.text + +class TestRuntimeIngestionError: + """Issue #60961 (part 2): a structurally-valid key whose credentials are + rejected by the Langfuse API at flush time must surface a clear error on + the first failure, not be swallowed silently.""" + + LOGGER_NAME = "plugins.observability.langfuse" + + def _fresh_plugin(self, monkeypatch): + mod_name = "plugins.observability.langfuse" + sys.modules.pop(mod_name, None) + mod = importlib.import_module(mod_name) + monkeypatch.setattr(mod, "Langfuse", _FakeLangfuse, raising=False) + _FakeLangfuse.instances.clear() + return mod + + def _clear_env(self, monkeypatch): + for k in ( + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", + ): + monkeypatch.delenv(k, raising=False) + + def test_on_unexpected_error_logs_first_failure(self, monkeypatch, caplog): + """If the SDK's on_unexpected_error callback fires, the plugin must log + an ERROR naming the likely cause (bad creds / missing project).""" + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "sk-lf-real-secret-xyz") + plugin = self._fresh_plugin(monkeypatch) + + # Build a real client so _get_langfuse() reaches the SDK-construction + # path and wires up on_unexpected_error. + client = plugin._get_langfuse() + assert client is not None, "client should construct with valid-looking keys" + + # The callback is stored on the constructed Langfuse instance kwargs. + cb = client.kwargs.get("on_unexpected_error") + assert cb is not None, "on_unexpected_error callback must be wired" + + with caplog.at_level(logging.ERROR, logger=self.LOGGER_NAME): + cb(RuntimeError("401 Unauthorized")) + assert any( + r.levelname == "ERROR" and "ingestion failed" in r.getMessage() + for r in caplog.records + ) + + def test_on_unexpected_error_fires_once(self, monkeypatch, caplog): + """The warning must not spam the log on every failed flush.""" + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "sk-lf-real-secret-xyz") + plugin = self._fresh_plugin(monkeypatch) + client = plugin._get_langfuse() + cb = client.kwargs.get("on_unexpected_error") + with caplog.at_level(logging.ERROR, logger=self.LOGGER_NAME): + for _ in range(10): + cb(RuntimeError("401")) + errors = [ + r + for r in caplog.records + if r.levelname == "ERROR" and "ingestion failed" in r.getMessage() + ] + assert len(errors) == 1, f"expected 1 error log, got {len(errors)}" + + +class TestIssue60961PlaceholderFragments: + """Issue #60961: placeholder fragments must be caught even when the key + carries the correct prefix (e.g. ``sk-lf-...`` from the report).""" + + LOGGER_NAME = "plugins.observability.langfuse" + + def _fresh_plugin(self, monkeypatch=None): + mod_name = "plugins.observability.langfuse" + sys.modules.pop(mod_name, None) + mod = importlib.import_module(mod_name) + if monkeypatch is not None: + _FakeLangfuse.instances.clear() + monkeypatch.setattr(mod, "Langfuse", _FakeLangfuse, raising=False) + return mod + + def _clear_env(self, monkeypatch): + for k in ( + "HERMES_LANGFUSE_PUBLIC_KEY", + "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", + "LANGFUSE_SECRET_KEY", + ): + monkeypatch.delenv(k, raising=False) + + @pytest.mark.parametrize( + "value", + [ + "pk-lf-...", + "sk-lf-...", + "pk-lf-placeholder-xyz", + "sk-lf-***", + "pk-lf-unset", + "sk-lf-change-me", + "pk-lf-example-key", + ], + ) + def test_validate_langfuse_key_rejects_placeholder_fragments( + self, monkeypatch, value + ): + self._clear_env(monkeypatch) + plugin = self._fresh_plugin(monkeypatch) + env_name = ( + "HERMES_LANGFUSE_PUBLIC_KEY" + if value.startswith("pk-lf-") + else "HERMES_LANGFUSE_SECRET_KEY" + ) + msg = plugin._validate_langfuse_key(env_name, value) + assert msg is not None, f"{value!r} should be flagged as placeholder" + assert "placeholder" in msg + def test_legacy_LANGFUSE_PUBLIC_KEY_also_validated(self, monkeypatch, caplog): """The plugin reads both the canonical HERMES_-prefixed env var and the legacy bare ``LANGFUSE_PUBLIC_KEY``. The validator must run on @@ -643,8 +814,11 @@ def test_missing_credentials_still_skip_silently(self, monkeypatch, caplog): plugin = self._fresh_plugin(monkeypatch) with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): assert plugin._get_langfuse() is None - warnings = [r for r in caplog.records if r.levelname == "WARNING" - and r.name == self.LOGGER_NAME] + warnings = [ + r + for r in caplog.records + if r.levelname == "WARNING" and r.name == self.LOGGER_NAME + ] assert warnings == [] def test_sdk_not_installed_still_skips_silently(self, monkeypatch, caplog): @@ -664,11 +838,16 @@ def test_sdk_not_installed_still_skips_silently(self, monkeypatch, caplog): monkeypatch.setattr(plugin, "Langfuse", None, raising=False) with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): assert plugin._get_langfuse() is None - warnings = [r for r in caplog.records if r.levelname == "WARNING" - and r.name == self.LOGGER_NAME] + warnings = [ + r + for r in caplog.records + if r.levelname == "WARNING" and r.name == self.LOGGER_NAME + ] assert warnings == [] - def test_valid_prefixes_do_not_trigger_placeholder_warning(self, monkeypatch, caplog): + def test_valid_prefixes_do_not_trigger_placeholder_warning( + self, monkeypatch, caplog + ): """Real Langfuse keys (``pk-lf-…`` / ``sk-lf-…``) must pass the guard and proceed to SDK init. We stub the SDK constructor with a recording fake so the assertion can confirm BOTH that the @@ -690,7 +869,9 @@ def test_valid_prefixes_do_not_trigger_placeholder_warning(self, monkeypatch, ca class TestRequestMessageCoercion: - def test_prefers_request_messages_then_messages_then_history_then_user_message(self): + def test_prefers_request_messages_then_messages_then_history_then_user_message( + self, + ): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") @@ -709,7 +890,9 @@ def test_prefers_request_messages_then_messages_then_history_then_user_message(s conversation_history=[{"role": "user", "content": "h"}], user_message="u", ) == [{"role": "user", "content": "h"}] - assert mod._coerce_request_messages(user_message="u") == [{"role": "user", "content": "u"}] + assert mod._coerce_request_messages(user_message="u") == [ + {"role": "user", "content": "u"} + ] class TestToolCallOutputBackfill: @@ -736,7 +919,9 @@ def test_post_tool_call_backfills_matching_turn_tool_call_output(self, monkeypat ended = {} - def fake_end_observation(obs, *, output=None, metadata=None, usage_details=None, cost_details=None): + def fake_end_observation( + obs, *, output=None, metadata=None, usage_details=None, cost_details=None + ): ended["observation"] = obs ended["output"] = output ended["metadata"] = metadata @@ -763,19 +948,23 @@ def test_serialize_messages_keeps_tool_name_and_call_id(self): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") - messages = [{ - "role": "tool", - "name": "web_extract", - "tool_call_id": "call-1", - "content": '{"ok": true}', - }] - - assert mod._serialize_messages(messages) == [{ - "role": "tool", - "name": "web_extract", - "tool_call_id": "call-1", - "content": {"ok": True}, - }] + messages = [ + { + "role": "tool", + "name": "web_extract", + "tool_call_id": "call-1", + "content": '{"ok": true}', + } + ] + + assert mod._serialize_messages(messages) == [ + { + "role": "tool", + "name": "web_extract", + "tool_call_id": "call-1", + "content": {"ok": True}, + } + ] def test_serialize_tool_calls_emits_openai_style_function_shape(self): sys.modules.pop("plugins.observability.langfuse", None) @@ -790,16 +979,18 @@ class _ToolCall: type = "function" function = _Fn() - assert mod._serialize_tool_calls([_ToolCall()]) == [{ - "id": "call-1", - "type": "function", - "name": "web_extract", - "arguments": '{"urls": ["https://example.com"]}', - "function": { + assert mod._serialize_tool_calls([_ToolCall()]) == [ + { + "id": "call-1", + "type": "function", "name": "web_extract", "arguments": '{"urls": ["https://example.com"]}', - }, - }] + "function": { + "name": "web_extract", + "arguments": '{"urls": ["https://example.com"]}', + }, + } + ] class TestToolObservationKeying: @@ -839,7 +1030,9 @@ def fake_end(o, *, output=None, metadata=None, **kw): assert ended["output"] == {"ok": True} assert state.pending_tools_by_name.get("my_tool") is None - def test_empty_tool_call_id_observations_are_fifo_within_tool_name(self, monkeypatch): + def test_empty_tool_call_id_observations_are_fifo_within_tool_name( + self, monkeypatch + ): """Two queued observations are consumed in FIFO order so the first post hook gets the first observation's output, not the second. @@ -864,12 +1057,20 @@ def fake_end(o, *, output=None, metadata=None, **kw): monkeypatch.setattr(mod, "_end_observation", fake_end) mod.on_post_tool_call( - tool_name="web_extract", args={}, result='{"val": "a"}', - task_id="task-1", session_id="sess-1", tool_call_id="", + tool_name="web_extract", + args={}, + result='{"val": "a"}', + task_id="task-1", + session_id="sess-1", + tool_call_id="", ) mod.on_post_tool_call( - tool_name="web_extract", args={}, result='{"val": "b"}', - task_id="task-1", session_id="sess-1", tool_call_id="", + tool_name="web_extract", + args={}, + result='{"val": "b"}', + task_id="task-1", + session_id="sess-1", + tool_call_id="", ) assert calls[0] == (obs_a, {"val": "a"}) @@ -906,8 +1107,12 @@ def fake_end(o, *, output=None, metadata=None, **kw): def worker(): barrier.wait() mod.on_post_tool_call( - tool_name="web_extract", args={}, result='{"ok": true}', - task_id="task-thr", session_id="sess-thr", tool_call_id="", + tool_name="web_extract", + args={}, + result='{"ok": true}', + task_id="task-thr", + session_id="sess-thr", + tool_call_id="", ) threads = [threading.Thread(target=worker) for _ in range(n)] @@ -940,8 +1145,12 @@ def fake_end(o, *, output=None, metadata=None, **kw): monkeypatch.setattr(mod, "_end_observation", fake_end) mod.on_post_tool_call( - tool_name="my_tool", args={}, result='{"status": "done"}', - task_id="task-1", session_id="sess-1", tool_call_id="call-99", + tool_name="my_tool", + args={}, + result='{"status": "done"}', + task_id="task-1", + session_id="sess-1", + tool_call_id="call-99", ) assert ended["obs"] is obs @@ -961,10 +1170,14 @@ def _setup(self, mod, monkeypatch): observation = object() state = mod.TraceState(trace_id="trace-1", root_ctx=None, root_span=None) state.generations[mod._request_key(1)] = observation - monkeypatch.setitem(mod._TRACE_STATE, mod._trace_key("task-1", "session-1"), state) + monkeypatch.setitem( + mod._TRACE_STATE, mod._trace_key("task-1", "session-1"), state + ) captured = {} - def fake_end_observation(obs, *, output=None, metadata=None, usage_details=None, cost_details=None): + def fake_end_observation( + obs, *, output=None, metadata=None, usage_details=None, cost_details=None + ): captured["usage_details"] = usage_details monkeypatch.setattr(mod, "_end_observation", fake_end_observation) @@ -981,7 +1194,10 @@ def test_sanitized_dict_response_uses_usage_dict(self, monkeypatch): session_id="session-1", api_call_count=1, model="gemini-3-flash-preview", - response={"model": "gemini-3-flash-preview", "usage": {"input_tokens": 100, "output_tokens": 20}}, + response={ + "model": "gemini-3-flash-preview", + "usage": {"input_tokens": 100, "output_tokens": 20}, + }, usage={"input_tokens": 100, "output_tokens": 20}, assistant_content_chars=42, )