feat(observability): export Langfuse v4 traces over OTLP

CHANGELOG.md

@@ -1,5 +1,24 @@
 # Changelog
 
+## Unreleased: Langfuse v4 trace export over OTLP
+
+Langfuse v4 ingests traces only over OTLP, and Langfuse Cloud removes the
+legacy `/api/public/ingestion` trace path on 2026-11-16.
+
+- `LangfuseClient::send_otlp` sends `x-langfuse-ingestion-version: 4` and
+  reports a 2xx answer with rejected spans (`partialSuccess.rejectedSpans`)
+  as an error.
+- New `trace_export::ingestion_otlp` (`ingestion_batch_to_otlp_spans`,
+  `ingestion_batch_otlp_requests`): converts a locally built ingestion batch
+  into v4 OTLP spans, with trace name, user, session, tags, metadata,
+  environment, release and version on every span and the trace input/output
+  on the root observation.
+- New `LangfuseClient::send_observations_otlp` / `send_batch_otlp` and
+  `GraphLangfuseExporter::send_observations_otlp` / `build_otlp_requests`.
+  The batch-path `send_observations` methods stay for legacy deployments.
+- `trace_export::otlp::chunk_otlp_spans` exposes the request chunking that
+  `otlp_requests` uses.
+
 ## Unreleased: the path-based transcript API is removed (tinyagents#453)
 
 **Breaking.** `tinyagents-session` no longer reads or writes a workspace

crates/tinyagents-graph/src/observability/README.md

@@ -49,6 +49,9 @@ so existing runs are unchanged.
   defaults its `traceId` to the run's `root_run_id`; because the graph-run span
   uses the same id scheme the harness exporter parents its agent run spans to,
   a graph run and the agent/tool runs its nodes spawn nest into one trace tree.
+  `send_observations_otlp` / `build_otlp_requests` export the same trace to
+  Langfuse v4 over OTLP (trace attributes on every span, deterministic span
+  ids); the batch methods remain for legacy-mode deployments.
 
 ## Persistence bridge
 

crates/tinyagents-graph/src/observability/langfuse.rs

@@ -18,6 +18,15 @@
 //! lands every graph step, node, model generation, and tool call under one
 //! Langfuse trace — full end-to-end telemetry including tool health.
 //!
+//! # Langfuse v4
+//!
+//! Langfuse v4 ingests traces only over OTLP. [`GraphLangfuseExporter::send_observations_otlp`]
+//! converts the batch with the harness `trace_export::ingestion_otlp`
+//! projection, which copies trace-level attributes onto every span and keeps
+//! the deterministic observation ids as span ids, so the unified-trace
+//! nesting above holds on v4 too. The batch methods remain for inspection and
+//! for Langfuse deployments still on the legacy path.
+//!
 //! The exporter is pull-based and best-effort by construction: build the batch
 //! from a completed (or in-progress) observation slice read back from a
 //! [`GraphEventJournal`](crate::GraphEventJournal), then send it. It does
@@ -33,6 +42,7 @@ use crate::observability::{GraphHealthSummary, GraphObservation};
 use crate::stream::GraphEvent;
 use tinyagents_harness::error::{Result, TinyAgentsError};
 use tinyagents_harness::ids::NodeId;
+use tinyagents_harness::observability::trace_export::ingestion_otlp::ingestion_batch_otlp_requests;
 use tinyagents_harness::observability::{LangfuseClient, LangfuseTraceConfig, clean_nulls, iso_ms};
 
 /// A FIFO queue of pending span starts, each `(observation index, ts_ms)`.
@@ -124,7 +134,44 @@ impl GraphLangfuseExporter {
         build_graph_langfuse_batch(trace, observations, self.span_metadata_fn.as_deref())
     }
 
