diff --git a/src/harness/observability/langfuse/mod.rs b/src/harness/observability/langfuse/mod.rs index 527bcc0..7a8a32d 100644 --- a/src/harness/observability/langfuse/mod.rs +++ b/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, diff --git a/src/harness/observability/langfuse/test.rs b/src/harness/observability/langfuse/test.rs index c062215..801cca8 100644 --- a/src/harness/observability/langfuse/test.rs +++ b/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 =