fix(langfuse): keep host OTel resource out, carry big metadata ints, tolerate bad flush and TTL env, stamp trace I/O under a parent

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/langfuse/langfuse.py

@@ -885,6 +885,7 @@ class LangFuseLogger:
 
             resolved_trace_id: Final = resolve_trace_id(call_trace_id)  # pyright: ignore[reportArgumentType]  # metadata value, str or None at runtime
             continued_trace: Final = existing_trace_id is not None
+            generation_is_trace_root: Final = not continued_trace and parent_observation_id is None
             trace_public: Final = _trace_public_flag(trace_params.get("public"))
             trace_input: Final = trace_params.get("input")
             trace_output: Final = trace_params.get("output")
@@ -897,8 +898,10 @@ class LangFuseLogger:
                 tags=trace_params.get("tags"),
                 metadata=trace_params.get("metadata"),
                 public=trace_public,
-                input=trace_input if continued_trace or trace_input != generation_params["input"] else None,
-                output=trace_output if continued_trace or trace_output != generation_params["output"] else None,
+                input=None if generation_is_trace_root and trace_input == generation_params["input"] else trace_input,
+                output=None
+                if generation_is_trace_root and trace_output == generation_params["output"]
+                else trace_output,
             )
             generation_attributes: Final = observation_attributes(
                 observation_type="generation",
@@ -1060,19 +1063,19 @@ class LangFuseLogger:
 
     @staticmethod
     def _get_langfuse_flush_interval(flush_interval: int) -> int:
-        """
-        Get the langfuse flush interval to initialize the Langfuse client
-
-        Reads `LANGFUSE_FLUSH_INTERVAL` from the environment variable.
-        If not set, uses the flush interval passed in as an argument.
-
-        Args:
-            flush_interval: The flush interval to use if LANGFUSE_FLUSH_INTERVAL is not set
-
-        Returns:
-            [int] The flush interval to use to initialize the Langfuse client
-        """
-        return int(os.getenv("LANGFUSE_FLUSH_INTERVAL") or flush_interval)
+        """``LANGFUSE_FLUSH_INTERVAL`` in whole seconds above 0 (the export scheduler's delay), else ``flush_interval``."""
+        raw: Final = os.getenv("LANGFUSE_FLUSH_INTERVAL")
+        if not raw:
+            return flush_interval
+        parsed: Final = int(raw) if raw.strip().isdigit() else None
+        if parsed is None or parsed <= 0:
+            verbose_logger.warning(
+                "LANGFUSE_FLUSH_INTERVAL=%r is not a whole number of seconds above 0; flushing every %d s",
+                raw,
+                flush_interval,
+            )
+            return flush_interval
+        return parsed
 
     def _log_guardrail_information_as_span(
         self,

litellm/integrations/langfuse/langfuse_sdk.py

@@ -69,6 +69,9 @@ _DEFAULT_FLUSH_AT: Final = 512
 _CHANNEL_RETIRE_GRACE_SECONDS: Final = 60.0
 _DEFAULT_TIMEOUT_SECONDS: Final = 20.0
 _DEFAULT_MAX_RETRIES: Final = 3
+_DEFAULT_PROMPT_CACHE_TTL_SECONDS: Final = 60.0
+_INT64_MIN: Final = -(2**63)
+_INT64_MAX: Final = 2**63 - 1
 _COMMON_RELEASE_ENVS: Final = (
     "RENDER_GIT_COMMIT",
     "CI_COMMIT_SHA",
@@ -158,16 +161,22 @@ def _present(entries: Iterable[tuple[str, AttributeValue | None]]) -> Mapping[st
     return MappingProxyType({key: value for key, value in entries if value is not None})
 
 
+def _metadata_value(value: object) -> AttributeValue | None:
+    """A metadata value as OTLP can carry it: ints past int64 go as strings, as the v2 serializer sent them."""
+    if isinstance(value, (str, bool)):
+        return value
+    if isinstance(value, int) and _INT64_MIN <= value <= _INT64_MAX:
+        return value
+    return _serialize(value)
+
+
 def _flattened_metadata(prefix: str, metadata: object) -> Mapping[str, AttributeValue]:
     """Mirror the SDK's wire shape: one ``<prefix>.<key>`` attribute per key, or ``<prefix>`` for a non-dict."""
     if metadata is None:
         return _present(())
     if not isinstance(metadata, Mapping):
         return _present(((prefix, _serialize(metadata)),))
-    return _present(
-        (f"{prefix}.{key}", value if isinstance(value, (str, int)) else _serialize(value))
-        for key, value in metadata.items()
-    )
+    return _present((f"{prefix}.{key}", _metadata_value(value)) for key, value in metadata.items())
 
 
 def trace_attributes(
@@ -424,12 +433,16 @@ class TraceIdHashSampler(Sampler):
         return f"TraceIdHashSampler{{{self.rate}}}"
 
 
-def _parse_sample_rate(raw: str) -> float | None:
+def _parse_float(raw: str) -> float | None:
     try:
-        rate: Final = float(raw)
+        return float(raw)
     except ValueError:
         return None
-    return rate if 0.0 <= rate <= 1.0 else None
+
+
+def _parse_sample_rate(raw: str) -> float | None:
+    rate: Final = _parse_float(raw)
+    return rate if rate is not None and 0.0 <= rate <= 1.0 else None
 
 
 def configured_sample_rate() -> float:
@@ -474,6 +487,22 @@ def configured_release() -> str | None:
     )
 
 
+def configured_prompt_cache_ttl() -> float:
+    """``LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS``, the SDK's knob, with its 60 s default when unset or unusable."""
+    raw: Final = os.environ.get("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS")
+    if raw is None:
+        return _DEFAULT_PROMPT_CACHE_TTL_SECONDS
+    parsed: Final = _parse_float(raw)
+    if parsed is None or parsed < 0:
+        verbose_logger.warning(
+            "LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS=%r is not a number of seconds at or above 0; caching prompts for %.0f s",
+            raw,
+            _DEFAULT_PROMPT_CACHE_TTL_SECONDS,
+        )
+        return _DEFAULT_PROMPT_CACHE_TTL_SECONDS
+    return parsed
+
+
 def configured_flush_at() -> int:
     """``LANGFUSE_FLUSH_AT`` as the export batch size, the SDK's own knob, with its default when unset or unusable."""
     raw: Final = os.environ.get("LANGFUSE_FLUSH_AT")
@@ -608,7 +637,8 @@ def _build_span_exporter(*, public_key: str, secret_key: str, base_url: str) ->
 
 
 def _resource(*, environment: str | None, release: str | None) -> Resource:
-    return Resource.create(
+    """Only litellm's own attributes: ``Resource.create`` would merge the host's ``OTEL_RESOURCE_ATTRIBUTES``."""
+    return Resource(
         _present(
             (
                 (LangfuseOtelSpanAttributes.ENVIRONMENT, environment),
@@ -939,5 +969,5 @@ def build_langfuse_client(
             httpx_client=httpx_client,
             timeout=configured_timeout(),
         ),
-        prompt_cache_ttl_seconds=float(os.getenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", "60")),
+        prompt_cache_ttl_seconds=configured_prompt_cache_ttl(),
     )

tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py

@@ -38,6 +38,7 @@ from litellm.integrations.langfuse.langfuse_sdk import (
     build_langfuse_client,
     build_langfuse_tracing,
     configured_flush_at,
+    configured_prompt_cache_ttl,
     configured_sample_rate,
     flush_langfuse_tracing,
     observation_attributes,
@@ -511,6 +512,70 @@ def test_host_otel_span_limits_do_not_truncate_langfuse_observations(
     assert all(span.attributes[key] == "v" * 32 for key in attributes)
 
 
+def test_host_otel_resource_env_does_not_reach_the_langfuse_resource(monkeypatch: pytest.MonkeyPatch):
+    """``OTEL_RESOURCE_ATTRIBUTES`` and ``OTEL_SERVICE_NAME`` belong to the host's tracing; Langfuse files a
+    trace under any ``deployment.environment`` it finds on the resource, and v2 shipped no resource at all."""
+    monkeypatch.setenv("OTEL_RESOURCE_ATTRIBUTES", "team.secret.note=internal-only,deployment.environment=hijack")
+    monkeypatch.setenv("OTEL_SERVICE_NAME", "the-hosts-own-service")
+    tracing = build_langfuse_tracing(
+        exporter=InMemorySpanExporter(), environment="prod", release="r1", sample_rate=1.0, flush_interval_millis=10
+    )
+
+    assert dict(tracing.provider.resource.attributes) == {A.ENVIRONMENT: "prod", A.RELEASE: "r1"}
+
+
+@pytest.mark.parametrize(
+    ("value", "encoded"),
+    [
+        (2**63 - 1, ("int_value", 2**63 - 1)),
+        (2**63, ("string_value", str(2**63))),
+        (10**20, ("string_value", str(10**20))),
+        (-(2**63) - 1, ("string_value", str(-(2**63) - 1))),
+        (True, ("bool_value", True)),
+    ],
+    ids=["int64-max", "int64-max-plus-one", "huge", "int64-min-minus-one", "bool"],
+)
+def test_metadata_ints_past_int64_reach_the_wire_as_strings(value, encoded):
+    """OTLP carries int64 only and its encoder silently drops any attribute it cannot fit, while the
+    export still succeeds; v2's serializer sent such ints as strings, so the value has to survive."""
+    from opentelemetry.exporter.otlp.proto.common.trace_encoder import encode_spans
+
+    exporter = InMemorySpanExporter()
+    tracing = build_langfuse_tracing(
+        exporter=exporter, environment=None, release=None, sample_rate=1.0, flush_interval_millis=10
+    )
+    attributes = observation_attributes(observation_type="generation", metadata={"order_id": value, "sibling": "kept"})
+    _generation(tracing, attributes=attributes).end(CALL_END)
+    tracing.flush()
+
+    (encoded_span,) = encode_spans(exporter.get_finished_spans()).resource_spans[0].scope_spans[0].spans
+    wire = {kv.key: kv.value for kv in encoded_span.attributes}
+    order_id = wire[f"{A.OBSERVATION_METADATA}.order_id"]
+    carried = {
+        "int_value": order_id.int_value,
+        "string_value": order_id.string_value,
+        "bool_value": order_id.bool_value,
+    }
+    assert (order_id.WhichOneof("value"), carried[order_id.WhichOneof("value")]) == encoded
+    assert wire[f"{A.OBSERVATION_METADATA}.sibling"].string_value == "kept"
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [(None, 60.0), ("5", 5.0), ("0", 0.0), ("2.5", 2.5), ("-1", 60.0), ("abc", 60.0)],
+    ids=["unset", "whole", "zero", "fraction", "negative", "text"],
+)
+def test_prompt_cache_ttl_env_falls_back_instead_of_raising(monkeypatch: pytest.MonkeyPatch, raw, expected, caplog):
+    """A typo in the SDK's TTL knob used to raise out of logger construction and fail the request."""
+    if raw is None:
+        monkeypatch.delenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", raising=False)
+    else:
+        monkeypatch.setenv("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS", raw)
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        assert configured_prompt_cache_ttl() == expected
+    assert ("LANGFUSE_PROMPT_CACHE_DEFAULT_TTL_SECONDS" in caplog.text) is (raw in ("-1", "abc"))
+
+
 def test_many_metadata_keys_never_evict_the_generation_input_and_output():
     """OTel's default 128-attribute cap drops the earliest attributes, and v2 never capped metadata."""
     exporter = InMemorySpanExporter()

tests/test_litellm/integrations/test_langfuse.py

@@ -1,5 +1,6 @@
 import datetime
 import json
+import logging
 import threading
 import time
 import types
@@ -1934,6 +1935,28 @@ def test_update_trace_keys_input_output_reach_the_trace_even_under_a_parent(monk
     assert "the-output" in str(span.attributes["langfuse.trace.output"])
 
 
+def test_a_fresh_trace_under_a_callers_parent_still_carries_its_own_input_and_output():
+    """Langfuse copies I/O onto a trace only from its root observation; a caller's ``parent_observation_id``
+    makes the generation a child, so the trace-level fields v2 set on ``trace(...)`` must be stamped."""
+    rig = _steering_logger()
+
+    _, _, span = _emit(rig, metadata={"parent_observation_id": "0123456789abcdef"})
+
+    assert span.parent is not None
+    assert "the-input" in str(span.attributes["langfuse.trace.input"])
+    assert "the-output" in str(span.attributes["langfuse.trace.output"])
+
+
+def test_a_fresh_trace_root_leaves_the_duplicate_io_to_langfuse():
+    rig = _steering_logger()
+
+    _, _, span = _emit(rig, metadata={"trace_id": "a" * 32})
+
+    assert span.parent is None
+    assert "langfuse.trace.input" not in (span.attributes or {})
+    assert "the-input" in str(span.attributes["langfuse.observation.input"])
+
+
 def test_existing_trace_id_appends_without_claiming_trace_root():
     """Langfuse copies a root observation's name and I/O onto the trace, so a
     continuation that claimed root would rename the trace after every request;
@@ -2155,6 +2178,40 @@ def test_parse_langfuse_debug_only_enables_on_true_strings():
     assert langfuse_module.parse_langfuse_debug(None) is False
 
 
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [(None, 1), ("", 1), ("3", 3), ("0", 1), ("-5", 1), ("abc", 1)],
+    ids=["unset", "empty", "valid", "zero", "negative", "text"],
+)
+def test_flush_interval_env_falls_back_instead_of_failing_the_first_request(monkeypatch, raw, expected, caplog):
+    """The batch scheduler rejects a non-positive delay; v2's consumer thread accepted 0, so the value must
+    not raise out of the lazily built logger and take Langfuse logging down for the worker."""
+    if raw is None:
+        monkeypatch.delenv("LANGFUSE_FLUSH_INTERVAL", raising=False)
+    else:
+        monkeypatch.setenv("LANGFUSE_FLUSH_INTERVAL", raw)
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        assert LangFuseLogger._get_langfuse_flush_interval(1) == expected  # pyright: ignore[reportPrivateUsage]  # the parser under test
+    assert ("LANGFUSE_FLUSH_INTERVAL" in caplog.text) is (raw in ("0", "-5", "abc"))
+
+
+def test_zero_flush_interval_still_builds_a_working_export_channel(monkeypatch):
+    receiver = _OtlpReceiver()
+    monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-flush-zero-test")
+    monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-flush-zero-test")
+    monkeypatch.delenv("LANGFUSE_MOCK", raising=False)
+    monkeypatch.setenv("LANGFUSE_FLUSH_INTERVAL", "0")
+    monkeypatch.setattr(litellm, "initialized_langfuse_clients", litellm.initialized_langfuse_clients)
+
+    try:
+        logger = LangFuseLogger(langfuse_host=receiver.url)
+        _log_one_completion(logger)
+    finally:
+        receiver.close()
+
+    assert receiver.received == ["/api/public/otel/v1/traces"]
+
+
 def test_langfuse_debug_env_string_false_stays_off(monkeypatch):
     """LANGFUSE_DEBUG=false must not reach the v4 client as a truthy string.