fix(observability): namespace Langfuse observation ids by trace to stop cross-trace overwrites

src/harness/observability/langfuse/mod.rs

@@ -290,26 +290,41 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
             usage,
             input,
             output,
-        } => json!({
-            "id": obs.event_id.as_str(),
-            "timestamp": timestamp,
-            "type": "generation-create",
-            "body": clean_nulls(json!({
-                "id": call_id.as_str(),
-                "traceId": trace_id,
-                "name": "model",
-                // 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
-                // existed.
-                "startTime": started_at_ms.map(iso_ms).unwrap_or_else(|| timestamp.clone()),
-                "endTime": timestamp,
-                "usage": usage.map(langfuse_usage),
-                "input": input,
-                "output": output,
-                "metadata": metadata,
-            })),
-        }),
+        } => {
+            // Langfuse upserts observations by `id` across the ENTIRE project, so
+            // the observation id must be globally unique. `call_id` is only
+            // unique within a single run (e.g. `agent_turn-model-1`), and the
+            // logical run id is reused by every interactive turn — so a bare
+            // call_id collides across turns and threads, and each new turn's
+            // generation silently overwrites the previous one, stealing it onto
+            // the newest trace (leaving every earlier trace with no model usage/
+            // cost/content). Namespace the id with the globally-unique, per-trace
+            // `trace_id`: it never collides across turns yet stays stable for
+            // idempotent re-ingestion of the same run. The raw `call_id` rides in
+            // metadata so in-run correlation (model.started ↔ this generation) is
+            // preserved.
+            let metadata = with_call_id(metadata, call_id.as_str());
+            json!({
+                "id": obs.event_id.as_str(),
+                "timestamp": timestamp,
+                "type": "generation-create",
+                "body": clean_nulls(json!({
+                    "id": scoped_observation_id(trace_id, call_id.as_str()),
+                    "traceId": trace_id,
+                    "name": "model",
+                    // 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
+                    // existed.
+                    "startTime": started_at_ms.map(iso_ms).unwrap_or_else(|| timestamp.clone()),
+                    "endTime": timestamp,
+                    "usage": usage.map(langfuse_usage),
+                    "input": input,
+                    "output": output,
+                    "metadata": metadata,
+                })),
+            })
+        }
         AgentEvent::ToolCompleted {
             call_id,
             tool_name,
@@ -328,7 +343,12 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
                 (Some(start), Some(dur)) => iso_ms(start.saturating_add(*dur)),
                 _ => timestamp.clone(),
             };
-            let mut tool_metadata = metadata.clone();
+            // Same project-wide id-collision hazard as ModelCompleted above:
+            // builtin tools take deterministic per-run call_ids (e.g.
+            // `agent_turn-tool-1`) that repeat every turn, so a bare call_id
+            // overwrites earlier tool spans onto the newest trace. Namespace with
+            // the per-trace id; keep the raw call_id in metadata for correlation.
+            let mut tool_metadata = with_call_id(metadata.clone(), call_id.as_str());
             if let (Some(map), Some(bytes)) = (tool_metadata.as_object_mut(), output_bytes) {
                 map.insert("output_bytes".into(), json!(bytes));
             }
