harness(langfuse): stamp model on the ModelCompleted generation

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((

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

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 =

src/harness/stream/mod.rs

@@ -143,6 +143,5 @@ pub fn stream(chunks: &[StreamChunk], modes: &[StreamMode]) -> Vec<StreamChunk>
         .collect()
 }
 
-#[cfg(test)]
 #[cfg(test)]
 mod test;