# NousResearch/hermes-agent · Langfuse observability missing LLM and tool I/O ## Open now - Nothing open on GitHub. Multiple bug reports and corresponding fixes address Langfuse tracing not capturing LLM input/output and tool outputs. The issues and pull requests appear to target the same observability gap in the Langfuse integration. ## TL;DR - 4 member(s): 0 open issue(s), 2 closed issue(s), 0 open PR(s), 1 merged PR(s), 1 PR(s) closed without merging. - A fix was merged to main in PR #26320. Verify it addresses your case. - **Recommended:** read PR #26320 (and any competing PRs below) before filing. Confirm the merged change covers your symptom. ## Maintainer resolution verdict - Internal-tracker verdict over this cluster's issue members: `partially_resolved` · deciding PR(s): `pr-26320`. This is distinct from the fix status above. Weigh it yourself. ## How this cluster was grouped - Link evidence (`STRUCTURAL_PARTIAL`): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text. - Members outside that structural core may belong to a different underlying problem. - Pairwise link density `0.39`: the share of member pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. `1.00` means every pair is linked. ## Closed / duplicate issues - #17712: \[Bug\]: Langfuse plugin cannot see llm call input and output - #22342: \[Bug\]: Langfuse traces missing LLM input/output and tool output ## Prior & competing PRs - **#26320: fix(langfuse): complete observability fix — trace I/O, tool outputs, placeholder credentials (closes #22342, #22763)**. merged to main · fix tier `verified` - maintainer direction: category `merged`; implication `superseded` - files: `plugins/observability/langfuse/__init__.py`, `run_agent.py`, `tests/plugins/test_langfuse_plugin.py`, `tests/run_agent/test_run_agent.py` - fix hunks: - `plugins/observability/langfuse/__init__.py` ```diff @@ -47,6 +47,7 @@ class TraceState: 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) last_updated_at: float = field(default_factory=time.time) ``` ```diff @@ -58,6 +59,17 @@ class TraceState: _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() ``` ```diff @@ -82,10 +94,49 @@ def _debug(message: str) -> None: # 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. Cleared by reset_cache_for_tests(). +# 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. ``` ```diff @@ -111,6 +162,33 @@ def _get_langfuse() -> Optional[Langfuse]: _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") ``` ```diff @@ -328,6 +406,21 @@ def _extract_last_user_message(messages: Any) -> Any: 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_messages(messages: Any) -> list[dict[str, Any]]: if not isinstance(messages, list): return [] ``` ```diff @@ -343,8 +436,11 @@ def _serialize_messages(messages: Any) -> list[dict[str, Any]]: parse_json_strings=(role == "tool"), ), } - if role == "tool" and message.get("tool_call_id"): - item["tool_call_id"] = message.get("tool_call_id") + 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"] = _safe_value(message.get("tool_calls"), parse_json_strings=True) serialized.append(item) ``` ```diff @@ -359,15 +455,16 @@ def _serialize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]: fn = getattr(tool_call, "function", None) name = getattr(fn, "name", None) if fn else None arguments = getattr(fn, "arguments", None) if fn else None - if isinstance(arguments, str): - try: - arguments = json.loads(arguments) - except Exception: - pass + safe_arguments = _safe_value(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_value(arguments, parse_json_strings=True), + "arguments": safe_arguments, + "function": { + "name": name, + "arguments": safe_arguments, + }, }) return serialized ``` ```diff @@ -564,6 +661,9 @@ def _finish_trace(task_key: str, *, output: Any = None) -> None: _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: state.root_span.set_trace_io(output=final_output) ``` ```diff @@ -636,6 +736,7 @@ def on_pre_llm_request( 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, ``` ```diff @@ -643,12 +744,21 @@ def on_pre_llm_request( approx_input_tokens: int = 0, request_char_count: int = 0, max_tokens: Any = None, + conversation_history: Any = None, + user_message: Any = None, **_: Any, ) -> None: client = _get_langfuse() if client is None: return + input_messages = _coerce_request_messages( + request_messages=request_messages, + messages=messages, + conversation_history=conversation_history, + user_message=user_message, + ) + task_key = _trace_key(task_id, session_id) req_key = _request_key(api_call_count) ``` ```diff @@ -663,7 +773,7 @@ def on_pre_llm_request( provider=provider, model=model, api_mode=api_mode, - messages=messages, + messages=input_messages, client=client, ) _TRACE_STATE[task_key] = state ``` ```diff @@ -676,7 +786,7 @@ def on_pre_llm_request( client=client, name=f"LLM call {api_call_count}", as_type="generation", - input_value=_serialize_messages(messages), + input_value=_serialize_messages(input_messages), metadata={ "provider": provider, "platform": platform, ``` ```diff @@ -815,13 +925,12 @@ def on_pre_tool_call(*, tool_name: str = "", args: Any = None, task_id: str = "" return task_key = _trace_key(task_id, session_id) - tool_key = tool_call_id or f"{tool_name}:{time.time_ns()}" with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: return - state.tools[tool_key] = _start_child_observation( + observation = _start_child_observation( state, client=client, name=f"Tool: {tool_name}", ``` ```diff @@ -829,22 +938,29 @@ def on_pre_tool_call(*, tool_name: str = "", args: Any = None, task_id: str = "" input_value=_safe_value(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 = "", **_: Any) -> None: task_key = _trace_key(task_id, session_id) - tool_key = tool_call_id or "" observation = None with _STATE_LOCK: state = _TRACE_STATE.get(task_key) if state is None: return - if tool_key: - observation = state.tools.pop(tool_key, None) - elif state.tools: - _, observation = state.tools.popitem() + 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 ``` ```diff @@ -854,10 +970,24 @@ def on_post_tool_call(*, tool_name: str = "", args: Any = None, result: Any = No 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_value(result_value, parse_json_strings=True), + output=safe_result_value, metadata={"tool_name": tool_name, "args": _safe_value(args, parse_json_strings=True)}, ) ``` - `run_agent.py` ```diff @@ -12668,16 +12668,30 @@ def run_conversation( try: from hermes_cli.plugins import invoke_hook as _invoke_hook + request_messages = api_kwargs.get("messages") + if not isinstance(request_messages, list): + request_messages = api_kwargs.get("input") + if not isinstance(request_messages, list): + request_messages = api_messages + # Shallow-copy the outer list so plugins that retain the + # reference for async snapshotting don't observe later + # mutations of api_messages. The inner dicts are not + # mutated by the agent loop, so a shallow copy is + # sufficient; a deepcopy would walk every tool result + # and base64 image on every API call. _invoke_hook( "pre_api_request", task_id=effective_task_id, session_id=self.session_id or "", + user_message=original_user_message, + conversation_history=list(messages), platform=self.platform or "", model=self.model, provider=self.provider, base_url=self.base_url, api_mode=self.api_mode, api_call_count=api_call_count, + request_messages=list(request_messages) if isinstance(request_messages, list) else [], message_count=len(api_messages), tool_count=len(self.tools or []), approx_input_tokens=[redacted:secret-assignment] ``` ```diff @@ -14582,7 +14596,9 @@ def _stop_spinner(): finish_reason=finish_reason, message_count=len(api_messages), response_model=getattr(response, "model", None), + response=response, usage=self._usage_summary_for_api_request_hook(response), + assistant_message=assistant_message, assistant_content_chars=len(_assistant_text), assistant_tool_call_count=len(_assistant_tool_calls), ) ``` - `tests/plugins/test_langfuse_plugin.py` ```diff @@ -2,6 +2,7 @@ from __future__ import annotations import importlib +import logging import sys from pathlib import Path ``` ```diff @@ -164,7 +165,542 @@ def test_hooks_noop_without_client(self, monkeypatch): # 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, messages=[]) + 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") + + +# --------------------------------------------------------------------------- +# Placeholder-credential guard (#23823). +# +# Regression coverage for the silent-failure bug: when an operator leaves +# HERMES_LANGFUSE_PUBLIC_KEY / SECRET_KEY at a template value like +# "placeholder", "test-key", or "your-langfuse-key", the SDK accepts the +# credentials at construction time (it does no server-side validation +# eagerly) but drops every trace at flush time, with no signal in the +# Hermes logs. The fix in `_get_langfuse()` validates the documented +# `pk-lf-` / `sk-lf-` prefix Langfuse always issues, surfaces a one-shot +# warning naming the offending env var(s), and short-circuits via the +# same `_INIT_FAILED` path used for missing credentials so subsequent +# hook invocations don't re-log. +# --------------------------------------------------------------------------- + + +class _FakeLangfuse: + """Stand-in for the real :class:`langfuse.Langfuse` so tests don't + need the optional ``langfuse`` SDK installed. The plugin's runtime + gate refuses to proceed past ``if Langfuse is None`` when the SDK + is missing, which would short-circuit before the placeholder check + can fire. Patching ``plugin.Langfuse`` with this class lets the + placeholder validator exercise its full code path.""" + + instances: list["_FakeLangfuse"] = [] + + def __init__(self, **kwargs): + self.kwargs = kwargs + _FakeLangfuse.instances.append(self) + + +class TestPlaceholderKeyDetection: + 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: + # Pretend the SDK is installed so `_get_langfuse()` actually + # reaches the placeholder check. Real SDK calls are never + # made because the placeholder/missing-credentials paths + # return before constructing a client. + _FakeLangfuse.instances.clear() + monkeypatch.setattr(mod, "Langfuse", _FakeLangfuse, raising=False) + return mod + + @staticmethod + def _clear_env(monkeypatch): + for k in ( + "HERMES_LANGFUSE_PUBLIC_KEY", "HERMES_LANGFUSE_SECRET_KEY", + "LANGFUSE_PUBLIC_KEY", "LANGFUSE_SECRET_KEY", + ): + monkeypatch.delenv(k, raising=False) + + # -- helper unit tests (no SDK stub needed: these don't go through + # _get_langfuse, they exercise the pure-Python helpers directly) ------ + + def test_redact_key_preview_empty(self, monkeypatch): + self._clear_env(monkeypatch) + plugin = self._fresh_plugin() + assert plugin._redact_key_preview("") == "" + + def test_redact_key_preview_short_value_echoed(self, monkeypatch): + """Short placeholder strings are echoed in full so the operator + can see exactly which template they forgot to replace.""" + self._clear_env(monkeypatch) + plugin = self._fresh_plugin() + assert plugin._redact_key_preview("placeholder") == "'placeholder'" + assert plugin._redact_key_preview("test-key") == "'test-key'" + + def test_redact_key_preview_long_value_truncated(self, monkeypatch): + """If an operator pasted a real secret into the wrong env var the + preview must NOT echo it in full — only the leading 6 chars.""" + self._clear_env(monkeypatch) + plugin = self._fresh_plugin() + result = plugin._redact_key_preview("[redacted:openai-key]") + assert "abcdefghij" not in result + assert result.startswith("'sk-lf-") + assert result.endswith("...'") + + 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", "[redacted:openai-key]" + ) 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" + ) + assert msg is not None + assert "HERMES_LANGFUSE_PUBLIC_KEY" in msg + assert "pk-lf-" in msg + + 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 + + # -- end-to-end _get_langfuse() behaviour -------------------------------- + # These tests pass `monkeypatch` to _fresh_plugin() so the helper can + # stub out `Langfuse` (the optional SDK). Without that, every call + # short-circuits at `if Langfuse is None` before reaching the + # placeholder validator — masking the very behaviour we're testing. + + def test_placeholder_public_key_warns_and_skips(self, monkeypatch, caplog): + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "placeholder") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "[redacted:openai-key]") + plugin = self._fresh_plugin(monkeypatch) + with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): + assert plugin._get_langfuse() is None + text = caplog.text + assert "HERMES_LANGFUSE_PUBLIC_KEY" in text + assert "'placeholder'" in text + assert "pk-lf-" in text + # The valid secret value must NOT appear (the var NAME does, in + # the "or unset ..." hint, but the value preview shouldn't). + assert "'sk-lf-" not in text + # Never constructed the SDK client — short-circuited before that. + assert _FakeLangfuse.instances == [] + + def test_placeholder_secret_key_warns_and_skips(self, monkeypatch, caplog): + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "test-key") + plugin = self._fresh_plugin(monkeypatch) + with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): + assert plugin._get_langfuse() is None + text = caplog.text + assert "HERMES_LANGFUSE_SECRET_KEY" in text + assert "'test-key'" in text + assert "sk-lf-" in text + # The valid public value must NOT appear. + assert "'pk-lf-" not in text + assert _FakeLangfuse.instances == [] + + def test_both_placeholders_one_warning_with_both_keys(self, monkeypatch, caplog): + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "placeholder") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "placeholder") + 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] + assert len(warnings) == 1, ( + f"Expected a single combined warning; got {len(warnings)}:\n" + + "\n".join(r.getMessage() for r in warnings) + ) + text = warnings[0].getMessage() + assert "HERMES_LANGFUSE_PUBLIC_KEY" in text + assert "HERMES_LANGFUSE_SECRET_KEY" in text + + def test_repeated_calls_do_not_re_warn(self, monkeypatch, caplog): + """The cached ``_INIT_FAILED`` sentinel must short-circuit + subsequent calls so each hook invocation isn't a fresh log + line — otherwise a busy gateway will spam the operator's + terminal.""" + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "placeholder") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "placeholder") + plugin = self._fresh_plugin(monkeypatch) + 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] + assert len(warnings) == 1, ( + f"Warning fired {len(warnings)} times across 15 calls; " + "expected 1 (cached via _INIT_FAILED)" + ) + + @pytest.mark.parametrize("placeholder", [ + "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", "[redacted:openai-key]") + 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 + + 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 + whichever value ``_get_langfuse()`` actually consumed.""" + self._clear_env(monkeypatch) + monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "placeholder") + monkeypatch.setenv("LANGFUSE_SECRET_KEY", "[redacted:openai-key]") + plugin = self._fresh_plugin(monkeypatch) + with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): + assert plugin._get_langfuse() is None + # Warning names the canonical user-facing env var (the bare + # LANGFUSE_PUBLIC_KEY is a backwards-compat alias for the + # HERMES_-prefixed one — operators set the HERMES_-prefixed one). + assert "HERMES_LANGFUSE_PUBLIC_KEY" in caplog.text + assert "'placeholder'" in caplog.text + + def test_missing_credentials_still_skip_silently(self, monkeypatch, caplog): + """Missing-creds is the documented opt-out path (operator hasn't + configured the plugin yet) — it must remain SILENT. Regression + guard against the placeholder validator accidentally running on + empty values and re-introducing log noise for unconfigured + installs.""" + self._clear_env(monkeypatch) + 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] + assert warnings == [] + + def test_sdk_not_installed_still_skips_silently(self, monkeypatch, caplog): + """If the langfuse SDK isn't installed at all, the placeholder + check should never run — there's nothing the operator can do + about a credential mismatch when the package is missing, and + re-warning here would dilute the actually-actionable SDK-missing + signal upstream. The ``Langfuse is None`` guard at the top of + ``_get_langfuse`` already handles this; this test pins that + behaviour.""" + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "placeholder") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "placeholder") + # NO monkeypatch on Langfuse here — falls back to whatever the + # plugin imported at module load (None if SDK absent). + plugin = self._fresh_plugin() + 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] + assert warnings == [] + + 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 + placeholder warning didn't fire AND that the client was actually + constructed — the latter is the success signal the bug report + wanted.""" + self._clear_env(monkeypatch) + monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-real-public-xyz") + monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "[redacted:openai-key]") + plugin = self._fresh_plugin(monkeypatch) + with caplog.at_level(logging.WARNING, logger=self.LOGGER_NAME): + client = plugin._get_langfuse() + assert isinstance(client, _FakeLangfuse) + assert client.kwargs["public_key"] == "pk-lf-real-public-xyz" + assert client.kwargs["secret_key"] == "[redacted:openai-key]" + assert "placeholders" not in caplog.text.lower(), ( + f"Valid Langfuse keys tripped the placeholder guard: {caplog.text!r}" + ) + + +class TestRequestMessageCoercion: + 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") + + assert mod._coerce_request_messages( + request_messages=[{"role": "system", "content": "s"}], + messages=[{"role": "user", "content": "m"}], + conversation_history=[{"role": "user", "content": "h"}], + user_message="u", + ) == [{"role": "system", "content": "s"}] + assert mod._coerce_request_messages( + messages=[{"role": "user", "content": "m"}], + conversation_history=[{"role": "user", "content": "h"}], + user_message="u", + ) == [{"role": "user", "content": "m"}] + assert mod._coerce_request_messages( + 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"}] + + +class TestToolCallOutputBackfill: + def test_post_tool_call_backfills_matching_turn_tool_call_output(self, monkeypatch): + sys.modules.pop("plugins.observability.langfuse", None) + mod = importlib.import_module("plugins.observability.langfuse") + + observation = object() + state = mod.TraceState(trace_id="trace-1", root_ctx=None, root_span=None) + state.tools["call-1"] = observation + state.turn_tool_calls.append({ + "id": "call-1", + "type": "function", + "name": "web_extract", + "arguments": '{"urls": ["https://example.com"]}', + "function": { + "name": "web_extract", + "arguments": '{"urls": ["https://example.com"]}', + }, + }) + + task_key = mod._trace_key("task-1", "session-1") + monkeypatch.setitem(mod._TRACE_STATE, task_key, state) + + ended = {} + + def fake_end_observation(obs, *, output=None, metadata=None, usage_details=None, cost_details=None): + ended["observation"] = obs + ended["output"] = output + ended["metadata"] = metadata + + monkeypatch.setattr(mod, "_end_observation", fake_end_observation) + + mod.on_post_tool_call( + tool_name="web_extract", + args={"urls": ["https://example.com"]}, + result='{"results": [{"url": "https://example.com", "content": "Example Domain"}]}', + task_id="task-1", + session_id="session-1", + tool_call_id="call-1", + ) + + assert ended["observation"] is observation + assert state.turn_tool_calls[0]["output"] == ended["output"] + assert state.turn_tool_calls[0]["function"]["output"] == ended["output"] + assert state.turn_tool_calls[0]["output"] == { + "results": [{"url": "https://example.com", "content": "Example Domain"}] + } + + 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}, + }] + + def test_serialize_tool_calls_emits_openai_style_function_shape(self): + sys.modules.pop("plugins.observability.langfuse", None) + mod = importlib.import_module("plugins.observability.langfuse") + + class _Fn: + name = "web_extract" + arguments = '{"urls": ["https://example.com"]}' + + class _ToolCall: + id = "call-1" + type = "function" + function = _Fn() + + 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: + """Tests for pre/post tool_call observation matching when tool_call_id is absent.""" + + def _make_mod(self): + sys.modules.pop("plugins.observability.langfuse", None) + return importlib.import_module("plugins.observability.langfuse") + + def test_empty_tool_call_id_single_tool_sets_output(self, monkeypatch): + mod = self._make_mod() + obs = object() + state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) + state.pending_tools_by_name.setdefault("my_tool", []).append(obs) + + task_key = mod._trace_key("task-1", "sess-1") + monkeypatch.setitem(mod._TRACE_STATE, task_key, state) + + ended = {} + + def fake_end(o, *, output=None, metadata=None, **kw): + ended["obs"] = o + ended["output"] = output + + monkeypatch.setattr(mod, "_end_observation", fake_end) + + mod.on_post_tool_call( + tool_name="my_tool", + args={}, + result='{"ok": true}', + task_id="task-1", + session_id="sess-1", + tool_call_id="", + ) + + assert ended["obs"] is obs + 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): + """Two queued observations are consumed in FIFO order so the first + post hook gets the first observation's output, not the second. + + Sequential-on-one-thread coverage; the real concurrent case is + guarded by ``_STATE_LOCK`` around every read-modify-write on + ``pending_tools_by_name`` and is exercised in + ``test_threaded_post_calls_preserve_fifo_under_lock`` below. + """ + mod = self._make_mod() + obs_a, obs_b = object(), object() + state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) + state.pending_tools_by_name["web_extract"] = [obs_a, obs_b] + + task_key = mod._trace_key("task-1", "sess-1") + monkeypatch.setitem(mod._TRACE_STATE, task_key, state) + + calls = [] + + def fake_end(o, *, output=None, metadata=None, **kw): + calls.append((o, output)) + + 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="", + ) + mod.on_post_tool_call( + 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"}) + assert calls[1] == (obs_b, {"val": "b"}) + assert state.pending_tools_by_name.get("web_extract") is None + + def test_threaded_post_calls_preserve_fifo_under_lock(self, monkeypatch): + """The actual concurrency contract: when 8 threads race to drain + the pending queue, no observation is consumed twice and none is + lost. Validates ``_STATE_LOCK`` discipline, not Python list + semantics.""" + import threading + + mod = self._make_mod() + n = 8 + observations = [object() for _ in range(n)] + state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) + state.pending_tools_by_name["web_extract"] = list(observations) + + task_key = mod._trace_key("task-thr", "sess-thr") + monkeypatch.setitem(mod._TRACE_STATE, task_key, state) + + recorded: list = [] + lock = threading.Lock() + + def fake_end(o, *, output=None, metadata=None, **kw): + with lock: + recorded.append(o) + + monkeypatch.setattr(mod, "_end_observation", fake_end) + + barrier = threading.Barrier(n) + + 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="", + ) + + threads = [threading.Thread(target=worker) for _ in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + # Every observation was consumed exactly once; queue is empty. + assert len(recorded) == n + assert set(map(id, recorded)) == set(map(id, observations)) + assert state.pending_tools_by_name.get("web_extract") is None + + def test_explicit_tool_call_id_uses_tools_dict(self, monkeypatch): + """When tool_call_id is present, pending_tools_by_name is not touched.""" + mod = self._make_mod() + obs = object() + state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) + state.tools["call-99"] = obs + + task_key = mod._trace_key("task-1", "sess-1") + monkeypatch.setitem(mod._TRACE_STATE, task_key, state) + + ended = {} + + def fake_end(o, *, output=None, metadata=None, **kw): + ended["obs"] = o + ended["output"] = output + + 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", + ) + + assert ended["obs"] is obs + assert ended["output"] == {"status": "done"} + assert not state.tools + ``` - `tests/run_agent/test_run_agent.py` ```diff @@ -2524,8 +2524,9 @@ def _record_hook(name, **kwargs): assert [call["api_call_count"] for call in pre_request_calls] == [1, 2] assert [call["api_call_count"] for call in post_request_calls] == [1, 2] assert all(call["session_id"] == agent.session_id for call in pre_request_calls) - assert all("message_count" in c and "messages" not in c for c in pre_request_calls) - assert all("usage" in c and "response" not in c for c in post_request_calls) + assert all("message_count" in c and isinstance(c.get("request_messages"), list) for c in pre_request_calls) + assert any(msg.get("role") == "user" and msg.get("content") == "search something" for msg in pre_request_calls[0]["request_messages"]) + assert all("usage" in c and "response" in c and "assistant_message" in c for c in post_request_calls) def test_content_with_tool_calls_stays_silent_for_non_cli_quiet_mode(self, agent): self._setup_agent(agent) ``` - **#22345: fix(langfuse): fixes langfuse observability - llm input/output and tool output missing**. closed without merge - maintainer direction: category `superseded_by_pr`; implication `superseded`; refs `#26320`, `#23831`, `#12928` ## Fix status - Merged to main in PR #26320 (cited above). No release data is tracked. Verify downstream. - Fix last verified (merged) `2026-05-15`.