diff --git a/plugins/observability/langfuse/__init__.py b/plugins/observability/langfuse/__init__.py index 6739916b96..cab4598d5a 100644 --- a/plugins/observability/langfuse/__init__.py +++ b/plugins/observability/langfuse/__init__.py @@ -633,9 +633,11 @@ def _finalize_all_traces() -> None: _end_children(state, include_subagents=True) _end_root(state, f"atexit finalize for {key}") if states: - # atexit runs unscoped; flush every profile's client, not just the launch profile's. - for client in (_get_langfuse(), *_LANGFUSE_CLIENT_BY_HOME.values()): - if client is not _INIT_FAILED: + # atexit runs with NO profile scope, so it must never build a client (a credential read + # here raises UnscopedSecretError under multiplex and would skip every flush). Flush only + # the clients that settled during the run — the launch profile's slot plus one per home. + for client in (_LANGFUSE_CLIENT, *_LANGFUSE_CLIENT_BY_HOME.values()): + if client is not None and client is not _INIT_FAILED: _flush(client) diff --git a/tests/plugins/test_langfuse_plugin.py b/tests/plugins/test_langfuse_plugin.py index 7772b0477f..3887491d0a 100644 --- a/tests/plugins/test_langfuse_plugin.py +++ b/tests/plugins/test_langfuse_plugin.py @@ -1741,6 +1741,55 @@ class TestAtexitFinalization(TestTurnTraceIsolation): 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)