Route Langfuse timestamps through OTel tracer

The v4.7 Langfuse SDK exposes timestamp control only on its
internal OTel tracer, not on the public start_observation() API
that the adapter was calling. Two quirks the original Protocol
widening got wrong, both surfaced by live-account verification:

start_observation() rejects a start_time kwarg with TypeError.
When start_time is supplied, the adapter now mirrors the SDK's
own create_event precedent: open the OTel span directly via
_otel_tracer.start_span(name=, start_time=int_ns) within a
trace context and wrap the result in LangfuseGeneration. The
existing no-start_time path still uses the public API.

LangfuseSpan.end(end_time) is typed Optional[int] (nanoseconds),
not datetime. The adapter now converts the Protocol's datetime
surface to nanoseconds before forwarding. Without the conversion
the OTel span_processor's formatter crashes with TypeError on
end_time / 1e9 deep in the SDK.

Strengthen the unit tests: spy on both _otel_tracer.start_span
and start_observation so the back-dated path asserts the OTel
route is taken and the public-API path asserts the OTel tracer
is NOT touched. The previous monkeypatch test accepted **kwargs
and would have passed even with the broken implementation.

Widen the integration test's REST poll budget to 180s and use
server-side name+type filters; add a diagnostic that lists
observation names actually projected under the trace_id when
the target Generation doesn't appear, so a future name-mismatch
SDK change surfaces explicitly.

Author: chris-colinsky

src/openarmature/observability/langfuse/adapter.py

@@ -134,10 +134,14 @@ def end(self, *, end_time: datetime | None = None, **fields: Any) -> None:
         # Apply any field updates first (so they're set BEFORE the
         # observation closes), then call end(). v4's end() takes only
         # an optional ``end_time``; field mutation happens via update().
+        # The SDK's end_time is typed Optional[int] nanoseconds —
+        # convert from the Protocol's datetime surface before passing
+        # through. Without the conversion the OTel span_processor's
+        # formatter raises TypeError when it tries ``end_time / 1e9``.
         if fields:
             self.update(**fields)
         if end_time is not None:
-            self._obs.end(end_time=end_time)
+            self._obs.end(end_time=int(end_time.timestamp() * 1_000_000_000))
         else:
             self._obs.end()
 
