"""Tests for the bundled observability/langfuse plugin.""" from __future__ import annotations import importlib import logging import sys from decimal import Decimal from pathlib import Path from types import SimpleNamespace import pytest # --------------------------------------------------------------------------- # Plugin discovery: langfuse is opt-in (not loaded unless explicitly enabled). # This guards against someone accidentally re-introducing a per-hook # 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.""" from hermes_cli import plugins as plugins_mod # Isolated HERMES_HOME so we don't read the developer's config.yaml. home = tmp_path / ".hermes" home.mkdir() monkeypatch.setenv("HERMES_HOME", str(home)) monkeypatch.setattr(Path, "home", lambda: tmp_path) manager = plugins_mod.PluginManager() manager.discover_and_load() # observability/langfuse appears in the plugin registry … loaded = manager._plugins.get("observability/langfuse") assert loaded is not None, "plugin not discovered" # … but is not loaded (opt-in default → no config.yaml means nothing enabled) assert loaded.enabled is False assert "not enabled" in (loaded.error or "").lower() # --------------------------------------------------------------------------- # Runtime gate: _get_langfuse() returns None and caches _INIT_FAILED when # credentials are missing. Guards against regressing toward the rejected # per-hook load_config() design. # --------------------------------------------------------------------------- class TestRuntimeGate: def _fresh_plugin(self): """Import the plugin module fresh (clears any cached client).""" mod_name = "plugins.observability.langfuse" sys.modules.pop(mod_name, None) return importlib.import_module(mod_name) 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", ): monkeypatch.delenv(k, raising=False) langfuse_plugin = self._fresh_plugin() assert langfuse_plugin._get_langfuse() is None def test_missing_sdk_logs_one_warning(self, monkeypatch, caplog): langfuse_plugin = self._fresh_plugin() monkeypatch.setattr(langfuse_plugin, "Langfuse", None) langfuse_plugin._LANGFUSE_CLIENT = None with caplog.at_level(logging.WARNING, logger=langfuse_plugin.__name__): assert langfuse_plugin._get_langfuse() is None assert langfuse_plugin._get_langfuse() is None messages = [record.getMessage() for record in caplog.records] assert len(messages) == 1 # --------------------------------------------------------------------------- # 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", ): 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_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") 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) result = mod._safe_value(payload) assert result == { "type": "data_uri", "media_type": "image/png", "omitted": True, "length": len(payload), } 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}}]} ]) assert serialized[0]["content"][0]["image_url"]["url"] == { "type": "data_uri", "media_type": "image/jpeg", "omitted": True, "length": len(payload), } # --------------------------------------------------------------------------- # End-to-end collision regression: two turns of ONE gateway session must not # share trace state. The helper-level tests above prove _trace_key returns # distinct keys; this drives the real pre/post hooks to prove the keys are # actually threaded through so the second turn gets its own root trace. # # Gateway reality this reproduces: # * task_id == session_id for every turn (gateway/run.py) # * turn_id is unique per turn (turn_context.py) # * api_call_count resets to 1 each turn (conversation_loop.py) # # Before the turn/request scoping, _trace_key collapsed to the constant # session_id. That worked only because _finish_trace pops the key on a clean # turn end. When turn 1 does NOT finalize (interrupted, tool-only final step, # or empty final content), its state lingered under session_id and turn 2 # silently merged into turn 1's trace instead of opening its own. # --------------------------------------------------------------------------- class TestTurnTraceIsolation: def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") @staticmethod def _fake_client(started): """A minimal Langfuse stand-in that records each root trace opened. ``_start_root_trace`` calls ``create_trace_id`` then opens a root via ``start_as_current_observation(...)`` (a context manager whose ``__enter__`` returns the root span). We record one entry per root actually opened so the test can count distinct traces. """ class _Span: def update(self, **kw): pass def end(self, **kw): pass def set_trace_io(self, **kw): pass def start_observation(self, **kw): return _Span() class _RootCM: def __enter__(self): return _Span() def __exit__(self, *exc): return False class _Client: def create_trace_id(self, seed=None): return f"trace::{seed}" def start_as_current_observation(self, **kw): started.append(kw.get("trace_context", {}).get("trace_id")) return _RootCM() def flush(self): pass return _Client() def _run_turn(self, mod, *, session, turn_n, finalize): """Drive one turn through the request-scoped hooks the gateway fires.""" task_id = session # gateway sets task_id == session_id turn_id = f"{session}:{task_id}:turn{turn_n}" api_call_count = 1 # resets every turn api_request_id = f"{turn_id}:api:{api_call_count}" mod.on_pre_llm_request( task_id=task_id, session_id=session, model="m", provider="p", api_mode="chat", api_call_count=api_call_count, request_messages=[{"role": "user", "content": "hi"}], turn_id=turn_id, api_request_id=api_request_id, ) # finalize=False => leave a tool call on the final response so # _finish_trace is skipped and the turn's state lingers. mod.on_post_llm_call( task_id=task_id, session_id=session, model="m", provider="p", api_mode="chat", api_call_count=api_call_count, assistant_content_chars=5 if finalize else 0, assistant_tool_call_count=0 if finalize else 1, usage={"input_tokens": 10, "output_tokens": 5}, turn_id=turn_id, api_request_id=api_request_id, ) def test_unfinalized_turn_does_not_capture_next_turn(self, monkeypatch): """A turn that never finalizes must not absorb the following turn.""" mod = self._fresh_plugin() started: list = [] monkeypatch.setattr(mod, "_get_langfuse", lambda: self._fake_client(started)) monkeypatch.setattr(mod, "_end_observation", lambda *a, **k: None) mod._TRACE_STATE.clear() # Turn 1 ends without finalizing (its final step still has a tool call). self._run_turn(mod, session="sess-iso", turn_n=1, finalize=False) # Turn 2 is a normal, fully finalizing turn in the SAME session. self._run_turn(mod, session="sess-iso", turn_n=2, finalize=True) # Each turn opened its OWN root trace. On the pre-fix code the second # turn reused turn 1's lingering state and only one trace was opened. assert len(started) == 2 # Turn 2 finalized and was popped by _finish_trace; only turn 1's # (non-finalizing) state lingers. Assert the surviving key is turn 1's # and that turn 2 never merged into it — `all(...)` over an empty set # would pass vacuously, so pin the exact surviving key instead. keys = list(mod._TRACE_STATE.keys()) assert len(keys) == 1 assert "turn1" in keys[0] assert "turn2" not in keys[0] def test_pre_and_post_hooks_share_one_key_within_a_turn(self, monkeypatch): """turn_id is preferred over api_request_id so the turn-scoped post_llm_call (which carries no api_request_id) still resolves to the same key as the request-scoped pre/post_api_request hooks. If the ordering were reversed, finalization would silently break.""" mod = self._fresh_plugin() 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_post_turn = mod._trace_key("T", "S", turn_id=turn_id, api_request_id="") assert k_pre_api == k_post_api == k_post_turn def test_non_finalizing_turns_do_not_grow_state_unboundedly(self, monkeypatch): """Per-turn keys mean a turn that never finalizes leaves a lingering entry. Without a cap that grows once per non-finalizing turn forever; the LRU eviction must bound _TRACE_STATE at _MAX_TRACE_STATE. """ mod = self._fresh_plugin() started: list = [] monkeypatch.setattr(mod, "_get_langfuse", lambda: self._fake_client(started)) monkeypatch.setattr(mod, "_end_observation", lambda *a, **k: None) monkeypatch.setattr(mod, "_MAX_TRACE_STATE", 8) mod._TRACE_STATE.clear() # Far more non-finalizing turns than the cap. for n in range(50): self._run_turn(mod, session="sess-leak", turn_n=n, finalize=False) assert len(mod._TRACE_STATE) <= 8 # The survivors are the most-recently-updated turns (LRU eviction). surviving = sorted(int(k.rsplit("turn", 1)[1]) for k in mod._TRACE_STATE) assert surviving == list(range(42, 50)) def test_finish_trace_exits_root_context_manager(self, monkeypatch): """_finish_trace must call root_ctx.__exit__(), not just root_span.end(). Regression for the "Exception ignored in: " traceback on CLI exit. The plugin enters the root observation's context manager (start_as_current_observation(...).__enter__()) but must also exit it; otherwise the generator is left suspended and is only unwound when the GC collects it during interpreter teardown. By then opentelemetry.trace.Span has been set to None, and the generator's close() -> use_span.__exit__ -> isinstance(span, Span) raises TypeError: isinstance() arg 2 must be a type. Exiting the context manager here unwinds the generator while modules are intact. """ mod = self._fresh_plugin() started: list = [] monkeypatch.setattr(mod, "_end_observation", lambda *a, **k: None) mod._TRACE_STATE.clear() exited: list = [] class _S: def update(self, **kw): pass def end(self, **kw): pass def set_trace_io(self, **kw): pass def start_observation(self, **kw): return _S() class _TrackingRootCM: def __enter__(self): return _S() def __exit__(self, *exc): exited.append(exc) return False class _TrackingClient: def create_trace_id(self, seed=None): return f"trace::{seed}" def start_as_current_observation(self, **kw): started.append(kw.get("trace_context", {}).get("trace_id")) return _TrackingRootCM() def flush(self): pass monkeypatch.setattr(mod, "_get_langfuse", lambda: _TrackingClient()) self._run_turn(mod, session="sess-exit", turn_n=1, finalize=True) assert exited, ( "_finish_trace did not call root_ctx.__exit__; the generator is " "left suspended and will raise TypeError on GC at interpreter " "teardown when opentelemetry.trace.Span is None" ) assert len(exited) == 1 assert exited[0] == (None, None, None) # --------------------------------------------------------------------------- # 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_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 # -- 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", "sk-lf-real-secret-xyz") 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)" ) 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 TestAssistantMessageSerialization: def test_serialize_assistant_message_prefers_reasoning(self): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") message = SimpleNamespace( content="answer", reasoning="primary reasoning", reasoning_content="fallback reasoning", reasoning_details=[{"type": "summary", "text": "structured reasoning"}], ) assert mod._serialize_assistant_message(message)["reasoning"] == "primary reasoning" def test_serialize_assistant_message_uses_reasoning_content_when_reasoning_absent(self): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") message = SimpleNamespace( content="answer", reasoning=None, reasoning_content="provider scratchpad", reasoning_details=[{"type": "summary", "text": "structured reasoning"}], ) assert mod._serialize_assistant_message(message)["reasoning"] == "provider scratchpad" def test_serialize_assistant_message_uses_structured_reasoning_details(self): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") reasoning_details = [ {"type": "summary", "text": "checked tools"}, {"type": "encrypted_content", "encrypted_content": b"opaque"}, ] message = SimpleNamespace( content="answer", reasoning=None, reasoning_content=None, reasoning_details=reasoning_details, ) assert mod._serialize_assistant_message(message)["reasoning"] == [ {"type": "summary", "text": "checked tools"}, {"type": "encrypted_content", "encrypted_content": {"type": "bytes", "len": 6}}, ] def test_serialize_assistant_message_without_reasoning_fields_sets_none(self): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") message = SimpleNamespace(content="answer") assert mod._serialize_assistant_message(message)["reasoning"] is None 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}, }] 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_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 class TestUsageFromSanitizedResponse: """Regression: ``post_api_request`` delivers ``response`` as a sanitized dict (no ``.usage`` attribute) plus a separate ``usage`` summary dict. The post-call handler must read the ``usage`` dict instead of treating the dict response as a usage-bearing object and dropping all token/cost data.""" def _setup(self, mod, monkeypatch): # Active client so on_post_llm_call does not early-return. monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) 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) captured = {} 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) return captured def test_sanitized_dict_response_uses_usage_dict(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") captured = self._setup(mod, monkeypatch) # A plain dict has no ``.usage`` attribute — mirrors post_api_request. mod.on_post_llm_call( task_id="task-1", 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}}, usage={"input_tokens": 100, "output_tokens": 20}, assistant_content_chars=42, ) # Before the fix the dict response shadowed the usage dict and tokens # were lost (usage_details == {}). assert captured["usage_details"] == {"input": 100, "output": 20} def test_real_response_object_with_usage_still_used(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") captured = self._setup(mod, monkeypatch) # A response object that genuinely carries usage must still take the # response-object path (post_llm_call / legacy behavior). seen = {} def fake_usage_and_cost(resp, **_): seen["resp"] = resp return {"input": 7, "output": 3}, {} monkeypatch.setattr(mod, "_usage_and_cost", fake_usage_and_cost) class _Resp: usage = {"prompt_tokens": 7, "completion_tokens": 3} resp = _Resp() mod.on_post_llm_call( task_id="task-1", session_id="session-1", api_call_count=1, model="gemini-3-flash-preview", response=resp, usage={"input_tokens": 999, "output_tokens": 999}, assistant_content_chars=42, ) assert seen["resp"] is resp assert captured["usage_details"] == {"input": 7, "output": 3} # --------------------------------------------------------------------------- # Model attribution: wire truth over stale agent attribute # --------------------------------------------------------------------------- class TestModelAttribution: def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _client_capturing_generations(self, gens): class _Gen: def update(self, **kw): pass def end(self, **kw): pass class _Span: def update(self, **kw): pass def end(self, **kw): pass def set_trace_io(self, **kw): pass def start_observation(self, **kw): gens.append(kw) return _Gen() class _RootCM: def __enter__(self): return _Span() def __exit__(self, *exc): return False class _Client: def create_trace_id(self, seed=None): return "t" def start_as_current_observation(self, **kw): return _RootCM() def flush(self): pass return _Client() def test_pre_api_request_prefers_request_body_model(self, monkeypatch): """Agent attribute says old model; request body says the switched one.""" mod = self._fresh_plugin() gens: list = [] monkeypatch.setattr(mod, "_get_langfuse", lambda: self._client_capturing_generations(gens)) mod._TRACE_STATE.clear() mod.on_pre_llm_request( task_id="t", session_id="s", turn_id="s:t:turn1", api_call_count=1, model="old-model-attr", provider="openrouter", request_messages=[{"role": "user", "content": "hi"}], request={"body": {"model": "switched/new-model"}}, ) assert gens, "no generation started" assert gens[0]["model"] == "switched/new-model" def test_pre_api_request_falls_back_to_attr_without_body_model(self, monkeypatch): mod = self._fresh_plugin() gens: list = [] monkeypatch.setattr(mod, "_get_langfuse", lambda: self._client_capturing_generations(gens)) mod._TRACE_STATE.clear() mod.on_pre_llm_request( task_id="t", session_id="s", turn_id="s:t:turn2", api_call_count=1, model="attr-model", request_messages=[{"role": "user", "content": "hi"}], request={"body": {}}, ) assert gens[0]["model"] == "attr-model" def test_post_api_request_uses_response_model_for_cost(self, monkeypatch): """Cost estimation must key off the model that actually served.""" mod = self._fresh_plugin() monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) mod._TRACE_STATE.clear() seen = {} def fake_usage_and_cost(response, *, provider, api_mode, model, base_url): seen["model"] = model return {"input": 1, "output": 1}, {} monkeypatch.setattr(mod, "_usage_and_cost", fake_usage_and_cost) class _Gen: def update(self, **kw): pass def end(self, **kw): pass class _Root: def update(self, **kw): pass def end(self, **kw): pass def set_trace_io(self, **kw): pass turn_id = "s:t:turn3" key = mod._trace_key("t", "s", turn_id=turn_id) state = mod.TraceState(trace_id="x", root_ctx=None, root_span=_Root()) state.generations["1"] = _Gen() mod._TRACE_STATE[key] = state class _Resp: usage = {"prompt_tokens": 1, "completion_tokens": 1} mod.on_post_llm_call( task_id="t", session_id="s", turn_id=turn_id, api_call_count=1, model="stale-attr-model", response_model="actual/served-model", response=_Resp(), assistant_content_chars=2, ) assert seen["model"] == "actual/served-model" # --------------------------------------------------------------------------- # Cost total: explicit "total" alongside the per-type breakdown # --------------------------------------------------------------------------- class TestCostTotal: """Langfuse ingests per-type ``cost_details`` keys but does not derive ``calculatedTotalCost`` from them. Without an explicit ``total`` the dashboard reads 0 for every priced generation.""" def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def test_response_path_totals_the_breakdown(self): mod = self._fresh_plugin() class _Usage: input_tokens = 1000 output_tokens = 500 cache_read_input_tokens = 2000 cache_creation_input_tokens = 0 class _Resp: usage = _Usage() _, cost_details = mod._usage_and_cost( _Resp(), provider="anthropic", api_mode="anthropic_messages", model="claude-sonnet-4-6", base_url="", ) components = {k: v for k, v in cost_details.items() if k != "total"} assert components assert cost_details["total"] == pytest.approx(sum(components.values())) def test_usage_summary_path_totals_the_breakdown(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) state = mod.TraceState(trace_id="trace-1", root_ctx=None, root_span=None) state.generations[mod._request_key(1)] = object() 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): captured["cost_details"] = cost_details monkeypatch.setattr(mod, "_end_observation", fake_end_observation) # A dict response has no ``.usage``, so the handler takes the # usage-summary path rather than the response-object path. mod.on_post_llm_call( task_id="task-1", session_id="session-1", api_call_count=1, model="claude-sonnet-4-6", provider="anthropic", response={"model": "claude-sonnet-4-6"}, usage={"input_tokens": 1000, "output_tokens": 500}, assistant_content_chars=42, ) cost_details = captured["cost_details"] components = {k: v for k, v in cost_details.items() if k != "total"} assert components assert cost_details["total"] == pytest.approx(sum(components.values())) def test_priced_model_with_no_tokens_reports_no_total(self): mod = self._fresh_plugin() class _Usage: input_tokens = 0 output_tokens = 0 class _Resp: usage = _Usage() _, cost_details = mod._usage_and_cost( _Resp(), provider="anthropic", api_mode="anthropic_messages", model="claude-sonnet-4-6", base_url="", ) # A priced model that billed nothing writes no per-type keys, so # summing them must not invent a 0.0 total on an empty breakdown. assert cost_details == {} # --------------------------------------------------------------------------- # Capture modes: metadata | sanitized | full (HERMES_LANGFUSE_CAPTURE) # --------------------------------------------------------------------------- class TestCaptureModes: def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def test_default_mode_is_sanitized(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.delenv("HERMES_LANGFUSE_CAPTURE", raising=False) assert mod._capture_mode() == "sanitized" def test_invalid_mode_falls_back_and_warns_once(self, monkeypatch, caplog): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "everything") with caplog.at_level(logging.WARNING): assert mod._capture_mode() == "sanitized" assert mod._capture_mode() == "sanitized" warnings = [r for r in caplog.records if "HERMES_LANGFUSE_CAPTURE" in r.getMessage()] assert len(warnings) == 1 def test_metadata_mode_omits_content(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "metadata") out = mod._capture_content("top secret prompt text") assert out == {"omitted": True, "type": "text", "chars": 22} obj = mod._capture_content({"password": "hunter22", "path": "/x"}) assert obj["omitted"] is True assert set(obj["keys"]) == {"password", "path"} assert "hunter22" not in str(obj) def test_metadata_mode_message_serialization_keeps_roles(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "metadata") msgs = mod._serialize_messages([ {"role": "user", "content": "my ssn is 123-45-6789"}, {"role": "assistant", "content": "noted"}, ]) assert [m["role"] for m in msgs] == ["user", "assistant"] assert all(isinstance(m["content"], dict) and m["content"]["omitted"] for m in msgs) assert "6789" not in str(msgs) def test_sanitized_mode_redacts_secrets(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "sanitized") samples = { "openai": "here sk-" + "a" * 20 + " done", "anthropic": "key sk-ant-" + "a" * 20 + " x", "github": "tok ghp_" + "a" * 36, "aws": "AKIA" + "A" * 16, "langfuse": "pk-lf-" + "a" * 20, "bearer": "Authorization: Bearer " + "a" * 20, "assignment": 'api_key="supersecretvalue"', } for name, text in samples.items(): out = mod._capture_content(text) # redact_sensitive_text masks secrets (e.g. "sk-aaa...aaaa") or # replaces them with "«redacted:...»" sentinels — check that the # original secret substring is gone, not for a specific marker. assert text != out, f"{name} not redacted: {out!r}" def test_sanitized_mode_redacts_before_truncation(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "sanitized") secret = "sk-" + "z" * 40 text = "x" * 100 + " " + secret + " " + "y" * 100 out = mod._truncate_text(text, 120) assert "z" * 10 not in out assert text != out, "secret was not redacted before truncation" def test_sanitized_mode_keeps_ordinary_text(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "sanitized") text = "refactor the memory manager to emit spans" assert mod._capture_content(text) == text def test_full_mode_keeps_secret_shaped_text(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "full") text = "here sk-abcdefghijklmnop1234 done" assert mod._capture_content(text) == text # --------------------------------------------------------------------------- # api_request_error hook # --------------------------------------------------------------------------- class TestApiRequestErrorHook: def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _seed_state(self, mod, task_key, gen_key="1"): class _Gen: def __init__(self): self.updates = [] self.ended = False def update(self, **kw): self.updates.append(kw) def end(self, **kw): self.ended = True class _Root: def __init__(self): self.ended = False def update(self, **kw): pass def end(self, **kw): self.ended = True def set_trace_io(self, **kw): pass gen = _Gen() root = _Root() state = mod.TraceState(trace_id="t", root_ctx=None, root_span=root) state.generations[gen_key] = gen mod._TRACE_STATE[task_key] = state return gen, root def test_retryable_error_closes_generation_keeps_turn(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) mod._TRACE_STATE.clear() turn_id = "s:t:turn1" task_key = mod._trace_key("t", "s", turn_id=turn_id) gen, root = self._seed_state(mod, task_key) mod.on_api_request_error( task_id="t", session_id="s", api_call_count=1, turn_id=turn_id, status_code=429, retryable=True, retry_count=1, max_retries=3, error={"type": "RateLimitError", "message": "slow down"}, ) assert gen.ended is True assert any(u.get("level") == "ERROR" for u in gen.updates) # error metadata landed meta = [u["metadata"] for u in gen.updates if "metadata" in u] assert meta and meta[0]["status_code"] == 429 # turn stays open for the retry assert task_key in mod._TRACE_STATE assert root.ended is False def test_terminal_error_finishes_turn(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setattr(mod, "_get_langfuse", lambda: type("C", (), {"flush": lambda self: None})()) mod._TRACE_STATE.clear() turn_id = "s:t:turn2" task_key = mod._trace_key("t", "s", turn_id=turn_id) gen, root = self._seed_state(mod, task_key) mod.on_api_request_error( task_id="t", session_id="s", api_call_count=1, turn_id=turn_id, status_code=401, retryable=False, error={"type": "AuthenticationError", "message": "bad key"}, ) assert gen.ended is True assert task_key not in mod._TRACE_STATE assert root.ended is True def test_error_hook_noops_without_state(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) mod._TRACE_STATE.clear() # Must not raise mod.on_api_request_error( task_id="t", session_id="s", api_call_count=1, error={"type": "X", "message": "y"}, retryable=False, ) def test_error_message_respects_capture_mode(self, monkeypatch): mod = self._fresh_plugin() monkeypatch.setenv("HERMES_LANGFUSE_CAPTURE", "metadata") monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) mod._TRACE_STATE.clear() turn_id = "s:t:turn3" task_key = mod._trace_key("t", "s", turn_id=turn_id) gen, _root = self._seed_state(mod, task_key) mod.on_api_request_error( task_id="t", session_id="s", api_call_count=1, turn_id=turn_id, retryable=True, error={"type": "APIError", "message": "secret prompt echo sk-abc"}, ) meta = [u["metadata"] for u in gen.updates if "metadata" in u][0] assert isinstance(meta["error_message"], dict) assert meta["error_message"]["omitted"] is True # --------------------------------------------------------------------------- # on_session_finalize hook # --------------------------------------------------------------------------- class TestSessionFinalizeHook: def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _client(self, flushes): class _Client: def flush(self): flushes.append(1) return _Client() def _state(self, mod): class _Root: def __init__(self): self.ended = False def update(self, **kw): pass def end(self, **kw): self.ended = True def set_trace_io(self, **kw): pass root = _Root() return mod.TraceState(trace_id="t", root_ctx=None, root_span=root), root def test_finalize_closes_matching_session_traces(self, monkeypatch): mod = self._fresh_plugin() flushes = [] client = self._client(flushes) monkeypatch.setattr(mod, "_LANGFUSE_CLIENT", client) monkeypatch.setattr(mod, "_get_langfuse", lambda: client) mod._TRACE_STATE.clear() s1, r1 = self._state(mod) s2, r2 = self._state(mod) mod._TRACE_STATE["session:sess-a:turn:1"] = s1 mod._TRACE_STATE["session:sess-b:turn:1"] = s2 mod.on_session_finalize(session_id="sess-a") assert "session:sess-a:turn:1" not in mod._TRACE_STATE assert "session:sess-b:turn:1" in mod._TRACE_STATE assert r1.ended is True assert r2.ended is False assert flushes # flushed at least once def test_finalize_without_session_closes_all(self, monkeypatch): mod = self._fresh_plugin() flushes = [] client = self._client(flushes) monkeypatch.setattr(mod, "_LANGFUSE_CLIENT", client) monkeypatch.setattr(mod, "_get_langfuse", lambda: client) mod._TRACE_STATE.clear() s1, r1 = self._state(mod) mod._TRACE_STATE["session:sess-x:turn:1"] = s1 mod.on_session_finalize() assert not mod._TRACE_STATE assert r1.ended is True def test_finalize_noop_when_client_never_initialized(self): mod = self._fresh_plugin() mod._TRACE_STATE.clear() # _LANGFUSE_CLIENT is None on a fresh module; must not raise or init. mod.on_session_finalize(session_id="whatever") def test_finalize_shuts_down_client_on_process_exit(self, monkeypatch): """reason="shutdown" must call client.shutdown() while the interpreter is alive, so the SDK's own atexit handler (which runs during interpreter finalization, after opentelemetry.trace.Span is torn down) becomes a no-op instead of raising the "isinstance() arg 2 must be a type" TypeError on quit.""" mod = self._fresh_plugin() events = [] class _Client: def flush(self): events.append("flush") def shutdown(self): events.append("shutdown") client = _Client() monkeypatch.setattr(mod, "_LANGFUSE_CLIENT", client) monkeypatch.setattr(mod, "_get_langfuse", lambda: client) mod._TRACE_STATE.clear() mod.on_session_finalize(session_id="sess-a", reason="shutdown") assert "shutdown" in events assert events.index("flush") < events.index("shutdown") def test_finalize_keeps_client_alive_on_session_rotation(self, monkeypatch): """/new, /reset, and gateway session expiry finalize the session but the process lives on — the cached client must NOT be shut down or later sessions silently stop exporting.""" mod = self._fresh_plugin() events = [] class _Client: def flush(self): events.append("flush") def shutdown(self): events.append("shutdown") client = _Client() monkeypatch.setattr(mod, "_LANGFUSE_CLIENT", client) monkeypatch.setattr(mod, "_get_langfuse", lambda: client) mod._TRACE_STATE.clear() for reason in ("session_boundary", "new_session", "session_expired", ""): mod.on_session_finalize(session_id="sess-a", reason=reason) assert "shutdown" not in events assert "flush" in events # --------------------------------------------------------------------------- # Subagent tracing: delegated children as spans under the parent turn # --------------------------------------------------------------------------- class TestSubagentTracing: """``tools/delegate_tool.py`` emits subagent_start/subagent_stop. The payloads carry ``parent_turn_id`` but no ``task_id``, so the parent trace must be resolved by turn id rather than by rebuilding the scope key.""" def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _state(self, mod, monkeypatch, key, spans): class _Obs: def __init__(self, kw): self.kw = kw self.ended = False self.updates = {} def update(self, **kw): self.updates.update(kw) def end(self, **kw): self.ended = True class _Root: def start_observation(self, **kw): obs = _Obs(kw) spans.append(obs) return obs monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) state = mod.TraceState(trace_id="trace-1", root_ctx=None, root_span=_Root()) monkeypatch.setitem(mod._TRACE_STATE, key, state) return state def test_start_attaches_span_despite_task_scoped_key(self, monkeypatch): mod = self._fresh_plugin() spans = [] # Key minted by the LLM hooks with a task id — a naive rebuild from # session_id alone would not match this. state = self._state(mod, monkeypatch, "task:task-9:turn:turn-7", spans) mod.on_subagent_start( parent_session_id="sess-1", parent_turn_id="turn-7", child_session_id="child-sess-1", child_subagent_id="sub-1", child_role="researcher", child_goal="find the thing", ) assert len(spans) == 1 assert spans[0].kw["name"] == "Subagent: researcher" assert spans[0].kw["metadata"]["child_subagent_id"] == "sub-1" assert "child-sess-1" in state.subagents def test_stop_ends_span_and_records_outcome(self, monkeypatch): mod = self._fresh_plugin() spans = [] state = self._state(mod, monkeypatch, "task:task-9:turn:turn-7", spans) mod.on_subagent_start( parent_session_id="sess-1", parent_turn_id="turn-7", child_session_id="child-sess-1", child_role="researcher", child_goal="find the thing", ) mod.on_subagent_stop( parent_session_id="sess-1", parent_turn_id="turn-7", child_session_id="child-sess-1", child_role="researcher", child_summary="found it", child_status="ok", tool_call_history=[{"name": "read_file"}, {"name": "grep"}], duration_ms=1234, ) assert spans[0].ended is True assert spans[0].updates["metadata"]["status"] == "ok" assert spans[0].updates["metadata"]["tool_call_count"] == 2 assert spans[0].updates["metadata"]["duration_ms"] == 1234 # Popped so a repeated stop cannot double-end the span. assert not state.subagents def test_unknown_turn_is_a_noop(self, monkeypatch): mod = self._fresh_plugin() spans = [] self._state(mod, monkeypatch, "task:task-9:turn:turn-7", spans) mod.on_subagent_start( parent_turn_id="turn-does-not-exist", child_session_id="child-sess-1", child_role="researcher", ) mod.on_subagent_stop( parent_turn_id="turn-does-not-exist", child_session_id="child-sess-1", ) assert spans == [] def test_start_without_child_session_is_a_noop(self, monkeypatch): mod = self._fresh_plugin() spans = [] self._state(mod, monkeypatch, "task:task-9:turn:turn-7", spans) # subagent_stop keys on child_session_id, so a start without one could # never be matched and must not open an unclosable span. mod.on_subagent_start( parent_turn_id="turn-7", child_session_id=None, child_role="researcher", ) assert spans == [] # --------------------------------------------------------------------------- # MoA fan-out: one generation per advisor, priced at the advisor's own model # --------------------------------------------------------------------------- class TestMoAReferenceGenerations: """MoA returns only the aggregator's response, so without per-advisor generations the whole fan-out collapses into one line priced at the aggregator's model. Advisors routinely run on a different provider.""" def _fresh_plugin(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _state(self, mod, monkeypatch, gens): class _Obs: def __init__(self, kw): self.kw = kw self.updates = {} self.ended = False def update(self, **kw): self.updates.update(kw) def end(self, **kw): self.ended = True class _Root: def start_observation(self, **kw): obs = _Obs(kw) gens.append(obs) return obs monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) return mod.TraceState(trace_id="t", root_ctx=None, root_span=_Root()) def _refs(self): return [ { "label": "anthropic:claude-sonnet-4-6", "model": "claude-sonnet-4-6", "provider": "anthropic", "usage": {"input_tokens": 100, "output_tokens": 50}, "cost_usd": 0.001, "cost_status": "ok", "cost_source": "pricing_table", }, { "label": "openai:gpt-5", "model": "gpt-5", "provider": "openai", "usage": {"input_tokens": 80, "output_tokens": 40, "reasoning_tokens": 10}, "cost_usd": 0.002, }, ] def test_one_generation_per_advisor_with_own_model_and_cost(self, monkeypatch): mod = self._fresh_plugin() gens = [] state = self._state(mod, monkeypatch, gens) mod._emit_moa_reference_generations(state, client=object(), references=self._refs()) assert len(gens) == 2 assert gens[0].kw["model"] == "claude-sonnet-4-6" assert gens[1].kw["model"] == "gpt-5" # Each advisor's dollars, not the aggregator's rate applied to all. assert gens[0].updates["cost_details"]["total"] == pytest.approx(0.001) assert gens[1].updates["cost_details"]["total"] == pytest.approx(0.002) assert gens[0].updates["usage_details"] == {"input": 100, "output": 50} assert gens[1].updates["usage_details"]["reasoning_tokens"] == 10 assert all(g.ended for g in gens) def test_repeat_emit_is_deduped_within_a_turn(self, monkeypatch): mod = self._fresh_plugin() gens = [] state = self._state(mod, monkeypatch, gens) # The MoA client holds its last fan-out until the next one, so a # tool-loop turn delivers the same references on every API call. refs = self._refs() mod._emit_moa_reference_generations(state, client=object(), references=refs) mod._emit_moa_reference_generations(state, client=object(), references=refs) mod._emit_moa_reference_generations(state, client=object(), references=list(refs)) assert len(gens) == 2 def test_a_new_fanout_emits_again(self, monkeypatch): mod = self._fresh_plugin() gens = [] state = self._state(mod, monkeypatch, gens) mod._emit_moa_reference_generations(state, client=object(), references=self._refs()) second = self._refs() second[0]["usage"]["output_tokens"] = 999 mod._emit_moa_reference_generations(state, client=object(), references=second) assert len(gens) == 4 def test_non_moa_turn_emits_nothing(self, monkeypatch): mod = self._fresh_plugin() gens = [] state = self._state(mod, monkeypatch, gens) for value in (None, [], "not-a-list", [None, "junk"]): mod._emit_moa_reference_generations(state, client=object(), references=value) assert gens == [] class TestAtexitFinalization(TestTurnTraceIsolation): """Short-lived processes (kanban workers, `hermes chat -q`, cron) can exit with tool calls still queued — the root span never ends and the backend shows an anonymous trace (no name/session/metadata). _finalize_all_traces (registered atexit after client construction) must end every open root.""" def test_finalize_all_ends_open_roots_and_clears_state(self, monkeypatch): mod = self._fresh_plugin() started: list = [] ended: list = [] client = self._fake_client(started) monkeypatch.setattr(mod, "_get_langfuse", lambda: client) monkeypatch.setattr( mod, "_end_observation", lambda obs, **k: ended.append(obs) ) mod._TRACE_STATE.clear() # Three worker-style turns that never finalize (tool calls pending). for n in range(3): self._run_turn(mod, session=f"worker-{n}", turn_n=0, finalize=False) assert len(mod._TRACE_STATE) == 3 root_ends: list = [] for state in mod._TRACE_STATE.values(): real_end = state.root_span.end state.root_span.end = lambda *a, _r=real_end, **k: root_ends.append(1) mod._finalize_all_traces() assert len(root_ends) == 3, "every open root span must be ended" assert mod._TRACE_STATE == {}, "state must be drained" # Idempotent: a second call (SDK/atexit re-entry) is a no-op. mod._finalize_all_traces() assert len(root_ends) == 3 def test_atexit_hook_is_registered_on_client_init(self, monkeypatch): mod = self._fresh_plugin() registered: list = [] import atexit as _atexit monkeypatch.setattr(mod, "Langfuse", lambda **kw: object()) monkeypatch.setattr( _atexit, "register", lambda fn, *a, **k: registered.append(fn) ) monkeypatch.setenv("HERMES_LANGFUSE_PUBLIC_KEY", "pk-lf-0123456789abcdef") monkeypatch.setenv("HERMES_LANGFUSE_SECRET_KEY", "sk-lf-0123456789abcdef") mod._LANGFUSE_CLIENT = None assert mod._get_langfuse() is not None assert mod._finalize_all_traces in registered def test_finalize_flushes_every_profile_without_an_ambient_scope(self, monkeypatch, tmp_path): """Multiplex gateway: every turn ran inside a profile scope (home override + secret scope), so only the per-home slots hold clients and the launch slot stays empty. atexit has no scope, and a credential read there raises UnscopedSecretError. The finalizer must not read credentials at all — it ends the open roots and flushes each settled client, so neither profile loses its pending traces.""" from agent import secret_scope from hermes_constants import reset_hermes_home_override, set_hermes_home_override mod = self._fresh_plugin() monkeypatch.setattr(secret_scope, "_MULTIPLEX_ACTIVE", True) monkeypatch.setattr(mod, "_end_observation", lambda obs, **k: None) mod._LANGFUSE_CLIENT = None mod._TRACE_STATE.clear() mod._LANGFUSE_CLIENT_BY_HOME.clear() flushed: list = [] fake_client = self._fake_client def _sdk(**kw): client = fake_client([]) client.flush = lambda pk=kw["public_key"]: flushed.append(pk) return client monkeypatch.setattr(mod, "Langfuse", _sdk) for profile in ("alpha", "beta"): home = tmp_path / profile home.mkdir() home_token = set_hermes_home_override(home) scope_token = secret_scope.set_secret_scope({ "HERMES_LANGFUSE_PUBLIC_KEY": f"pk-lf-{profile}-0123456789", "HERMES_LANGFUSE_SECRET_KEY": f"sk-lf-{profile}-0123456789", }) try: self._run_turn(mod, session=f"{profile}-turn", turn_n=0, finalize=False) finally: secret_scope.reset_secret_scope(scope_token) reset_hermes_home_override(home_token) assert len(mod._TRACE_STATE) == 2 and len(mod._LANGFUSE_CLIENT_BY_HOME) == 2 assert mod._LANGFUSE_CLIENT is None and secret_scope.current_secret_scope() is None mod._finalize_all_traces() # no scope: must not raise, must not build a client assert sorted(flushed) == ["pk-lf-alpha-0123456789", "pk-lf-beta-0123456789"] assert mod._TRACE_STATE == {} and mod._LANGFUSE_CLIENT is None class TestSystemPromptInGenerationInput: """The generation input must carry the system prompt even for providers that move it out of ``messages``: Anthropic Messages (``system`` kwarg) and the Responses/Codex API (``instructions``). Hermes forwards it to hooks as ``system_prompt``; the plugin prepends a ``role: system`` entry. Regression for the trace gap discussed in PR #32175 (Anthropic) and its Codex sibling: without this, hosted traces show conversations without the agent's instructions, skills, and memory.""" def _make_mod(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _capture_generation(self, mod, monkeypatch): """Route on_pre_llm_request into a seeded TraceState and record the generation observation kwargs.""" captured = {} monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) task_key = mod._trace_key("task-1", "sess-1") monkeypatch.setitem(mod._TRACE_STATE, task_key, state) def fake_child(state_, **kw): captured["input"] = kw.get("input_value") captured["metadata"] = kw.get("metadata") return object() monkeypatch.setattr(mod, "_start_child_observation", fake_child) return captured def _fire(self, mod, *, request_messages, system_prompt=None): kwargs = dict( task_id="task-1", session_id="sess-1", model="m", provider="p", api_mode="codex_responses", api_call_count=1, request_messages=request_messages, ) if system_prompt is not None: kwargs["system_prompt"] = system_prompt mod.on_pre_llm_request(**kwargs) def test_anthropic_block_list_flattened(self, monkeypatch): """Anthropic OAuth mode sends ``system`` as content blocks (with cache_control); the trace should carry the readable text.""" mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) blocks = [ {"type": "text", "text": "part one", "cache_control": {"type": "ephemeral"}}, {"type": "text", "text": "part two"}, ] self._fire( mod, request_messages=[{"role": "user", "content": "hi"}], system_prompt=blocks, ) first = captured["input"][0] assert first["role"] == "system" assert "part one" in first["content"] assert "part two" in first["content"] def test_absent_system_prompt_keeps_previous_shape(self, monkeypatch): mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) self._fire(mod, request_messages=[{"role": "user", "content": "hi"}]) assert captured["input"][0]["role"] == "user" assert "system_prompt_chars" not in (captured["metadata"] or {}) def test_system_survives_serialization_window(self, monkeypatch): """_serialize_messages keeps only the last 12 messages; the system prompt must be prepended after windowing so long conversations never drop it.""" mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) many = [ {"role": "user" if i % 2 == 0 else "assistant", "content": f"m{i}"} for i in range(30) ] self._fire(mod, request_messages=many, system_prompt="SYS") assert captured["input"][0]["role"] == "system" assert captured["input"][0]["content"] == "SYS" class TestSystemPromptCrossesHookBoundary: """End-to-end across the hook seam with real transport-built kwargs — the regression coverage PR #32175's review asked for: verify the provider-specific request shape (Anthropic ``system`` kwarg, Codex ``instructions``) actually reaches the Langfuse generation input, with no Hermes internals mocked (only the Langfuse client is faked).""" def _make_mod(self): sys.modules.pop("plugins.observability.langfuse", None) return importlib.import_module("plugins.observability.langfuse") def _capture_generation(self, mod, monkeypatch): captured = {} monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) state = mod.TraceState(trace_id="t", root_ctx=None, root_span=None) task_key = mod._trace_key("task-1", "sess-1") monkeypatch.setitem(mod._TRACE_STATE, task_key, state) def fake_child(state_, **kw): captured["input"] = kw.get("input_value") return object() monkeypatch.setattr(mod, "_start_child_observation", fake_child) return captured def _derive_and_fire(self, mod, api_kwargs, api_messages): """Mirror agent/conversation_loop.py's pre_api_request emission: derive request_messages exactly the way the loop does, derive system_prompt via the loop's helper, and invoke the plugin hook.""" from agent.conversation_loop import _system_prompt_for_hooks 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 mod.on_pre_llm_request( task_id="task-1", session_id="sess-1", model="m", provider="p", api_mode="x", api_call_count=1, request_messages=list(request_messages), system_prompt=_system_prompt_for_hooks(api_kwargs, request_messages), ) def test_codex_instructions_reach_generation_input(self, monkeypatch): from agent.transports.codex import ResponsesApiTransport api_messages = [ {"role": "system", "content": "SYS-CODEX"}, {"role": "user", "content": "hi"}, ] api_kwargs = ResponsesApiTransport().build_kwargs("gpt-x", api_messages, None) # Premise: the Responses API moves the system prompt out of the input. assert api_kwargs["instructions"] == "SYS-CODEX" assert all(i.get("role") != "system" for i in api_kwargs["input"] if isinstance(i, dict)) mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) self._derive_and_fire(mod, api_kwargs, api_messages) assert captured["input"][0]["role"] == "system" assert captured["input"][0]["content"] == "SYS-CODEX" def test_anthropic_system_kwarg_reaches_generation_input(self, monkeypatch): from agent.transports.anthropic import AnthropicTransport api_messages = [ {"role": "system", "content": "SYS-ANTHROPIC"}, {"role": "user", "content": "hi"}, ] api_kwargs = AnthropicTransport().build_kwargs( "claude-x", api_messages, None, max_tokens=64 ) # Premise: the Messages API moves the system prompt to a kwarg. assert "system" in api_kwargs assert all(m.get("role") != "system" for m in api_kwargs["messages"]) mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) self._derive_and_fire(mod, api_kwargs, api_messages) assert captured["input"][0]["role"] == "system" assert "SYS-ANTHROPIC" in captured["input"][0]["content"] def test_bedrock_system_kwarg_reaches_generation_input(self, monkeypatch): from agent.transports.bedrock import BedrockTransport api_messages = [ {"role": "system", "content": "SYS-BEDROCK"}, {"role": "user", "content": "hi"}, ] api_kwargs = BedrockTransport().build_kwargs( "anthropic.claude-x", api_messages, None, max_tokens=64 ) # Premise: Bedrock Converse moves system into a separate 'system' kwarg, # shaped as [{"text": ...}] blocks — no "type" key, unlike Anthropic. # (The transport may append extra blocks, e.g. cachePoint markers.) assert {"text": "SYS-BEDROCK"} in api_kwargs["system"] assert all(m.get("role") != "system" for m in api_kwargs["messages"]) mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) self._derive_and_fire(mod, api_kwargs, api_messages) assert captured["input"][0]["role"] == "system" assert "SYS-BEDROCK" in captured["input"][0]["content"] def test_chat_completions_shape_needs_no_fallback(self, monkeypatch): """When system stays in messages[0] (chat_completions), the helper returns it but the plugin must not duplicate the entry.""" from agent.conversation_loop import _system_prompt_for_hooks api_kwargs = { "messages": [ {"role": "system", "content": "SYS-CHAT"}, {"role": "user", "content": "hi"}, ] } sp = _system_prompt_for_hooks(api_kwargs, api_kwargs["messages"]) assert sp == "SYS-CHAT" mod = self._make_mod() captured = self._capture_generation(mod, monkeypatch) self._derive_and_fire(mod, api_kwargs, api_kwargs["messages"]) roles = [m["role"] for m in captured["input"]] assert roles.count("system") == 1 class TestFinishTraceUsesUpdateTrace: """Regression: SDK v3 has update_trace, not set_trace_io. Calling the non-existent set_trace_io raised AttributeError inside _finish_trace's try block and skipped root_span.end(). Generations/tools still exported, so the Langfuse list showed Observation Levels + Latency but blank Input/Output columns (no CHAIN root). """ def test_finish_ends_root_and_calls_update_trace(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") roots: list = [] class _Span: def __init__(self): self.ended = False self.updates = [] self.trace_updates = [] def update(self, **kw): self.updates.append(kw) def end(self, **kw): self.ended = True def update_trace(self, **kw): self.trace_updates.append(kw) def start_observation(self, **kw): return _Span() # Deliberately NO set_trace_io — mirrors real LangfuseChain. class _RootCM: def __init__(self): self.span = _Span() roots.append(self.span) def __enter__(self): return self.span def __exit__(self, *exc): return False class _Client: def create_trace_id(self, seed=None): return f"trace::{seed}" def start_as_current_observation(self, **kw): return _RootCM() def flush(self): pass monkeypatch.setattr(mod, "_get_langfuse", lambda: _Client()) monkeypatch.setattr(mod, "_end_observation", lambda *a, **k: None) mod._TRACE_STATE.clear() mod.on_pre_llm_request( task_id="t1", session_id="s1", model="m", provider="p", api_mode="chat", api_call_count=1, request_messages=[{"role": "user", "content": "hi"}], turn_id="turn-1", ) mod.on_post_llm_call( task_id="t1", session_id="s1", model="m", provider="p", api_mode="chat", api_call_count=1, assistant_content_chars=12, assistant_tool_call_count=0, assistant_response="hello world!", turn_id="turn-1", ) assert len(roots) == 1 root = roots[0] assert root.ended is True assert any("output" in u for u in root.trace_updates) assert any("output" in u for u in root.updates) assert mod._TRACE_STATE == {} def test_finish_still_ends_when_update_trace_raises(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") roots: list = [] class _Span: def __init__(self): self.ended = False def update(self, **kw): pass def end(self, **kw): self.ended = True def update_trace(self, **kw): raise RuntimeError("simulated update_trace failure") def start_observation(self, **kw): return _Span() class _RootCM: def __init__(self): self.span = _Span() roots.append(self.span) def __enter__(self): return self.span def __exit__(self, *exc): return False class _Client: def create_trace_id(self, seed=None): return f"trace::{seed}" def start_as_current_observation(self, **kw): return _RootCM() def flush(self): pass monkeypatch.setattr(mod, "_get_langfuse", lambda: _Client()) monkeypatch.setattr(mod, "_end_observation", lambda *a, **k: None) mod._TRACE_STATE.clear() mod.on_pre_llm_request( task_id="t1", session_id="s1", model="m", provider="p", api_mode="chat", api_call_count=1, request_messages=[{"role": "user", "content": "hi"}], turn_id="turn-1", ) mod.on_post_llm_call( task_id="t1", session_id="s1", model="m", provider="p", api_mode="chat", api_call_count=1, assistant_content_chars=5, assistant_tool_call_count=0, assistant_response="done", turn_id="turn-1", ) assert roots[0].ended is True assert mod._TRACE_STATE == {} class TestCanonicalCostExport: """Both supported response paths must export the same complete cost.""" @staticmethod def _response(input_tokens, output_tokens, cache_read=0, cache_write=0): cache_details = SimpleNamespace( cached_tokens=cache_read, cache_write_tokens=cache_write, ) usage = SimpleNamespace( # Anthropic response shape. input_tokens=input_tokens, output_tokens=output_tokens, cache_read_input_tokens=cache_read, cache_creation_input_tokens=cache_write, # OpenAI chat response shape used by the included-route case. prompt_tokens=input_tokens + cache_read + cache_write, completion_tokens=output_tokens, prompt_tokens_details=cache_details, ) return SimpleNamespace(usage=usage) @staticmethod def _summary(input_tokens, output_tokens, cache_read=0, cache_write=0, request_count=1): return { "input_tokens": input_tokens, "output_tokens": output_tokens, "cache_read_tokens": cache_read, "cache_write_tokens": cache_write, "reasoning_tokens": 0, "request_count": request_count, } @staticmethod def _capture_summary_path(mod, monkeypatch, usage, *, provider, model, api_mode): monkeypatch.setattr(mod, "_get_langfuse", lambda: object()) observation = object() state = mod.TraceState(trace_id="trace-cost", root_ctx=None, root_span=None) state.generations[mod._request_key(1)] = observation task_key = mod._trace_key("task-cost", "session-cost") monkeypatch.setitem(mod._TRACE_STATE, task_key, state) captured = {} def fake_end_observation( obs, *, output=None, metadata=None, usage_details=None, cost_details=None, ): captured["usage_details"] = usage_details captured["cost_details"] = cost_details monkeypatch.setattr(mod, "_end_observation", fake_end_observation) mod.on_post_llm_call( task_id="task-cost", session_id="session-cost", api_call_count=1, model=model, provider=provider, api_mode=api_mode, response={"model": model}, usage=usage, ) return captured["usage_details"], captured["cost_details"] def _run_both_paths( self, mod, monkeypatch, usage, *, provider="anthropic", model="priced-model", api_mode="anthropic_messages", ): response_result = mod._usage_and_cost( self._response( usage["input_tokens"], usage["output_tokens"], usage.get("cache_read_tokens", 0), usage.get("cache_write_tokens", 0), ), provider=provider, api_mode=api_mode, model=model, base_url="", ) summary_result = self._capture_summary_path( mod, monkeypatch, usage, provider=provider, model=model, api_mode=api_mode, ) assert response_result[0] == summary_result[0] return response_result[1], summary_result[1] @pytest.mark.parametrize( ("cache_read", "cache_write", "expected_total"), [ (0, 0, 0.00002), (2, 3, 0.0000255), ], ids=("no-cache", "cached"), ) def test_known_costs_include_canonical_total_on_both_paths( self, monkeypatch, cache_read, cache_write, expected_total, ): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") import agent.usage_pricing as pricing entry = pricing.PricingEntry( input_cost_per_million=Decimal("1"), output_cost_per_million=Decimal("2"), cache_read_cost_per_million=Decimal("0.5"), cache_write_cost_per_million=Decimal("1.5"), source="custom_contract", ) monkeypatch.setattr(pricing, "get_pricing_entry", lambda *_, **__: entry) usage = self._summary(10, 5, cache_read, cache_write) response_cost, summary_cost = self._run_both_paths(mod, monkeypatch, usage) expected = { "total": expected_total, "input": 0.00001, "output": 0.00001, } if cache_read: expected["cache_read_input_tokens"] = 0.000001 if cache_write: expected["cache_creation_input_tokens"] = 0.0000045 assert response_cost == pytest.approx(expected) assert summary_cost == pytest.approx(expected) def test_total_uses_request_cost_instead_of_component_sum(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") import agent.usage_pricing as pricing entry = pricing.PricingEntry( input_cost_per_million=Decimal("1"), output_cost_per_million=Decimal("2"), cache_read_cost_per_million=Decimal("0.5"), cache_write_cost_per_million=Decimal("1.5"), request_cost=Decimal("0.01"), source="provider_models_api", ) monkeypatch.setattr(pricing, "get_pricing_entry", lambda *_, **__: entry) usage = self._summary(10, 5, cache_read=2, cache_write=3) response_cost, summary_cost = self._run_both_paths(mod, monkeypatch, usage) for cost_details in (response_cost, summary_cost): component_sum = sum( value for key, value in cost_details.items() if key != "total" ) assert cost_details["total"] == pytest.approx(0.0100255) assert component_sum == pytest.approx(0.0000255) def test_request_only_price_still_exports_total(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") import agent.usage_pricing as pricing entry = pricing.PricingEntry( request_cost=Decimal("0.01"), source="provider_models_api", ) monkeypatch.setattr(pricing, "get_pricing_entry", lambda *_, **__: entry) usage = self._summary(0, 0) response_cost, summary_cost = self._run_both_paths(mod, monkeypatch, usage) assert response_cost == {"total": 0.01} assert summary_cost == {"total": 0.01} def test_partial_cache_pricing_exports_no_costs(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") import agent.usage_pricing as pricing entry = pricing.PricingEntry( input_cost_per_million=Decimal("1"), output_cost_per_million=Decimal("2"), cache_read_cost_per_million=None, source="provider_models_api", ) monkeypatch.setattr(pricing, "get_pricing_entry", lambda *_, **__: entry) usage = self._summary(10, 5, cache_read=2) response_cost, summary_cost = self._run_both_paths(mod, monkeypatch, usage) assert response_cost == {} assert summary_cost == {} def test_unknown_pricing_exports_no_costs(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") import agent.usage_pricing as pricing monkeypatch.setattr(pricing, "get_pricing_entry", lambda *_, **__: None) usage = self._summary(10, 5) response_cost, summary_cost = self._run_both_paths(mod, monkeypatch, usage) assert response_cost == {} assert summary_cost == {} def test_included_route_does_not_pin_total(self, monkeypatch): sys.modules.pop("plugins.observability.langfuse", None) mod = importlib.import_module("plugins.observability.langfuse") usage = self._summary(10, 5, cache_read=2) response_cost, summary_cost = self._run_both_paths( mod, monkeypatch, usage, provider="openai-codex", model="gpt-5.3-codex", api_mode="chat_completions", ) assert response_cost == summary_cost assert "total" not in response_cost # Subscription-included routes must send NO cost keys at all — # explicit zeros are treated as authoritative by Langfuse and block # its own model-based estimation (#43129). assert response_cost == {}