-    /// Sends `observations` as one Langfuse ingestion batch.
+    /// Builds the Langfuse v4 OTLP/HTTP JSON requests for `observations`
+    /// without sending them: the trace of [`Self::build_ingestion_batch`],
+    /// with trace name, session, tags, metadata (including node health) and
+    /// release copied onto every span, under resource `service.name =
+    /// service` and scope `<service>.graph`.
+    pub fn build_otlp_requests(
+        &self,
+        trace: LangfuseTraceConfig,
+        observations: &[GraphObservation],
+        service: &str,
+    ) -> Result<Vec<Value>> {
+        let payload = self.build_ingestion_batch(trace, observations)?;
+        ingestion_batch_otlp_requests(&payload, service, &format!("{service}.graph"))
+    }
+
+    /// Sends `observations` to Langfuse v4 over OTLP
+    /// ([`Self::build_otlp_requests`] through
+    /// [`LangfuseClient::send_otlp`]), stopping at the first failed request.
+    /// Prefer this over [`Self::send_observations`]: Langfuse v4 no longer
+    /// accepts traces on the legacy batch path.
+    pub async fn send_observations_otlp(
+        &self,
+        trace: LangfuseTraceConfig,
+        observations: &[GraphObservation],
+        service: &str,
+    ) -> Result<Vec<Value>> {
+        let requests = self.build_otlp_requests(trace, observations, service)?;
+        let mut responses = Vec::with_capacity(requests.len());
+        for request in requests {
+            responses.push(self.client.send_otlp(request).await?);
+        }
+        Ok(responses)
+    }
+
+    /// Sends `observations` as one legacy Langfuse ingestion batch.
+    ///
+    /// Langfuse Cloud removes this path for traces on 2026-11-16; use
+    /// [`Self::send_observations_otlp`].
     pub async fn send_observations(
         &self,
         trace: LangfuseTraceConfig,

crates/tinyagents-graph/src/observability/langfuse/langfuse_tests.rs

@@ -450,3 +450,69 @@ fn transport_free_batch_uses_root_identity_without_an_explicit_trace() {
     assert!(find(batch, "span-create", "root-1:run:root-1").is_some());
     assert!(find(batch, "span-create", "root-1:node:b:1").is_some());
 }
+
+/// The string value of OTLP span attribute `key`, if present.
+fn otlp_attr<'a>(span: &'a Value, key: &str) -> Option<&'a str> {
+    span["attributes"]
+        .as_array()?
+        .iter()
+        .find(|item| item["key"] == key)?["value"]["stringValue"]
+        .as_str()
+}
+
+#[test]
+fn otlp_export_is_a_v4_trace_with_trace_attributes_on_every_span() {
+    let trace = LangfuseTraceConfig {
+        name: Some("flow.run:demo".into()),
+        session_id: Some("flow:demo".into()),
+        release: Some("1.0.0".into()),
+        tags: vec!["run:flow".into()],
+        ..Default::default()
+    };
+    let requests = exporter()
+        .build_otlp_requests(trace, &sample_run(), "openhuman")
+        .unwrap();
+    assert_eq!(requests.len(), 1);
+    let scope = &requests[0]["resourceSpans"][0]["scopeSpans"][0];
+    assert_eq!(scope["scope"]["name"], "openhuman.graph");
+    let spans = scope["spans"].as_array().unwrap();
+    for span in spans {
+        assert_eq!(
+            otlp_attr(span, "langfuse.trace.name"),
+            Some("flow.run:demo")
+        );
+        assert_eq!(otlp_attr(span, "langfuse.session.id"), Some("flow:demo"));
+        assert_eq!(otlp_attr(span, "langfuse.release"), Some("1.0.0"));
+        assert_eq!(
+            otlp_attr(span, "langfuse.trace.metadata.graph_id"),
+            Some("demo-graph")
+        );
+        assert!(otlp_attr(span, "langfuse.trace.metadata.health").is_some());
+    }
+
+    // One root: the structural graph-run span, everything else nests below it.
+    let roots: Vec<_> = spans
+        .iter()
+        .filter(|s| s.get("parentSpanId").is_none())
+        .collect();
+    assert_eq!(roots.len(), 1);
+    assert_eq!(roots[0]["name"], "demo-graph");
+
+    let by_name = |name: &str| spans.iter().find(|s| s["name"] == name).unwrap();
+    let (step, node_b) = (by_name("step 1"), by_name("b"));
+    assert_eq!(step["parentSpanId"], roots[0]["spanId"]);
+    assert_eq!(node_b["parentSpanId"], step["spanId"]);
+    assert_eq!(
+        otlp_attr(node_b, "langfuse.observation.level"),
+        Some("ERROR")
+    );
+    assert_eq!(
+        otlp_attr(node_b, "langfuse.observation.metadata.langgraph_node"),
+        Some("b")
+    );
+    let failed = by_name("run.failed");
+    assert_eq!(
+        otlp_attr(failed, "langfuse.observation.type"),
+        Some("event")
+    );
+}

