From 7967b4f7f85e64c7ab7a382182788029b15df89f Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Tue, 7 Jul 2026 21:00:56 +0530 Subject: [PATCH 1/2] harness(langfuse): stamp model on the ModelCompleted generation The per-call generation-create observation was projected solely from AgentEvent::ModelCompleted, which carries no model name (only ModelStarted does), so the exporter never set body["model"]. Langfuse could not map its pricing table and every agent-run generation recorded cost $0. Pre-scan the batch to correlate each call's model by call_id from ModelStarted, and stamp body["model"] on the matching ModelCompleted generation. clean_nulls drops the key when a model can't be correlated, leaving the payload unchanged for that case. --- src/harness/observability/langfuse/mod.rs | 38 +++++++++++++++++- src/harness/observability/langfuse/test.rs | 46 ++++++++++++++++++++++ 2 files changed, 82 insertions(+), 2 deletions(-) diff --git a/src/harness/observability/langfuse/mod.rs b/src/harness/observability/langfuse/mod.rs index 7a8a32d79..22ea18052 100644 --- a/src/harness/observability/langfuse/mod.rs +++ b/src/harness/observability/langfuse/mod.rs @@ -4,6 +4,8 @@ //! 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; + use serde_json::{Value, json}; use crate::error::{Result, TinyAgentsError}; @@ -109,6 +111,13 @@ impl LangfuseClient { .unwrap_or_else(|| iso_ms(now_ms())); let metadata = trace_metadata(&trace, observations); + // The per-call generation is projected from `ModelCompleted`, which does + // not carry the model name — only `ModelStarted` does. Pre-scan the batch + // to correlate each call's model by `call_id` so the generation can stamp + // `body["model"]`; without it Langfuse can't map pricing and every + // generation's cost is $0. + let call_models = collect_call_models(observations); + let mut batch = Vec::with_capacity(observations.len() + 1); batch.push(json!({ "id": format!("{}:trace", trace_id), @@ -129,7 +138,7 @@ impl LangfuseClient { })); for obs in observations { - batch.push(observation_event(&trace_id, obs)); + batch.push(observation_event(&trace_id, obs, &call_models)); } Ok(json!({ "batch": batch })) @@ -269,7 +278,27 @@ fn trace_metadata(trace: &LangfuseTraceConfig, observations: &[AgentObservation] } } -fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { +/// Collect the model each `ModelStarted` announced, keyed by its `call_id`. +/// +/// The per-call `generation-create` is built from `ModelCompleted`, which does +/// not carry the model name; this correlation lets the generation stamp +/// `body["model"]` so Langfuse can price the call. Borrows from `observations`, +/// which outlive the batch build. +fn collect_call_models(observations: &[AgentObservation]) -> BTreeMap<&str, &str> { + let mut models = BTreeMap::new(); + for obs in observations { + if let AgentEvent::ModelStarted { call_id, model } = &obs.event { + models.insert(call_id.as_str(), model.as_str()); + } + } + models +} + +fn observation_event( + trace_id: &str, + obs: &AgentObservation, + call_models: &BTreeMap<&str, &str>, +) -> Value { let timestamp = iso_ms(obs.ts_ms); // Attach only run lineage, offset, and the event *kind* — not the full event // payload. The event's meaningful fields (input/output/usage/name) are @@ -312,6 +341,11 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value { "id": scoped_observation_id(trace_id, call_id.as_str()), "traceId": trace_id, "name": "model", + // The model the matching `ModelStarted` announced for this + // call. Langfuse maps its pricing table off this field, so + // without it every generation's cost is $0. `clean_nulls` + // drops the key when the model can't be correlated. + "model": call_models.get(call_id.as_str()).copied(), // Use the loop-captured start time so the generation has a // real duration; fall back to the completion timestamp (a // zero-width point) for events journaled before the field diff --git a/src/harness/observability/langfuse/test.rs b/src/harness/observability/langfuse/test.rs index 801cca8d0..db8049bb9 100644 --- a/src/harness/observability/langfuse/test.rs +++ b/src/harness/observability/langfuse/test.rs @@ -95,6 +95,52 @@ fn builds_trace_and_generation_batch() { assert_eq!(events[2]["body"]["startTime"], events[2]["body"]["endTime"]); } +#[test] +fn generation_carries_model_from_model_started() { + let client = + LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t").unwrap(); + let batch = client + .build_ingestion_batch( + LangfuseTraceConfig::default(), + &[ + obs( + 0, + AgentEvent::ModelStarted { + call_id: CallId::new("model-call"), + model: "managed.chat-v1".to_string(), + }, + ), + obs( + 1, + AgentEvent::ModelCompleted { + call_id: CallId::new("model-call"), + started_at_ms: None, + usage: Some(Usage { + input_tokens: 3, + output_tokens: 4, + total_tokens: 7, + ..Default::default() + }), + input: None, + output: None, + }, + ), + ], + ) + .unwrap(); + + let events = batch["batch"].as_array().unwrap(); + let generation = events + .iter() + .find(|e| e["type"] == "generation-create") + .expect("a generation-create observation"); + // The generation is projected from `ModelCompleted` (which carries no model); + // it is correlated by `call_id` to the model that `ModelStarted` announced, + // so Langfuse can map pricing instead of recording cost $0. + assert_eq!(generation["body"]["model"], "managed.chat-v1"); + assert_eq!(generation["body"]["metadata"]["call_id"], "model-call"); +} + #[test] fn populates_generation_and_tool_io_when_captured() { let client = From 3f2c64a7a9ca6a9197576da041386575c2fb036f Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Tue, 7 Jul 2026 19:38:09 +0530 Subject: [PATCH 2/2] harness: fix clippy lints surfaced by stable 1.96 collapsible_if in StreamAccumulator::push_tool_chunk and a duplicated #[cfg(test)] on the stream test module both fail CI's cargo clippy --all-targets -- -D warnings under rust 1.96. Behavior-neutral. --- src/harness/model/mod.rs | 8 ++++---- src/harness/stream/mod.rs | 1 - 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/src/harness/model/mod.rs b/src/harness/model/mod.rs index aa24cb174..ea3e6b8ce 100644 --- a/src/harness/model/mod.rs +++ b/src/harness/model/mod.rs @@ -753,10 +753,10 @@ impl StreamAccumulator { fn push_tool_chunk(&mut self, call_id: &str, content: &str, tool_name: Option<&str>) { if let Some(entry) = self.tool_chunks.iter_mut().find(|(id, ..)| id == call_id) { entry.1.push_str(content); - if entry.2.is_none() { - if let Some(name) = tool_name.filter(|n| !n.is_empty()) { - entry.2 = Some(name.to_string()); - } + if entry.2.is_none() + && let Some(name) = tool_name.filter(|n| !n.is_empty()) + { + entry.2 = Some(name.to_string()); } } else { self.tool_chunks.push(( diff --git a/src/harness/stream/mod.rs b/src/harness/stream/mod.rs index 14411f0b3..17121426b 100644 --- a/src/harness/stream/mod.rs +++ b/src/harness/stream/mod.rs @@ -143,6 +143,5 @@ pub fn stream(chunks: &[StreamChunk], modes: &[StreamMode]) -> Vec .collect() } -#[cfg(test)] #[cfg(test)] mod test;