diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f2263cfb..ed7969768 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/tinyagents-graph/src/observability/README.md b/crates/tinyagents-graph/src/observability/README.md index eaae42695..e00ce4fea 100644 --- a/crates/tinyagents-graph/src/observability/README.md +++ b/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 diff --git a/crates/tinyagents-graph/src/observability/langfuse.rs b/crates/tinyagents-graph/src/observability/langfuse.rs index 38aa912b7..7edaa9484 100644 --- a/crates/tinyagents-graph/src/observability/langfuse.rs +++ b/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 `.graph`. + pub fn build_otlp_requests( + &self, + trace: LangfuseTraceConfig, + observations: &[GraphObservation], + service: &str, + ) -> Result> { + 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> { + 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, diff --git a/crates/tinyagents-graph/src/observability/langfuse/langfuse_tests.rs b/crates/tinyagents-graph/src/observability/langfuse/langfuse_tests.rs index 06c7dc8fc..4b960213c 100644 --- a/crates/tinyagents-graph/src/observability/langfuse/langfuse_tests.rs +++ b/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") + ); +} diff --git a/crates/tinyagents-harness/src/observability/langfuse/README.md b/crates/tinyagents-harness/src/observability/langfuse/README.md index 17cf44217..c6486e2df 100644 --- a/crates/tinyagents-harness/src/observability/langfuse/README.md +++ b/crates/tinyagents-harness/src/observability/langfuse/README.md @@ -19,8 +19,8 @@ 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 @@ -28,8 +28,15 @@ natural export boundary (end of run, end of turn, or on a timer). - `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": diff --git a/crates/tinyagents-harness/src/observability/langfuse/mod.rs b/crates/tinyagents-harness/src/observability/langfuse/mod.rs index 81c0a479c..faa23ea5f 100644 --- a/crates/tinyagents-harness/src/observability/langfuse/mod.rs +++ b/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 { - 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> { + 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 `.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> { + 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, diff --git a/crates/tinyagents-harness/src/observability/langfuse/mod_tests.rs b/crates/tinyagents-harness/src/observability/langfuse/mod_tests.rs index 808095d5c..11307fe5a 100644 --- a/crates/tinyagents-harness/src/observability/langfuse/mod_tests.rs +++ b/crates/tinyagents-harness/src/observability/langfuse/mod_tests.rs @@ -631,9 +631,22 @@ fn derives_the_otlp_trace_endpoint_beside_the_ingestion_endpoint() { ); } -/// Accepts one HTTP request on a loopback port, answers `status`, and returns -/// the request line, the authorization header and the decoded body. -fn serve_once(status: u16) -> (String, std::sync::mpsc::Receiver<(String, String, Value)>) { +/// One request captured by [`serve_once_with`]: the request line, the +/// authorization header, the `x-langfuse-ingestion-version` header and the +/// decoded body. +type Captured = (String, String, String, Value); + +/// Accepts one HTTP request on a loopback port, answers `status` with an empty +/// JSON object, and returns what it received. +fn serve_once(status: u16) -> (String, std::sync::mpsc::Receiver) { + serve_once_with(status, "{}") +} + +/// [`serve_once`] answering with `response` as the JSON body. +fn serve_once_with( + status: u16, + response: &'static str, +) -> (String, std::sync::mpsc::Receiver) { use std::io::{BufRead, BufReader, Read, Write}; let socket = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let base = format!("http://{}", socket.local_addr().unwrap()); @@ -643,7 +656,7 @@ fn serve_once(status: u16) -> (String, std::sync::mpsc::Receiver<(String, String let mut reader = BufReader::new(stream); let mut line = String::new(); reader.read_line(&mut line).unwrap(); - let (mut auth, mut len) = (String::new(), 0_usize); + let (mut auth, mut version, mut len) = (String::new(), String::new(), 0_usize); loop { let mut header = String::new(); reader.read_line(&mut header).unwrap(); @@ -653,6 +666,7 @@ fn serve_once(status: u16) -> (String, std::sync::mpsc::Receiver<(String, String let (name, value) = header.split_once(':').unwrap(); match name.trim().to_ascii_lowercase().as_str() { "authorization" => auth = value.trim().to_owned(), + "x-langfuse-ingestion-version" => version = value.trim().to_owned(), "content-length" => len = value.trim().parse().unwrap(), _ => {} } @@ -662,12 +676,14 @@ fn serve_once(status: u16) -> (String, std::sync::mpsc::Receiver<(String, String let mut stream = reader.into_inner(); write!( stream, - "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: 2\r\nconnection: close\r\n\r\n{{}}" + "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{response}", + response.len() ) .unwrap(); tx.send(( line.trim().to_owned(), auth, + version, serde_json::from_slice(&body).unwrap(), )) .unwrap(); @@ -681,12 +697,55 @@ async fn sends_otlp_json_to_the_trace_endpoint_with_basic_auth() { let client = LangfuseClient::direct(base, "pk-test", "sk-test").unwrap(); let body = json!({"resourceSpans": []}); client.send_otlp(body.clone()).await.unwrap(); - let (line, auth, received) = rx.recv().unwrap(); + let (line, auth, _, received) = rx.recv().unwrap(); assert_eq!(line, "POST /api/public/otel/v1/traces HTTP/1.1"); assert_eq!(auth, "Basic cGstdGVzdDpzay10ZXN0"); assert_eq!(received, body); } +#[tokio::test] +async fn otlp_requests_select_the_langfuse_v4_ingestion_path() { + for proxied in [false, true] { + let (base, rx) = serve_once(200); + let client = if proxied { + LangfuseClient::proxy(base, "token").unwrap() + } else { + LangfuseClient::direct(base, "pk", "sk").unwrap() + }; + client + .send_otlp(json!({"resourceSpans": []})) + .await + .unwrap(); + let (_, _, version, _) = rx.recv().unwrap(); + assert_eq!(version, "4", "proxied={proxied}"); + } +} + +#[tokio::test] +async fn reports_spans_the_otlp_endpoint_partially_rejected() { + let (base, _rx) = serve_once_with( + 200, + r#"{"partialSuccess":{"rejectedSpans":2,"errorMessage":"bad span"}}"#, + ); + let client = LangfuseClient::direct(base, "pk", "sk").unwrap(); + let err = client + .send_otlp(json!({"resourceSpans": []})) + .await + .unwrap_err() + .to_string(); + assert!(err.contains("rejected 2"), "{err}"); +} + +#[tokio::test] +async fn an_empty_partial_success_is_a_success() { + let (base, _rx) = serve_once_with(200, r#"{"partialSuccess":{}}"#); + let client = LangfuseClient::direct(base, "pk", "sk").unwrap(); + client + .send_otlp(json!({"resourceSpans": []})) + .await + .unwrap(); +} + #[tokio::test] async fn reports_a_refused_otlp_request() { let (base, _rx) = serve_once(413); @@ -698,3 +757,58 @@ async fn reports_a_refused_otlp_request() { .is_err() ); } + +#[tokio::test] +async fn sends_observations_as_v4_otlp_spans() { + let (base, rx) = serve_once(200); + let client = LangfuseClient::direct(base, "pk", "sk").unwrap(); + let trace = LangfuseTraceConfig { + trace_id: Some("trace-otlp".into()), + name: Some("agent.turn".into()), + session_id: Some("thread-1".into()), + ..Default::default() + }; + let responses = client + .send_observations_otlp( + trace, + &[ + obs( + 0, + AgentEvent::RunStarted { + run_id: RunId::new("run-1"), + thread_id: None, + }, + ), + obs( + 1, + AgentEvent::ModelCompleted { + call_id: CallId::new("model-call"), + started_at_ms: None, + usage: None, + input: None, + output: None, + }, + ), + ], + "tinyagents", + ) + .await + .unwrap(); + assert_eq!(responses.len(), 1); + let (line, _, version, body) = rx.recv().unwrap(); + assert_eq!(line, "POST /api/public/otel/v1/traces HTTP/1.1"); + assert_eq!(version, "4"); + let spans = body["resourceSpans"][0]["scopeSpans"][0]["spans"] + .as_array() + .unwrap(); + assert!(!spans.is_empty()); + for span in spans { + let session = span["attributes"] + .as_array() + .unwrap() + .iter() + .find(|attr| attr["key"] == "langfuse.session.id") + .map(|attr| attr["value"]["stringValue"].clone()); + assert_eq!(session, Some(json!("thread-1"))); + } +} diff --git a/crates/tinyagents-harness/src/observability/trace_export/README.md b/crates/tinyagents-harness/src/observability/trace_export/README.md index 169649ce3..91c0743cf 100644 --- a/crates/tinyagents-harness/src/observability/trace_export/README.md +++ b/crates/tinyagents-harness/src/observability/trace_export/README.md @@ -27,8 +27,16 @@ Product identity stays out of the crate: the host passes an `ExportBrand` first, with an omitted-messages marker), `truncate_chars`, `truncate_capture_text`, the small `json_*` value builders, `status_of`, and `spans_to_ndjson` with its `SpanEnvelope`. -- `otlp`: `span_to_otlp`, `otlp_requests`, `prepare_subagent_root`, - `compact_internal_searches`. +- `otlp`: `span_to_otlp`, `otlp_requests`, `chunk_otlp_spans`, + `prepare_subagent_root`, `compact_internal_searches`. +- `ingestion_otlp`: `ingestion_batch_to_otlp_spans` and + `ingestion_batch_otlp_requests`, the Langfuse v4 projection of a locally + built ingestion batch. Trace name, user, session, tags, metadata, + environment, release and version go on every span (v4 filters observations + directly, so a root-only value is invisible on a child); trace input/output + move to the root observation; parents ride the OTEL span context with span + ids hashed from the batch's deterministic observation ids. Scores and other + non-trace events are skipped. - `ingestion_batch`: `split_ingestion_batch` (bounded by `LANGFUSE_MAX_BATCH_EVENTS`), `iso_millis`, `new_event_id`. - `journal_export` (feature `langfuse`): `RunTotals`, @@ -48,5 +56,9 @@ Product identity stays out of the crate: the host passes an `ExportBrand` exceeds the budget, a bounded preview of it is kept. - Langfuse caps the events per ingestion request, so payloads go through `split_ingestion_batch` before sending. +- Langfuse v4 ingests traces only over OTLP (`/api/public/otel/v1/traces` with + `x-langfuse-ingestion-version: 4`); Langfuse Cloud removes the legacy + `/api/public/ingestion` trace path on 2026-11-16. v4 duplicates a span id it + receives twice instead of merging, so export each finished span once. - Callers decide whether content capture is on at all; these helpers do not redact secrets. Run them on events that already passed a `RedactingSink`. diff --git a/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp.rs b/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp.rs new file mode 100644 index 000000000..68198b40d --- /dev/null +++ b/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp.rs @@ -0,0 +1,285 @@ +//! Langfuse v4 OTLP projection of a locally built ingestion batch. +//! +//! Langfuse v4 drops the legacy `/api/public/ingestion` trace path: a trace is +//! no longer a record of its own, only the OTEL trace id its observations +//! share, and every observation is written once to an immutable table. This +//! module converts a `{"batch": [...]}` payload that an exporter assembled in +//! memory (one `trace-create` plus completed `span-create` / +//! `event-create` / `generation-create` events, as the graph and harness +//! exporters build it) into OTLP spans that follow the v4 attribute rules: +//! +//! - trace name, user, session, tags, metadata, environment, release and +//! version are copied onto **every** span, because v4 filters observations +//! directly and a root-only value is invisible on a child; +//! - trace input/output move to the root observation +//! (`langfuse.observation.input` / `langfuse.observation.output`); +//! - parentage rides the OTEL parent span context, and span ids are derived +//! deterministically from the batch's observation ids, so a run exported in +//! another request still nests under the same parent. +//! +//! The conversion is pure. Events of any other type (`score-create`, +//! `*-update`, `sdk-log`) are not traces in v4 and are skipped; scores keep +//! using the Langfuse scores API. Each observation in the batch must already be +//! final: v4 does not merge a re-exported span id, it duplicates the +//! observation, so a caller exports each span once. + +use serde_json::{Map, Value, json}; + +use super::otlp::{attribute, chunk_otlp_spans, hex_id, json_attribute}; +use crate::error::{Result, TinyAgentsError}; + +/// The `langfuse.observation.type` for a batch event type, or `None` when the +/// event is not an observation (trace, score, update, log). +fn observation_type(event_type: &str) -> Option<&'static str> { + match event_type { + "span-create" => Some("span"), + "event-create" => Some("event"), + "generation-create" => Some("generation"), + _ => None, + } +} + +/// A metadata key as an OTEL attribute suffix: dots would read as nesting. +fn metadata_key(key: &str) -> String { + key.replace('.', "_") +} + +/// A scalar string stays as written; anything else is serialized as JSON, +/// which is what Langfuse expects for structured attribute values. +fn value_attribute(key: &str, value: &Value) -> Value { + match value { + Value::String(text) => attribute(key, text.clone()), + other => json_attribute(key, other), + } +} + +/// `key = value` attributes for each entry of a metadata object, under +/// `prefix.`. A non-object value lands under `prefix` itself. +fn metadata_attributes(prefix: &str, metadata: Option<&Value>, out: &mut Vec) { + match metadata { + Some(Value::Object(map)) => { + for (key, value) in map { + if !value.is_null() { + out.push(value_attribute( + &format!("{prefix}.{}", metadata_key(key)), + value, + )); + } + } + } + Some(Value::Null) | None => {} + Some(other) => out.push(value_attribute(prefix, other)), + } +} + +/// An ISO-8601 batch timestamp as OTLP unix nanoseconds. +fn iso_to_nanos(value: Option<&Value>) -> Option { + let text = value?.as_str()?; + let parsed = chrono::DateTime::parse_from_rfc3339(text).ok()?; + u64::try_from(parsed.timestamp_nanos_opt()?).ok() +} + +/// The trace-level attributes v4 needs on every observation. +fn trace_attributes(trace: &Map) -> Vec { + let mut attrs = Vec::new(); + let text = |key: &str| trace.get(key).and_then(Value::as_str); + if let Some(name) = text("name") { + attrs.push(attribute("langfuse.trace.name", name)); + } + if let Some(user) = text("userId") { + attrs.push(attribute("langfuse.user.id", user)); + } + if let Some(session) = text("sessionId") { + attrs.push(attribute("langfuse.session.id", session)); + } + for (field, key) in [ + ("environment", "langfuse.environment"), + ("release", "langfuse.release"), + ("version", "langfuse.version"), + ] { + if let Some(value) = text(field) { + attrs.push(attribute(key, value)); + } + } + if let Some(tags) = trace.get("tags").and_then(Value::as_array) { + let values: Vec = tags + .iter() + .filter_map(Value::as_str) + .map(|tag| json!({ "stringValue": tag })) + .collect(); + if !values.is_empty() { + attrs.push(json!({ + "key": "langfuse.trace.tags", + "value": { "arrayValue": { "values": values } }, + })); + } + } + metadata_attributes("langfuse.trace.metadata", trace.get("metadata"), &mut attrs); + attrs +} + +/// One observation event as an OTLP span. `root_io` carries the trace's +/// input/output onto the root observation when that span has none of its own. +fn observation_span( + trace_hex: &str, + kind: &str, + body: &Map, + trace_attrs: &[Value], + root_io: Option<(&Value, &Value)>, +) -> Option { + let id = body.get("id")?.as_str()?; + let start = iso_to_nanos(body.get("startTime")).or_else(|| iso_to_nanos(body.get("timestamp"))); + let start = start?; + let end = iso_to_nanos(body.get("endTime")); + + let mut attrs = vec![attribute("langfuse.observation.type", kind)]; + attrs.extend(trace_attrs.iter().cloned()); + + let present = |key: &str| body.get(key).filter(|value| !value.is_null()); + let input = present("input").or(root_io.map(|(input, _)| input).filter(|v| !v.is_null())); + let output = present("output").or(root_io.map(|(_, output)| output).filter(|v| !v.is_null())); + if let Some(input) = input { + attrs.push(json_attribute("langfuse.observation.input", input)); + } + if let Some(output) = output { + attrs.push(json_attribute("langfuse.observation.output", output)); + } + let level = present("level").and_then(Value::as_str); + let status_message = present("statusMessage").and_then(Value::as_str); + if let Some(level) = level { + attrs.push(attribute("langfuse.observation.level", level)); + } + if let Some(message) = status_message { + attrs.push(attribute("langfuse.observation.status_message", message)); + } + if let Some(version) = present("version").and_then(Value::as_str) { + attrs.push(attribute("langfuse.version", version)); + } + if let Some(model) = present("model").and_then(Value::as_str) { + attrs.push(attribute("langfuse.observation.model.name", model)); + } + for (field, key) in [ + ("modelParameters", "langfuse.observation.model.parameters"), + ("usageDetails", "langfuse.observation.usage_details"), + ("costDetails", "langfuse.observation.cost_details"), + ] { + if let Some(value) = present(field) { + attrs.push(json_attribute(key, value)); + } + } + if present("usageDetails").is_none() + && let Some(usage) = present("usage") + { + attrs.push(json_attribute("langfuse.observation.usage_details", usage)); + } + if let Some(at) = present("completionStartTime").and_then(Value::as_str) { + attrs.push(attribute("langfuse.observation.completion_start_time", at)); + } + metadata_attributes( + "langfuse.observation.metadata", + present("metadata"), + &mut attrs, + ); + if end.is_none() && kind != "event" { + // A span still open when the batch was built: v4 exports a span once, + // so it goes out closed at its start and is flagged for readers. + attrs.push(attribute( + "langfuse.observation.metadata.unfinished", + "true", + )); + } + + let name = body.get("name").and_then(Value::as_str).unwrap_or(kind); + let mut span = json!({ + "traceId": trace_hex, + "spanId": hex_id(id, 8), + "name": name, + "startTimeUnixNano": start.to_string(), + "endTimeUnixNano": end.unwrap_or(start).max(start).to_string(), + "attributes": attrs, + }); + if let Some(parent) = present("parentObservationId").and_then(Value::as_str) { + span["parentSpanId"] = json!(hex_id(parent, 8)); + } + if level == Some("ERROR") { + span["status"] = json!({ "code": 2, "message": status_message.unwrap_or_default() }); + } + Some(span) +} + +/// Convert a locally built Langfuse ingestion batch into Langfuse v4 OTLP +/// spans (see the module docs for the attribute rules). +/// +/// The OTEL trace id is derived from the `trace-create` id, the same way +/// [`super::otlp::otlp_requests`] derives it from a span's trace id, so two +/// exporters that agree on a trace id land in one Langfuse trace. The first +/// observation without a parent is the root and receives the trace's +/// input/output when it has none of its own. +/// +/// # Errors +/// +/// Returns [`TinyAgentsError::Validation`] when the payload has no `batch` +/// array or no `trace-create` event with an id. +pub fn ingestion_batch_to_otlp_spans(payload: &Value) -> Result> { + let events = payload + .get("batch") + .and_then(Value::as_array) + .ok_or_else(|| TinyAgentsError::Validation("ingestion payload has no batch".into()))?; + let trace = events + .iter() + .find(|event| event["type"] == "trace-create") + .and_then(|event| event["body"].as_object()) + .ok_or_else(|| TinyAgentsError::Validation("ingestion batch has no trace-create".into()))?; + let trace_id = trace + .get("id") + .and_then(Value::as_str) + .ok_or_else(|| TinyAgentsError::Validation("trace-create has no id".into()))?; + let trace_hex = hex_id(trace_id, 16); + let trace_attrs = trace_attributes(trace); + let null = Value::Null; + let trace_io = ( + trace.get("input").unwrap_or(&null), + trace.get("output").unwrap_or(&null), + ); + + let mut root_assigned = false; + let mut spans = Vec::with_capacity(events.len()); + for event in events { + let Some(kind) = event["type"].as_str().and_then(observation_type) else { + continue; + }; + let Some(body) = event["body"].as_object() else { + continue; + }; + let is_root = !root_assigned && body.get("parentObservationId").is_none_or(Value::is_null); + let root_io = is_root.then_some((trace_io.0, trace_io.1)); + if let Some(span) = observation_span(&trace_hex, kind, body, &trace_attrs, root_io) { + root_assigned |= is_root; + spans.push(span); + } + } + Ok(spans) +} + +/// [`ingestion_batch_to_otlp_spans`] wrapped into OTLP/HTTP JSON request +/// bodies (resource `service.name = service`, scope `scope`), chunked like +/// [`super::otlp::otlp_requests`]. +/// +/// # Errors +/// +/// As [`ingestion_batch_to_otlp_spans`]. +pub fn ingestion_batch_otlp_requests( + payload: &Value, + service: &str, + scope: &str, +) -> Result> { + Ok(chunk_otlp_spans( + ingestion_batch_to_otlp_spans(payload)?, + service, + scope, + )) +} + +#[cfg(test)] +#[path = "ingestion_otlp_tests.rs"] +mod tests; diff --git a/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp_tests.rs b/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp_tests.rs new file mode 100644 index 000000000..3fba1abc8 --- /dev/null +++ b/crates/tinyagents-harness/src/observability/trace_export/ingestion_otlp_tests.rs @@ -0,0 +1,206 @@ +use serde_json::{Value, json}; + +use super::*; + +fn batch() -> Value { + json!({ "batch": [ + { "id": "e0", "type": "trace-create", "body": { + "id": "flow:demo:run-1", + "name": "flow.run:demo", + "userId": "user-7", + "sessionId": "flow:demo", + "release": "1.2.3", + "environment": "staging", + "tags": ["run:flow", "trigger:manual"], + "input": { "goal": "ship" }, + "output": { "ok": true }, + "metadata": { "flow_id": "demo", "health": { "failed": 0 } }, + } }, + { "id": "e1", "type": "span-create", "body": { + "id": "flow:demo:run-1:run:r1", + "traceId": "flow:demo:run-1", + "name": "demo", + "startTime": "2026-10-15T10:00:00.000Z", + "endTime": "2026-10-15T10:00:02.000Z", + "metadata": { "graph_id": "demo" }, + } }, + { "id": "e2", "type": "span-create", "body": { + "id": "flow:demo:run-1:node:fetch:0", + "traceId": "flow:demo:run-1", + "parentObservationId": "flow:demo:run-1:run:r1", + "name": "fetch", + "startTime": "2026-10-15T10:00:00.500Z", + "endTime": "2026-10-15T10:00:01.000Z", + "level": "ERROR", + "statusMessage": "boom", + "metadata": { "langgraph_node": "fetch", "langgraph_step": 0 }, + } }, + { "id": "e3", "type": "event-create", "body": { + "id": "evt-1", + "traceId": "flow:demo:run-1", + "parentObservationId": "flow:demo:run-1:run:r1", + "name": "route", + "startTime": "2026-10-15T10:00:01.500Z", + } }, + { "id": "e4", "type": "generation-create", "body": { + "id": "gen-1", + "traceId": "flow:demo:run-1", + "parentObservationId": "flow:demo:run-1:node:fetch:0", + "name": "model", + "startTime": "2026-10-15T10:00:00.600Z", + "endTime": "2026-10-15T10:00:00.900Z", + "model": "gpt-x", + "usage": { "input": 10, "output": 5 }, + "costDetails": { "total": 0.01 }, + } }, + { "id": "e5", "type": "score-create", "body": { "id": "s1", "name": "q", "value": 1 } }, + ] }) +} + +fn attr<'a>(span: &'a Value, key: &str) -> Option<&'a Value> { + span["attributes"] + .as_array()? + .iter() + .find(|item| item["key"] == key) + .map(|item| &item["value"]) +} + +fn text<'a>(span: &'a Value, key: &str) -> Option<&'a str> { + attr(span, key)?["stringValue"].as_str() +} + +#[test] +fn every_observation_carries_the_trace_attributes_v4_filters_on() { + let spans = ingestion_batch_to_otlp_spans(&batch()).unwrap(); + assert_eq!( + spans.len(), + 4, + "trace-create and score-create are not spans" + ); + for span in &spans { + assert_eq!(text(span, "langfuse.trace.name"), Some("flow.run:demo")); + assert_eq!(text(span, "langfuse.user.id"), Some("user-7")); + assert_eq!(text(span, "langfuse.session.id"), Some("flow:demo")); + assert_eq!(text(span, "langfuse.release"), Some("1.2.3")); + assert_eq!(text(span, "langfuse.environment"), Some("staging")); + assert_eq!(text(span, "langfuse.trace.metadata.flow_id"), Some("demo")); + assert_eq!( + text(span, "langfuse.trace.metadata.health"), + Some(r#"{"failed":0}"#) + ); + let tags = &attr(span, "langfuse.trace.tags").unwrap()["arrayValue"]["values"]; + assert_eq!(tags[1]["stringValue"], "trigger:manual"); + assert_eq!(span["traceId"], spans[0]["traceId"]); + } +} + +#[test] +fn trace_io_moves_to_the_root_observation_only() { + let spans = ingestion_batch_to_otlp_spans(&batch()).unwrap(); + let root = &spans[0]; + assert!(root.get("parentSpanId").is_none()); + assert_eq!(text(root, "langfuse.observation.type"), Some("span")); + assert_eq!( + text(root, "langfuse.observation.input"), + Some(r#"{"goal":"ship"}"#) + ); + assert_eq!( + text(root, "langfuse.observation.output"), + Some(r#"{"ok":true}"#) + ); + for child in &spans[1..] { + assert!(text(child, "langfuse.observation.input").is_none()); + assert!(attr(child, "langfuse.trace.input").is_none()); + } +} + +#[test] +fn parentage_levels_and_generation_fields_map_to_otel() { + let spans = ingestion_batch_to_otlp_spans(&batch()).unwrap(); + let (root, node, event, generation) = (&spans[0], &spans[1], &spans[2], &spans[3]); + assert_eq!(node["parentSpanId"], root["spanId"]); + assert_eq!(generation["parentSpanId"], node["spanId"]); + assert_eq!(text(node, "langfuse.observation.level"), Some("ERROR")); + assert_eq!( + text(node, "langfuse.observation.status_message"), + Some("boom") + ); + assert_eq!(node["status"]["code"], 2); + assert_eq!( + text(node, "langfuse.observation.metadata.langgraph_node"), + Some("fetch") + ); + assert_eq!( + text(node, "langfuse.observation.metadata.langgraph_step"), + Some("0") + ); + assert_eq!(text(event, "langfuse.observation.type"), Some("event")); + assert_eq!(event["startTimeUnixNano"], event["endTimeUnixNano"]); + assert_eq!( + text(generation, "langfuse.observation.type"), + Some("generation") + ); + assert_eq!( + text(generation, "langfuse.observation.model.name"), + Some("gpt-x") + ); + assert_eq!( + text(generation, "langfuse.observation.usage_details"), + Some(r#"{"input":10,"output":5}"#) + ); + assert_eq!( + text(generation, "langfuse.observation.cost_details"), + Some(r#"{"total":0.01}"#) + ); + assert_eq!(root["startTimeUnixNano"], "1792058400000000000"); + assert_eq!(root["endTimeUnixNano"], "1792058402000000000"); +} + +#[test] +fn span_ids_are_deterministic_across_exports() { + let first = ingestion_batch_to_otlp_spans(&batch()).unwrap(); + let second = ingestion_batch_to_otlp_spans(&batch()).unwrap(); + assert_eq!(first, second); + assert_eq!(first[0]["spanId"].as_str().unwrap().len(), 16); + assert_eq!(first[0]["traceId"].as_str().unwrap().len(), 32); +} + +#[test] +fn an_open_span_is_closed_at_its_start_and_flagged() { + let mut payload = batch(); + payload["batch"][2]["body"] + .as_object_mut() + .unwrap() + .remove("endTime"); + let spans = ingestion_batch_to_otlp_spans(&payload).unwrap(); + assert_eq!(spans[1]["startTimeUnixNano"], spans[1]["endTimeUnixNano"]); + assert_eq!( + text(&spans[1], "langfuse.observation.metadata.unfinished"), + Some("true") + ); +} + +#[test] +fn requests_wrap_spans_under_one_resource() { + let requests = ingestion_batch_otlp_requests(&batch(), "openhuman", "openhuman.flows").unwrap(); + assert_eq!(requests.len(), 1); + let resource = &requests[0]["resourceSpans"][0]; + assert_eq!( + resource["resource"]["attributes"][0]["value"]["stringValue"], + "openhuman" + ); + assert_eq!( + resource["scopeSpans"][0]["scope"]["name"], + "openhuman.flows" + ); + assert_eq!( + resource["scopeSpans"][0]["spans"].as_array().unwrap().len(), + 4 + ); +} + +#[test] +fn a_batch_without_a_trace_is_refused() { + assert!(ingestion_batch_to_otlp_spans(&json!({ "batch": [] })).is_err()); + assert!(ingestion_batch_to_otlp_spans(&json!({})).is_err()); +} diff --git a/crates/tinyagents-harness/src/observability/trace_export/mod.rs b/crates/tinyagents-harness/src/observability/trace_export/mod.rs index 7e516a4d8..7796c277d 100644 --- a/crates/tinyagents-harness/src/observability/trace_export/mod.rs +++ b/crates/tinyagents-harness/src/observability/trace_export/mod.rs @@ -7,6 +7,7 @@ //! [`ExportBrand`] so product identity strings stay out of this crate. pub mod ingestion_batch; +pub mod ingestion_otlp; #[cfg(feature = "langfuse")] pub mod journal_export; pub mod otlp; diff --git a/crates/tinyagents-harness/src/observability/trace_export/otlp.rs b/crates/tinyagents-harness/src/observability/trace_export/otlp.rs index 85178c1fe..24bec4a6b 100644 --- a/crates/tinyagents-harness/src/observability/trace_export/otlp.rs +++ b/crates/tinyagents-harness/src/observability/trace_export/otlp.rs @@ -31,7 +31,7 @@ fn span_level(span: &TraceSpan) -> Option<&'static str> { const MAX_SPANS_PER_REQUEST: usize = 20; const MAX_REQUEST_BYTES: usize = 8_000_000; -fn hex_id(source: &str, bytes: usize) -> String { +pub(super) fn hex_id(source: &str, bytes: usize) -> String { let digest = Sha256::digest(source.as_bytes()); digest[..bytes] .iter() @@ -39,11 +39,11 @@ fn hex_id(source: &str, bytes: usize) -> String { .collect() } -fn attribute(key: &str, value: impl Into) -> Value { +pub(super) fn attribute(key: &str, value: impl Into) -> Value { json!({ "key": key, "value": { "stringValue": value.into() } }) } -fn json_attribute(key: &str, value: &Value) -> Value { +pub(super) fn json_attribute(key: &str, value: &Value) -> Value { attribute(key, value.to_string()) } @@ -457,11 +457,27 @@ pub fn otlp_requests( // only once so Langfuse usage metrics do not count it twice. let visible = without_child_run_details(spans); let compacted = compact_internal_searches(&visible); + let converted = compacted + .iter() + .map(|span| span_to_otlp(span, root, environment, brand)) + .collect(); + chunk_otlp_spans( + converted, + brand.product, + &format!("{}.agent", brand.product), + ) +} + +/// Wrap already-converted OTLP spans into OTLP/HTTP JSON request bodies under +/// one resource (`service.name = service`) and instrumentation scope, chunked +/// so no request exceeds the transport caps (20 spans, ~8 MB). Span order is +/// preserved; an empty input yields no requests. +pub fn chunk_otlp_spans(spans: Vec, service: &str, scope: &str) -> Vec { let make_request = |spans: Vec| { json!({ "resourceSpans": [{ - "resource": { "attributes": [attribute("service.name", brand.product)] }, + "resource": { "attributes": [attribute("service.name", service)] }, "scopeSpans": [{ - "scope": { "name": format!("{}.agent", brand.product) }, + "scope": { "name": scope }, "spans": spans, }], }] }) @@ -469,8 +485,7 @@ pub fn otlp_requests( let mut requests = Vec::new(); let mut batch = Vec::new(); let mut bytes = 256; - for span in &compacted { - let converted = span_to_otlp(span, root, environment, brand); + for converted in spans { let span_bytes = converted.to_string().len(); if !batch.is_empty() && (batch.len() >= MAX_SPANS_PER_REQUEST || bytes + span_bytes > MAX_REQUEST_BYTES)