@@ -340,7 +360,7 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
                 // Langfuse rejects it, silently dropping every tool observation.
                 "type": "span-create",
                 "body": clean_nulls(json!({
-                    "id": call_id.as_str(),
+                    "id": scoped_observation_id(trace_id, call_id.as_str()),
                     "traceId": trace_id,
                     "name": tool_name,
                     "startTime": started_at_ms.map(iso_ms).unwrap_or_else(|| timestamp.clone()),
@@ -382,6 +402,27 @@ fn observation_event(trace_id: &str, obs: &AgentObservation) -> Value {
     }
 }
 
+/// Build a globally-unique-yet-stable Langfuse observation id for a call-scoped
+/// observation (model generation / tool span). Langfuse keys observations by
+/// `id` across the whole project and upserts on collision, so we prefix the
+/// (only run-unique) `call_id` with the globally-unique `trace_id`. The result
+/// is deterministic for a given (trace, call), keeping re-ingestion idempotent
+/// while never colliding across turns or threads.
+fn scoped_observation_id(trace_id: &str, call_id: &str) -> String {
+    format!("{trace_id}:{call_id}")
+}
+
+/// Fold the raw `call_id` into an observation's metadata so in-run correlation
+/// (e.g. `model.started` ↔ its `model` generation) survives even though the
+/// public observation id is now trace-namespaced. Returns the metadata
+/// unchanged when it is not a JSON object.
+fn with_call_id(mut metadata: Value, call_id: &str) -> Value {
+    if let Some(map) = metadata.as_object_mut() {
+        map.insert("call_id".into(), json!(call_id));
+    }
+    metadata
+}
+
 fn langfuse_usage(usage: Usage) -> Value {
     json!({
         "input": usage.input_tokens,

src/harness/observability/langfuse/test.rs

@@ -80,7 +80,12 @@ fn builds_trace_and_generation_batch() {
     assert_eq!(events[0]["body"]["metadata"]["root_run_id"], "root-1");
     assert_eq!(events[0]["body"]["metadata"]["run_id"], "run-1");
     assert_eq!(events[2]["type"], "generation-create");
-    assert_eq!(events[2]["body"]["id"], "model-call");
+    // The observation id is namespaced by the (globally-unique) trace id so it
+    // cannot collide across turns/threads that reuse the same run-scoped call_id.
+    // Here the trace id defaults to the root run id ("root-1").
+    assert_eq!(events[2]["body"]["id"], "root-1:model-call");
+    // The raw call_id survives in metadata for in-run correlation.
+    assert_eq!(events[2]["body"]["metadata"]["call_id"], "model-call");
     assert_eq!(events[2]["body"]["usage"]["input"], 3);
     // Payload-free generation: no input/output body fields.
     assert!(events[2]["body"].get("input").is_none());
@@ -156,6 +161,91 @@ fn populates_generation_and_tool_io_when_captured() {
     assert_eq!(gen_meta["run_id"], "run-1");
 }
 
+#[test]
+fn call_scoped_observation_ids_are_unique_per_trace() {
+    // Regression for the Langfuse id-collision bug: two turns on two different
+    // threads reuse the SAME run-scoped call_id (`agent_turn-model-1`, the
+    // deterministic id every interactive turn gets). Because Langfuse upserts
+    // observations by `id` project-wide, a bare call_id made each new turn's
+    // model generation overwrite the previous one onto the newest trace — so
+    // every earlier trace lost its model usage/cost/content. The id must now be
+    // distinct per trace.
+    let client =
+        LangfuseClient::proxy("https://backend.test/telemetry/langfuse/ingestion", "t").unwrap();
+
+    let build = |trace_id: &str| {
+        client
+            .build_ingestion_batch(
+                LangfuseTraceConfig {
+                    trace_id: Some(trace_id.to_string()),
+                    ..Default::default()
+                },
+                &[
+                    obs(
+                        1,
+                        AgentEvent::ModelCompleted {
+                            call_id: CallId::new("agent_turn-model-1"),
+                            started_at_ms: None,
+                            usage: Some(Usage {
+                                input_tokens: 1,
+                                output_tokens: 1,
+                                total_tokens: 2,
+                                ..Default::default()
+                            }),
+                            input: None,
+                            output: None,
+                        },
+                    ),
+                    obs(
+                        2,
+                        AgentEvent::ToolCompleted {
+                            call_id: CallId::new("agent_turn-tool-1"),
+                            tool_name: "lookup".to_string(),
+                            started_at_ms: None,
+                            input: None,
+                            output: None,
+                            duration_ms: None,
+                            output_bytes: None,
+                            error: None,
+                        },
+                    ),
+                ],
+            )
+            .unwrap()
+    };
+
+    let extract_ids = |batch: &Value| -> (String, String) {
+        let events = batch["batch"].as_array().unwrap();
+        let gen_id = events
+            .iter()
+            .find(|e| e["type"] == "generation-create")
+            .unwrap()["body"]["id"]
+            .as_str()
+            .unwrap()
+            .to_string();
+        let span_id = events.iter().find(|e| e["type"] == "span-create").unwrap()["body"]["id"]
+            .as_str()
+            .unwrap()
+            .to_string();
+        (gen_id, span_id)
+    };
+
+    let (gen_a, span_a) = extract_ids(&build("thread-A:turn-1"));
+    let (gen_b, span_b) = extract_ids(&build("thread-B:turn-2"));
+
+    // Same call_id, different traces → different observation ids (no overwrite).
+    assert_ne!(
+        gen_a, gen_b,
+        "model generation ids must not collide across traces"
+    );
+    assert_ne!(
+        span_a, span_b,
+        "tool span ids must not collide across traces"
+    );
+    assert_eq!(gen_a, "thread-A:turn-1:agent_turn-model-1");
+    assert_eq!(span_b, "thread-B:turn-2:agent_turn-tool-1");
+}
+
 #[test]
 fn merges_caller_trace_metadata_over_defaults() {
     let client =