Emit trace.input/output via Langfuse SDK adapter

CHANGELOG.md

@@ -28,7 +28,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The
 ### Notes
 
 - **Pinned spec version bumped from v0.31.0 to v0.35.0.** Absorbs proposals 0042 (reserved-key extension), 0043 (Langfuse trace.input/output sourcing), and the textual additions in v0.32.0 (Gemini wire-format mapping, 0038, not yet implemented) and v0.33.0 (sessions capability, 0020, not yet implemented).
-- The SDK adapter caches `input` / `output` in its `_trace_info` map; landing the values on the live Langfuse Trace from outside an active span context requires SDK-version-specific calls (v4's `langfuse.update_current_trace` works inside a context; cross-context REST updates need `client.api.trace.update`). The `InMemoryLangfuseClient` used by tests applies the fields directly. SDK-adapter end-to-end emit lands in a follow-up.
+- `LangfuseSDKAdapter` now applies `trace.input` / `trace.output` to the live Langfuse Trace. Input lands on the first real observation under the trace via `set_trace_io`; output uses a synthetic short-lived `openarmature.trace_io` observation as the carrier. The InMemoryLangfuseClient used by tests applies the fields directly.
+- The Langfuse v4 SDK marks `set_current_trace_io` / `Span.set_trace_io` deprecated ("removal in a future major version"). Empirical verification against Langfuse Cloud (v4.7.1, 2026-05-29) confirms it remains the **only** path that populates the Traces list view's headline `Input` / `Output` columns; `propagate_attributes(metadata=...)` does not substitute for it in the current UI. We will revisit when Langfuse publishes a concrete migration guide for v5.
 
 ## [0.10.0] — 2026-05-27
 

src/openarmature/observability/langfuse/adapter.py

@@ -226,38 +226,80 @@ def update_trace(
     ) -> None:
         # Merge into the trace_info cache so subsequent observations
         # (and the first one if not yet created) pick up the updated
-        # values. Since propagate_attributes runs on every observation
-        # using cached info, update_trace takes effect on the NEXT
-        # observation under this trace_id, not retroactively on prior
-        # observations.
+        # values. ``name`` / ``metadata`` are propagated via
+        # ``propagate_attributes`` around every observation under
+        # ``id``; ``input`` / ``output`` follow the SDK's
+        # ``set_trace_io`` path (per proposal 0043 + the
+        # empirically-validated v4.7.1 behaviour — see CHANGELOG).
         #
-        # Proposal 0043 ``input`` / ``output`` are cached but landing
-        # them on the live Langfuse Trace from outside an active
-        # span context is SDK-version-dependent (v4 exposes
-        # ``langfuse.update_current_trace(input=..., output=...)``
-        # only inside a context; cross-context REST updates need
-        # ``client.api.trace.update``). The InMemoryLangfuseClient
-        # surface used by tests applies them directly. The SDK
-        # adapter's apply path is a follow-up — caching here so the
-        # Protocol contract is satisfied without breaking SDK-adapter
-        # users.
+        # ``input`` is staged on the cache; applied to the FIRST real
+        # observation that opens under this trace_id (``_start_observation``
+        # below). Piggybacks on a real span so the trace tree gains no
+        # extra observation in the common case.
+        #
+        # ``output`` is applied immediately via a synthetic short-lived
+        # observation. By the time the LangfuseObserver dispatches the
+        # invocation-completed event all real spans have ended, so a
+        # synthetic span is the only path that has an active OTel span
+        # context for ``set_trace_io`` to find.
         entry = self._trace_info.get(id)
         if entry is None:
-            self._trace_info[id] = {
+            entry = {
                 "name": name,
                 "metadata": dict(metadata) if metadata is not None else {},
-                "input": input,
-                "output": output,
             }
-            return
-        if name is not None:
-            entry["name"] = name
-        if metadata is not None:
-            entry["metadata"].update(metadata)
+            self._trace_info[id] = entry
+        else:
+            if name is not None:
+                entry["name"] = name
+            if metadata is not None:
+                entry["metadata"].update(metadata)
         if input is not None:
-            entry["input"] = input
+            entry["pending_input"] = input
         if output is not None:
-            entry["output"] = output
+            self._emit_trace_output_synthetic(id, output)
+
+    def _emit_trace_output_synthetic(self, trace_id: str, output: Any) -> None:
+        # Open a synthetic short-lived observation, set
+        # ``trace.output`` on it via ``set_trace_io``, end immediately.
+        # The synthetic span shows in the trace as a small observation
+        # named ``openarmature.trace_io``; the value lands on the
+        # Langfuse Trace's ``output`` headline field through the
+        # ``langfuse.trace.output`` OTel attribute set inside.
+        #
+        # Edge case: if no real node observation ever opened for this
+        # trace (e.g., a resume-path validation failure aborted the
+        # invocation before any node fired), the cached ``pending_input``
+        # has no real span to piggyback on. Apply it here so the input
+        # still lands — the synthetic observation becomes the sole
+        # carrier for both fields. Pops the cache so we don't re-apply
+        # if ``update_trace`` is called more than once.
+        entry = self._trace_info.get(trace_id)
+        pending_input = entry.pop("pending_input", None) if entry is not None else None
+
+        trace_context: TraceContext = {"trace_id": _to_otel_trace_id(trace_id)}
+        with ExitStack() as stack:
+            if entry is not None:
+                stack.enter_context(
+                    propagate_attributes(
+                        trace_name=entry["name"],
+                        metadata=_stringify_metadata(entry["metadata"]),
+                    )
+                )
+            obs = cast(
+                "Any",
+                self._client.start_observation(
+                    name="openarmature.trace_io",
+                    as_type="span",
+                    trace_context=trace_context,
+                ),
+            )
+            try:
+                # Deprecation rationale on the equivalent call in
+                # ``_start_observation``.
+                obs.set_trace_io(input=pending_input, output=output)  # pyright: ignore[reportDeprecated]
+            finally:
+                obs.end()
 
     def span(
         self,
@@ -395,7 +437,25 @@ def _start_observation(
                         metadata=_stringify_metadata(trace_entry["metadata"]),
                     )
                 )
-            return cast("Any", self._client.start_observation(**kwargs))
+            obs = cast("Any", self._client.start_observation(**kwargs))
+            # Proposal 0043 (PR 8.5a): apply any pending ``trace.input``
+            # cached by ``update_trace`` to the FIRST real observation
+            # under this trace. ``set_trace_io`` needs an active OTel
+            # span context — piggybacking on the just-created
+            # observation is the lowest-overhead path. ``pop`` so
+            # subsequent observations under the same trace_id don't
+            # re-apply (the value is one-shot per trace).
+            #
+            # The Langfuse SDK marks ``set_trace_io`` deprecated as of
+            # v4.6 ("removal in a future major version"); per the
+            # empirical verification in PR 8.5a it remains the only
+            # path that surfaces ``trace.input`` in the Langfuse UI's
+            # Traces list view. See CHANGELOG for the deprecation note.
+            if trace_entry is not None:
+                pending_input = trace_entry.pop("pending_input", None)
+                if pending_input is not None:
+                    obs.set_trace_io(input=pending_input)  # pyright: ignore[reportDeprecated]
+            return obs
 
 
 __all__ = ["LangfuseSDKAdapter"]

tests/integration/test_langfuse_sdk_adapter.py

@@ -0,0 +1,138 @@
+"""Integration tests for LangfuseSDKAdapter against the live Langfuse
+test account.
+
+Gated by the presence of ``LANGFUSE_PUBLIC_KEY`` / ``LANGFUSE_SECRET_KEY``
+env vars. Skipped in CI and local runs that don't have credentials in
+scope; runs end-to-end against Langfuse Cloud when invoked from a
+shell with credentials (the documented test-account env vars per
+[[reference_langfuse_test_account.md]]).
+
+Each test polls the REST API after ``flush()`` with retries to absorb
+the eventual-consistency lag between ingestion and the REST projection.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+import uuid
+from typing import Any
+
+import pytest
+
+# Skip the entire module when credentials aren't sourced. Avoids a
+# cryptic ``ImportError`` / ``ValueError`` cascade from the SDK when
+# the test environment is bare.
+pytestmark = pytest.mark.skipif(
+    not (os.environ.get("LANGFUSE_PUBLIC_KEY") and os.environ.get("LANGFUSE_SECRET_KEY")),
+    reason="Requires LANGFUSE_PUBLIC_KEY + LANGFUSE_SECRET_KEY (live Langfuse account)",
+)
+
+
+def _poll_trace_with_retry(client: Any, hex_id: str, *, attempts: int = 12, sleep_s: float = 5.0) -> Any:
+    """Poll Langfuse's REST API until the trace appears or the budget
+    runs out. The Langfuse list-view UI updates faster than the REST
+    GET projection, so a freshly-flushed trace can 404 for ~30-60s.
+    Linear backoff is fine; the API is rate-limited gently and the
+    test runs once per CI invocation."""
+    from langfuse.api import NotFoundError
+
+    last_exc: Exception | None = None
+    for _ in range(attempts):
+        try:
+            return client.api.trace.get(hex_id)
+        except NotFoundError as exc:
+            last_exc = exc
+            time.sleep(sleep_s)
+    raise AssertionError(
+        f"Trace {hex_id} did not appear in REST API after {attempts * sleep_s:.0f}s; last error: {last_exc!r}"
+    )
+
+
+async def test_sdk_adapter_emits_trace_input_output_to_live_langfuse() -> None:
+    """End-to-end: open a trace via LangfuseSDKAdapter, push input via
+    update_trace at the start, push output via update_trace at the end,
+    flush, and confirm both fields populate on the live Trace entity."""
+    from langfuse import Langfuse
+
+    from openarmature.observability.langfuse.adapter import LangfuseSDKAdapter
+
+    client = Langfuse()
+    adapter = LangfuseSDKAdapter(client)
+
+    invocation_id = str(uuid.uuid4())
+    expected_input = {"entry_node": "verify_entry", "correlation_id": "test-corr-1"}
+    expected_output = {"final_node": "verify_entry", "status": "completed"}
+
+    # Simulate the LangfuseObserver call sequence: trace open →
+    # InvocationStartedEvent (update_trace with input) → first node
+    # span → node ends → InvocationCompletedEvent (update_trace with
+    # output).
+    adapter.trace(
+        id=invocation_id,
+        name="test_sdk_adapter_trace_io",
+        metadata={"test_run": "trace_io_emit"},
+    )
+    adapter.update_trace(id=invocation_id, input=expected_input)
+    # Open + close a real observation so the cached pending_input has
+    # something to piggyback on.
+    span_handle = adapter.span(trace_id=invocation_id, name="verify_entry")
+    span_handle.end()
+    adapter.update_trace(id=invocation_id, output=expected_output)
+
+    adapter.force_flush()
+    # Brief settle for the UI projection; REST poll handles the
+    # longer-tail consistency window separately.
+    time.sleep(2)
+
+    hex_id = invocation_id.replace("-", "")
+    trace = _poll_trace_with_retry(client, hex_id)
+
+    # `trace.input` / `trace.output` are the headline columns proposal
+    # 0043 motivates; assert both ingested correctly. These are the
+    # spec-compliance signal — they project off OTel attributes on
+    # incoming spans and populate on the Trace entity directly.
+    assert trace.input == expected_input, f"trace.input mismatch: got {trace.input!r}"
+    assert trace.output == expected_output, f"trace.output mismatch: got {trace.output!r}"
+    # Note: we deliberately do NOT assert on ``trace.observations``
+    # here. Langfuse's REST projection for the observations list
+    # lags the Trace's headline fields by an indeterminate window —
+    # the two tests in this module hit different consistency points
+    # against the same backend. The synthetic ``openarmature.trace_io``
+    # observation is verified in the next test (which uses ONLY the
+    # synthetic carrier and reliably shows in the list).
+
+
+async def test_sdk_adapter_handles_invocation_with_no_real_observation() -> None:
+    """Edge case: invocation fails before any node observation opens
+    (resume-path validation failure, etc.). The cached pending_input
+    has no real span to piggyback on, so the synthetic output
+    observation becomes the sole carrier for BOTH fields."""
+    from langfuse import Langfuse
+
+    from openarmature.observability.langfuse.adapter import LangfuseSDKAdapter
+
+    client = Langfuse()
+    adapter = LangfuseSDKAdapter(client)
+
+    invocation_id = str(uuid.uuid4())
+    expected_input = {"entry_node": "fail_fast", "correlation_id": "test-corr-2"}
+    expected_output = {"final_node": "fail_fast", "status": "failed"}
+
+    adapter.trace(id=invocation_id, name="test_sdk_adapter_no_real_span")
+    adapter.update_trace(id=invocation_id, input=expected_input)
+    # NO real span opens — straight to the output update.
+    adapter.update_trace(id=invocation_id, output=expected_output)
+
+    adapter.force_flush()
+    time.sleep(2)
+
+    hex_id = invocation_id.replace("-", "")
+    trace = _poll_trace_with_retry(client, hex_id)
+
+    # Both input and output land on the Trace even with the synthetic
+    # observation as the sole span.
+    assert trace.input == expected_input
+    assert trace.output == expected_output
+    assert len(trace.observations) == 1
+    assert trace.observations[0].name == "openarmature.trace_io"