fix(observability): chunk Langfuse ingestion batches to <=500 events#4676
Conversation
Large agent turns — especially ones that spawn sub-agents — can produce far more than 500 trace events. push_observations POSTed them as a single Langfuse ingestion batch, which Langfuse rejects with 400 "batch cannot exceed 500 events", dropping the ENTIRE trace. Split the batch into <=500-event chunks and send each; Langfuse dedupes by event id and resolves observations to their trace by traceId, so a run's events may span several requests.
📝 WalkthroughWalkthroughAdds batching support to Langfuse ingestion to avoid 400 errors when payloads exceed the 500-event limit. Introduces a ChangesLangfuse Ingestion Batch Chunking
Estimated code review effort: 2 (Simple) | ~10 minutes Sequence Diagram(s)sequenceDiagram
participant PushObservations as push_observations
participant Splitter as split_ingestion_batch
participant Client as LangfuseClient
PushObservations->>Splitter: split_ingestion_batch(payload, 500)
Splitter-->>PushObservations: Vec of chunked payloads
loop each chunk
PushObservations->>Client: send_batch(chunk)
Client-->>PushObservations: result / timeout error
end
Possibly related PRs
Suggested labels: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
coderabbitai
Bot
added
agent
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/openhuman/agent/progress_tracing/langfuse.rs (1)
480-493: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueMinor: redundant cloning for large batches.
For a batch that needs chunking, this clones the events vec once (line 482), then clones the full
payloadonce per chunk (line 488) before overwritingbatch, pluschunk.to_vec()copies each chunk again. For a ~1200-event payload this means the roughly-full JSON structure gets copied multiple times. Given this only triggers on rare oversized turns, the cost is likely negligible, but if oversized turns become common (e.g. many sub-agents), building each chunk from a sharedMap(cloning only non-batchkeys once) would avoid the repeated full-payload clones.♻️ Possible optimization
fn split_ingestion_batch(payload: Value, max: usize) -> Vec<Value> { 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() + let mut base = payload; + if let Some(obj) = base.as_object_mut() { + obj.remove("batch"); + } + events + .chunks(max) + .map(|chunk| { + let mut part = base.clone(); + part["batch"] = Value::Array(chunk.to_vec()); + part + }) + .collect() }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/agent/progress_tracing/langfuse.rs` around lines 480 - 493, The chunking path in split_ingestion_batch redundantly clones the full payload for every chunk and copies each chunk again, which is wasteful for large oversized batches. Refactor split_ingestion_batch to build each part from shared non-batch data instead of cloning the entire Value repeatedly, and only materialize the per-chunk batch contents once; keep the same behavior for max <= 0 or non-array batch inputs.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/openhuman/agent/progress_tracing/langfuse.rs`:
- Around line 480-493: The chunking path in split_ingestion_batch redundantly
clones the full payload for every chunk and copies each chunk again, which is
wasteful for large oversized batches. Refactor split_ingestion_batch to build
each part from shared non-batch data instead of cloning the entire Value
repeatedly, and only materialize the per-chunk batch contents once; keep the
same behavior for max <= 0 or non-array batch inputs.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 297418d5-4b66-4e40-8e78-2258144265e9
📒 Files selected for processing (1)
src/openhuman/agent/progress_tracing/langfuse.rs
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 164e92848a
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| for chunk in split_ingestion_batch(payload, LANGFUSE_MAX_BATCH_EVENTS) { | ||
| tokio::time::timeout(PUSH_TIMEOUT, client.send_batch(chunk)) |
There was a problem hiding this comment.
When a large trace is split into many chunks and the Langfuse proxy is slow or hung, this applies the 10s PUSH_TIMEOUT to each chunk rather than to the whole best-effort push. The previous single POST could delay teardown by at most one timeout, but a 20-chunk trace can now block this export path for roughly 200s before returning. Wrap the whole chunk loop in a single timeout/deadline, or track the remaining budget per chunk.
Useful? React with 👍 / 👎.
2 participants
Large agent turns (especially ones that spawn sub-agents) produce more than Langfuse's 500-event-per-batch limit.
push_observationssent them as one batch, so Langfuse rejected the entire trace with400 "batch cannot exceed 500 events"and the turn never appeared in the dashboard — even though the turn itself succeeded.Fix: split the ingestion batch into ≤500-event chunks and send each sequentially. Langfuse dedupes by event id and resolves observations to their trace by
traceId, so a run's events can be delivered across several requests (thetrace-createstays in chunk 1). Adds two unit tests.Note: the sibling legacy
push_spanspath has the same one-shot POST and latent limit — left as a follow-up since the live journal path (push_observations) is the one producing these traces.Verified:
GGML_NATIVE=OFF cargo checkpasses.Closes #4675
Summary by CodeRabbit