crates/tinyagents-harness/src/observability/langfuse/README.md

@@ -19,17 +19,24 @@ natural export boundary (end of run, end of turn, or on a timer).
   backend), or `::from_env` (reads `LANGFUSE_BASE_URL`/`LANGFUSE_PUBLIC_KEY`/
   `LANGFUSE_SECRET_KEY`, or `TINYHUMANS_LANGFUSE_PROXY_URL`/
   `TINYHUMANS_AUTH_TOKEN` when the proxy URL is set).
-- `LangfuseClient::send_observations` — builds and sends a trace's ingestion
-  batch; `build_ingestion_batch` does the same without sending, for callers
+- `LangfuseClient::send_observations` — builds and sends a trace's legacy
+  ingestion batch (not accepted for traces by Langfuse v4); `build_ingestion_batch` does the same without sending, for callers
   that want to inspect or batch the payload themselves.
 - `LangfuseClient::create_score` / `build_score_batch` — attaches a
   post-hoc evaluation (`LangfuseScore`) to an already-exported trace or
   observation.
 - `LangfuseClient::send_otlp` / `otlp_endpoint` — posts an OTLP/HTTP JSON
   trace request (for example from `trace_export::otlp::otlp_requests`) to
   `/api/public/otel/v1/traces`, or `/telemetry/langfuse/otel/v1/traces`