@@ -362,19 +366,97 @@ def generation(
                 usage_details["total"] = usage.total
             extra_kwargs["usage_details"] = usage_details
         if start_time is not None:
-            extra_kwargs["start_time"] = start_time
-        obs = self._start_observation(
-            as_type="generation",
-            trace_id=trace_id,
-            name=name,
-            metadata=metadata,
-            parent_observation_id=parent_observation_id,
-            level=level,
-            status_message=status_message,
-            **{k: v for k, v in extra_kwargs.items() if v is not None},
-        )
+            # v4's public ``start_observation`` does NOT accept a
+            # ``start_time`` kwarg — only the internal OTel tracer
+            # does. Mirror the SDK's own ``create_event`` precedent
+            # (langfuse/_client/client.py:1518-1551): open the OTel
+            # span directly via the private ``_otel_tracer`` with the
+            # back-dated timestamp, then wrap it in LangfuseGeneration.
+            # This is the only path to a back-dated Generation in
+            # v4.7; the live-account integration test catches a future
+            # SDK break.
+            obs = self._start_back_dated_generation(
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                start_time=start_time,
+                **{k: v for k, v in extra_kwargs.items() if v is not None},
+            )
+        else:
+            obs = self._start_observation(
+                as_type="generation",
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                **{k: v for k, v in extra_kwargs.items() if v is not None},
+            )
         return _SpanHandle(obs)
 
+    def _start_back_dated_generation(
+        self,
+        *,
+        trace_id: str,
+        name: str | None,
+        metadata: dict[str, Any] | None,
+        parent_observation_id: str | None,
+        level: ObservationLevel,
+        status_message: str | None,
+        start_time: datetime,
+        **extra: Any,
+    ) -> Any:
+        """Open a LangfuseGeneration at a back-dated timestamp by going
+        through the private OTel tracer rather than the public
+        ``start_observation`` API (which doesn't accept ``start_time``
+        in v4.7). Mirrors the SDK's ``create_event`` precedent."""
+        from langfuse._client.span import LangfuseGeneration
+        from opentelemetry import trace as otel_trace_api
+
+        trace_entry = self._trace_info.get(trace_id)
+        trace_context: TraceContext = {"trace_id": _to_otel_trace_id(trace_id)}
+        if parent_observation_id is not None:
+            trace_context["parent_span_id"] = parent_observation_id
+
+        # OTel's ``start_span(start_time=...)`` takes int nanoseconds
+        # since epoch. The SDK uses ``time_ns()`` for its instant
+        # events; for back-dating, convert from the supplied datetime.
+        start_time_ns = int(start_time.timestamp() * 1_000_000_000)
+
+        remote_parent_span = self._client._create_remote_parent_span(  # pyright: ignore[reportPrivateUsage] # noqa: SLF001
+            trace_id=trace_context["trace_id"],
+            parent_span_id=trace_context.get("parent_span_id"),
+        )
+
+        with ExitStack() as stack:
+            if trace_entry is not None:
+                stack.enter_context(
+                    propagate_attributes(
+                        trace_name=trace_entry["name"],
+                        metadata=_stringify_metadata(trace_entry["metadata"]),
+                    )
+                )
+            stack.enter_context(otel_trace_api.use_span(remote_parent_span))
+            otel_span = self._client._otel_tracer.start_span(  # pyright: ignore[reportPrivateUsage] # noqa: SLF001
+                name=name or "observation",
+                start_time=start_time_ns,
+            )
+            generation_kwargs: dict[str, Any] = {
+                "otel_span": otel_span,
+                "langfuse_client": self._client,
+                "metadata": metadata,
+            }
+            if level != "DEFAULT":
+                generation_kwargs["level"] = level
+            if status_message is not None:
+                generation_kwargs["status_message"] = status_message
+            generation_kwargs.update(extra)
+            return LangfuseGeneration(**generation_kwargs)
+
     def force_flush(self, timeout_ms: int = 30_000) -> bool:
         """Best-effort flush of the underlying Langfuse client.
 

tests/integration/test_langfuse_sdk_adapter.py

@@ -185,30 +185,39 @@ async def test_sdk_adapter_generation_timestamps_round_trip_through_langfuse() -
     hex_id = invocation_id.replace("-", "")
     try:
         _poll_trace_with_retry(client, hex_id)
-        # Pull observations directly: the trace.observations list lags
-        # the headline trace fields in the REST projection, so query
-        # the observations endpoint with a retry budget similar to the
-        # trace poll. Filter by name rather than indexing the first
-        # entry so future synthetic observations (Langfuse adds
-        # ``openarmature.trace_io`` carriers for trace input/output
-        # under some flows) don't shadow the target Generation.
+        # Pull observations via REST. The observations-list endpoint
+        # lags the headline trace fields by an "indeterminate window"
+        # per Langfuse's own caveat (mirrored in the trace_io test
+        # above), so the retry budget here is wider (180s vs the
+        # trace_io test's 60s). Filter server-side by name + type so
+        # the response is small + scoped; track seen names across
+        # polls so a name-mismatch failure surfaces with diagnostics
+        # rather than a generic "not found".
         observation: Any = None
         last_exc: Exception | None = None
-        for _ in range(12):
+        seen_names: set[str] = set()
+        for _ in range(36):
             try:
-                response = client.api.observations.get_many(trace_id=hex_id)
-                observation = next(
-                    (o for o in response.data if o.name == "openarmature.llm.complete"),
-                    None,
+                response = client.api.observations.get_many(
+                    trace_id=hex_id,
+                    name="openarmature.llm.complete",
+                    type="GENERATION",
                 )
-                if observation is not None:
+                if response.data:
+                    observation = response.data[0]
                     break
+                # Fall back to an unfiltered query so we know whether
+                # ANY observations have projected — distinguishes "REST
+                # lag" from "observation projected with unexpected name".
+                fallback = client.api.observations.get_many(trace_id=hex_id)
+                seen_names.update(o.name or "<unnamed>" for o in fallback.data)
             except Exception as exc:  # noqa: BLE001
                 last_exc = exc
             time.sleep(5)
         assert observation is not None, (
-            f"openarmature.llm.complete observation for trace {hex_id} did not appear via REST "
-            f"after retry budget; last error: {last_exc!r}"
+            f"openarmature.llm.complete Generation for trace {hex_id} did not appear via REST "
+            f"after 180s retry budget. Observations seen under trace (any name): {seen_names or 'none'}. "
+            f"Last error: {last_exc!r}"
         )
 
         # The REST projection's start/end timestamps MUST match the

tests/unit/test_observability_langfuse_adapter.py

@@ -161,38 +161,93 @@ def test_adapter_force_flush_delegates_to_client() -> None:
     assert wrapped.flush.call_count == 2
 
 
-def test_adapter_generation_forwards_start_time_to_sdk(monkeypatch: pytest.MonkeyPatch) -> None:
-    # The typed-event LLM path back-dates the Generation observation
-    # using ``start_time = end_time - timedelta(milliseconds=
-    # latency_ms)``. The adapter MUST flow ``start_time`` through to
-    # the v4 SDK's ``start_observation(start_time=...)`` kwarg.
-    # Without this passthrough, Langfuse would show the call-time
-    # timestamp instead of the back-dated one — the verification gap
-    # the InMemory client can't catch on its own.
+def test_adapter_generation_routes_back_dated_calls_via_otel_tracer(monkeypatch: pytest.MonkeyPatch) -> None:
+    # v4.7's public ``Langfuse.start_observation`` does NOT accept
+    # ``start_time`` — only the internal ``_otel_tracer.start_span``
+    # does. The adapter MUST route back-dated generation() calls via
+    # the OTel tracer path (mirroring the SDK's own ``create_event``
+    # precedent). This test spies on BOTH paths: ``start_observation``
+    # is patched to fail loudly if the back-dated path ever falls
+    # through to it (the prior monkeypatch test's gap), and the OTel
+    # tracer's ``start_span`` is spied to assert the back-dated
+    # nanosecond timestamp lands on the right surface.
     from datetime import UTC, datetime
     from unittest.mock import MagicMock
 
+    client = _dummy_client()
+    captured_otel_kwargs: dict[str, Any] = {}
+
+    def _otel_spy(**kwargs: Any) -> MagicMock:
+        captured_otel_kwargs.update(kwargs)
+        # The SDK calls get_span_context(), set_attribute(), etc. on
+        # the returned span during LangfuseGeneration construction.
+        # Plain MagicMock auto-creates most attrs, but trace_id /
+        # span_id MUST be real ints because the SDK formats them as
+        # 32 / 16-char hex internally.
+        span = MagicMock()
+        span.get_span_context.return_value = MagicMock(
+            trace_id=int("a" * 32, 16),
+            span_id=int("b" * 16, 16),
+        )
+        return span
+
+    def _start_observation_should_not_be_called(**_kwargs: Any) -> None:
+        raise AssertionError(
+            "start_observation MUST NOT be called on the back-dated path; "
+            "v4 SDK rejects start_time= and the adapter should route via _otel_tracer"
+        )
+
+    monkeypatch.setattr(client._otel_tracer, "start_span", _otel_spy)  # noqa: SLF001
+    monkeypatch.setattr(client, "start_observation", _start_observation_should_not_be_called)
+    adapter = LangfuseSDKAdapter(client)
+    adapter.trace(id="trace-ts", name="t")
+
+    start = datetime(2026, 6, 8, 12, 0, 0, tzinfo=UTC)
+    adapter.generation(trace_id="trace-ts", name="g", model="m", start_time=start)
+
+    expected_ns = int(start.timestamp() * 1_000_000_000)
+    assert captured_otel_kwargs.get("start_time") == expected_ns
+    assert captured_otel_kwargs.get("name") == "g"
+
+
+def test_adapter_generation_without_start_time_uses_public_api(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Companion to the back-dated test: when ``start_time`` is NOT
+    # supplied, the adapter falls back to the v4 SDK's public
+    # ``start_observation`` API and does NOT touch the private OTel
+    # tracer.
+    from unittest.mock import MagicMock
+
     client = _dummy_client()
     captured_kwargs: dict[str, Any] = {}
 
     def _spy(**kwargs: Any) -> MagicMock:
         captured_kwargs.update(kwargs)
         return MagicMock(id="obs-spy", end=MagicMock())
 
+    def _otel_tracer_should_not_be_called(**_kwargs: Any) -> None:
+        raise AssertionError(
+            "_otel_tracer.start_span MUST NOT be called when start_time is None; "
+            "the public start_observation API should handle this path"
+        )
+
     monkeypatch.setattr(client, "start_observation", _spy)
+    monkeypatch.setattr(client._otel_tracer, "start_span", _otel_tracer_should_not_be_called)  # noqa: SLF001
     adapter = LangfuseSDKAdapter(client)
     adapter.trace(id="trace-ts", name="t")
 
-    start = datetime(2026, 6, 8, 12, 0, 0, tzinfo=UTC)
-    adapter.generation(trace_id="trace-ts", name="g", model="m", start_time=start)
+    adapter.generation(trace_id="trace-ts", name="g", model="m")
 
-    assert captured_kwargs.get("start_time") == start
+    assert captured_kwargs.get("name") == "g"
+    assert "start_time" not in captured_kwargs
 
 
-def test_adapter_generation_handle_end_forwards_end_time_to_sdk() -> None:
-    # Companion to the start_time test: the handle's end() MUST flow
-    # ``end_time`` through to the underlying v4 obs's ``end(end_time
-    # =...)`` call so the Langfuse UI shows the back-dated duration.
+def test_adapter_generation_handle_end_converts_end_time_to_nanoseconds() -> None:
+    # Companion to the start_time test: the handle's end() MUST
+    # convert the datetime to int nanoseconds before forwarding to
+    # the underlying v4 obs's end(). LangfuseSpan.end is typed
+    # ``Optional[int]`` (nanoseconds); passing a datetime through
+    # crashes the OTel span_processor's formatter with TypeError on
+    # ``end_time / 1e9`` deep in the SDK.
     from datetime import UTC, datetime
     from unittest.mock import MagicMock
 
@@ -205,7 +260,8 @@ def test_adapter_generation_handle_end_forwards_end_time_to_sdk() -> None:
     end = datetime(2026, 6, 8, 12, 0, 1, tzinfo=UTC)
     handle.end(end_time=end)
 
-    sdk_obs.end.assert_called_once_with(end_time=end)
+    expected_ns = int(end.timestamp() * 1_000_000_000)
+    sdk_obs.end.assert_called_once_with(end_time=expected_ns)
 
 
 def test_adapter_generation_handle_end_omits_end_time_when_unspecified() -> None: