diff --git a/src/openhuman/agent/progress_tracing/langfuse.rs b/src/openhuman/agent/progress_tracing/langfuse.rs index 674e70b332..447b587907 100644 --- a/src/openhuman/agent/progress_tracing/langfuse.rs +++ b/src/openhuman/agent/progress_tracing/langfuse.rs @@ -464,6 +464,34 @@ fn insert_run_telemetry_generation(payload: &mut Value, telemetry: Option<&RunTe /// exporter. The journal is already redacted before persistence, and this /// exporter additionally strips model/tool payloads unless `capture_content` /// is explicitly enabled. +/// Langfuse rejects an ingestion request whose `batch` holds more than 500 +/// events (`400 "Langfuse ingestion batch cannot exceed 500 events"`). Large +/// turns — especially ones that spawn sub-agents — routinely exceed this. +const LANGFUSE_MAX_BATCH_EVENTS: usize = 500; + +/// Split a `{"batch": [...]}` ingestion payload into multiple payloads, each +/// carrying at most `max` events and preserving any other top-level keys. +/// +/// Langfuse dedupes ingestion events by id and resolves each observation to its +/// trace by `traceId`, so delivering one run's events across several requests is +/// safe (the `trace-create` event stays in the first chunk). A payload at or +/// under the limit — or without a `batch` array — passes through unchanged as a +/// single element. +fn split_ingestion_batch(payload: Value, max: usize) -> Vec { + let events = match payload.get("batch").and_then(Value::as_array) { + Some(events) if max > 0 && events.len() > max => events.clone(), + _ => return vec![payload], + }; + events + .chunks(max) + .map(|chunk| { + let mut part = payload.clone(); + part["batch"] = Value::Array(chunk.to_vec()); + part + }) + .collect() +} + pub(crate) async fn push_observations( config: &Config, trace_ctx: &TraceContext, @@ -506,10 +534,15 @@ pub(crate) async fn push_observations( "[agent-tracing] no run telemetry aggregate added to Langfuse journal batch" ); } - tokio::time::timeout(PUSH_TIMEOUT, client.send_batch(payload)) - .await - .map_err(|_| format!("Langfuse journal push timed out after {PUSH_TIMEOUT:?}"))? - .map_err(|err| format!("Langfuse journal ingestion failed: {err}"))?; + // Langfuse caps a single ingestion request at 500 events; a large run (e.g. + // one that spawns sub-agents) can far exceed that and previously had its + // ENTIRE trace rejected with a 400. Send in <=500-event chunks instead. + for chunk in split_ingestion_batch(payload, LANGFUSE_MAX_BATCH_EVENTS) { + tokio::time::timeout(PUSH_TIMEOUT, client.send_batch(chunk)) + .await + .map_err(|_| format!("Langfuse journal push timed out after {PUSH_TIMEOUT:?}"))? + .map_err(|err| format!("Langfuse journal ingestion failed: {err}"))?; + } tracing::debug!( target: LOG_TARGET, @@ -593,6 +626,41 @@ mod tests { use super::*; use std::collections::BTreeMap; + #[test] + fn split_ingestion_batch_chunks_over_the_limit() { + // 1201 events with max 500 -> chunks of 500, 500, 201. + let events: Vec = (0..1201).map(|i| json!({ "id": i })).collect(); + let payload = json!({ "batch": events, "metadata": { "k": "v" } }); + + let parts = split_ingestion_batch(payload, 500); + assert_eq!(parts.len(), 3); + let sizes: Vec = parts + .iter() + .map(|p| p["batch"].as_array().unwrap().len()) + .collect(); + assert_eq!(sizes, vec![500, 500, 201]); + // Every chunk stays within the limit and preserves other top-level keys. + for p in &parts { + assert!(p["batch"].as_array().unwrap().len() <= 500); + assert_eq!(p["metadata"]["k"], json!("v")); + } + // The first event of the run (e.g. the trace-create) lands in chunk 0. + assert_eq!(parts[0]["batch"][0]["id"], json!(0)); + // Order is preserved across the split. + assert_eq!(parts[2]["batch"][0]["id"], json!(1000)); + } + + #[test] + fn split_ingestion_batch_passes_small_payloads_through() { + let payload = json!({ "batch": [json!({ "id": 1 }), json!({ "id": 2 })] }); + let parts = split_ingestion_batch(payload.clone(), 500); + assert_eq!(parts.len(), 1); + assert_eq!(parts[0], payload); + // A payload without a `batch` array is returned untouched. + let no_batch = json!({ "hello": "world" }); + assert_eq!(split_ingestion_batch(no_batch.clone(), 500), vec![no_batch]); + } + use crate::openhuman::agent::progress_tracing::SpanKind; use tinyagents::harness::ids::{CallId, EventId, RunId}; use tinyagents::harness::usage::Usage;