-  through the proxy. OTLP is Langfuse's supported trace path; the legacy
-  ingestion API is sunset on Langfuse Cloud on 2026-11-16.
+  through the proxy, with `x-langfuse-ingestion-version: 4`
+  (`LANGFUSE_INGESTION_VERSION_HEADER`) to select Langfuse v4's write path.
+  A 2xx answer whose OTLP `partialSuccess.rejectedSpans` is non-zero is an
+  error. OTLP is Langfuse's supported trace path; the legacy ingestion API is
+  sunset on Langfuse Cloud on 2026-11-16.
+- `LangfuseClient::send_observations_otlp` / `send_batch_otlp` — the v4
+  export: build the same trace as `build_ingestion_batch`, convert it with
+  `trace_export::ingestion_otlp` and send each request through `send_otlp`.
+  Use these instead of `send_observations`.
 - `LangfuseClient::send_batch` — the shared HTTP transport (auth header,
   endpoint normalization, `207 Multi-Status` handling); reused by other
   exporters (e.g. `graph::observability`) that build their own `{"batch":

crates/tinyagents-harness/src/observability/langfuse/mod.rs

@@ -1,8 +1,12 @@
 //! Langfuse ingestion exporter for durable harness observations.
 //!
-//! The exporter targets Langfuse's `/api/public/ingestion` batch API. It can
-//! send directly to a self-hosted Langfuse instance with public/secret keys, or
-//! to the TinyHumans backend proxy with a bearer token.
+//! The exporter builds a Langfuse ingestion batch (`trace-create` plus one
+//! event per observation) and sends it either over OTLP to Langfuse v4's
+//! `/api/public/otel/v1/traces` (`send_observations_otlp`, `send_batch_otlp`,
+//! `send_otlp`), or to the legacy `/api/public/ingestion` batch API, which
+//! Langfuse v4 keeps only for scores. It can send directly to a self-hosted
+//! Langfuse instance with public/secret keys, or to the TinyHumans backend
+//! proxy with a bearer token.
 
 use std::collections::BTreeMap;
 
@@ -20,6 +24,11 @@ pub use types::{
     LangfuseAuth, LangfuseClient, LangfuseScore, LangfuseScoreValue, LangfuseTraceConfig,
 };
 
+/// Header that selects Langfuse's ingestion write path for an OTLP request.
+pub const LANGFUSE_INGESTION_VERSION_HEADER: &str = "x-langfuse-ingestion-version";
+/// The Langfuse v4 (observations-first) ingestion version.
+pub const LANGFUSE_INGESTION_VERSION: &str = "4";
+
 impl LangfuseClient {
     /// Creates a client from a base URL and auth mode.
     ///
@@ -127,12 +136,25 @@ impl LangfuseClient {
     /// such as one built by `trace_export::otlp::otlp_requests`, to
     /// [`Self::otlp_endpoint`] with this client's authentication.
     ///
+    /// The request carries `x-langfuse-ingestion-version: 4`, which selects
+    /// Langfuse v4's observations-first write path (the header does not make a
+    /// span v4-shaped; the `trace_export` converters do that).
+    ///
     /// # Errors
     ///
-    /// Returns an error when the request cannot be sent or Langfuse answers
-    /// with a non-success status.
+    /// Returns an error when the request cannot be sent, Langfuse answers
+    /// with a non-success status, or a 2xx answer reports rejected spans in
+    /// its OTLP `partialSuccess`.
     pub async fn send_otlp(&self, body: Value) -> Result<Value> {
-        let req = self.authorized(self.client.post(self.otlp_endpoint()).json(&body));
+        let req = self.authorized(
+            self.client
+                .post(self.otlp_endpoint())
+                .header(
+                    LANGFUSE_INGESTION_VERSION_HEADER,
+                    LANGFUSE_INGESTION_VERSION,
+                )
+                .json(&body),
+        );
         let response = req
             .send()
             .await
@@ -147,6 +169,18 @@ impl LangfuseClient {
                 "Langfuse OTLP ingestion returned {status}: {parsed}"
             )));
         }
+        let partial = &parsed["partialSuccess"];
+        // OTLP/JSON encodes int64 as a string; accept either spelling.
+        let rejected = partial["rejectedSpans"]
+            .as_u64()
+            .or_else(|| partial["rejectedSpans"].as_str()?.parse().ok())
+            .unwrap_or(0);
+        if rejected > 0 {
+            let message = partial["errorMessage"].as_str().unwrap_or_default();
+            return Err(TinyAgentsError::Model(format!(
+                "Langfuse OTLP ingestion rejected {rejected} span(s): {message}"
+            )));
+        }
         Ok(parsed)
     }
 
@@ -267,7 +301,54 @@ impl LangfuseClient {
         self.send_batch(payload).await
     }
 
-    /// Sends observations as one Langfuse ingestion batch.
+    /// Sends observations to Langfuse v4 over OTLP: builds the same trace as
+    /// [`Self::build_ingestion_batch`] and exports it through
+    /// [`Self::send_batch_otlp`]. Prefer this over [`Self::send_observations`],
+    /// whose legacy batch path Langfuse Cloud removes on 2026-11-16.
+    ///
+    /// # Errors
+    ///
+    /// As [`Self::build_ingestion_batch`] and [`Self::send_batch_otlp`].
+    pub async fn send_observations_otlp(
+        &self,
+        trace: LangfuseTraceConfig,
+        observations: &[AgentObservation],
+        service: &str,
+    ) -> Result<Vec<Value>> {
+        let payload = self.build_ingestion_batch(trace, observations)?;
+        self.send_batch_otlp(&payload, service).await
+    }
+
+    /// Exports a locally built ingestion `payload` (one `trace-create` plus its
+    /// completed observations) to Langfuse v4 over OTLP, converted by
+    /// [`ingestion_batch_otlp_requests`](crate::observability::trace_export::ingestion_otlp::ingestion_batch_otlp_requests)
+    /// under resource `service.name = service` and scope `<service>.trace`.
+    /// Requests are sent in order and the first failure stops the export.
+    /// Non-trace events (scores) in the payload are skipped; send those with
+    /// [`Self::send_batch`].
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when the payload has no trace, or any request fails
+    /// as described on [`Self::send_otlp`].
+    pub async fn send_batch_otlp(&self, payload: &Value, service: &str) -> Result<Vec<Value>> {
+        let requests =
+            crate::observability::trace_export::ingestion_otlp::ingestion_batch_otlp_requests(
+                payload,
+                service,
+                &format!("{service}.trace"),
+            )?;
+        let mut responses = Vec::with_capacity(requests.len());
+        for request in requests {
+            responses.push(self.send_otlp(request).await?);
+        }
+        Ok(responses)
+    }
+
+    /// Sends observations as one legacy Langfuse ingestion batch.
+    ///
+    /// Langfuse v4 no longer accepts trace payloads on this path (Langfuse
+    /// Cloud removes it on 2026-11-16); use [`Self::send_observations_otlp`].
     pub async fn send_observations(
         &self,
         trace: LangfuseTraceConfig,