From f8243e31a9d3b286aefdc066209860609fabcafb Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 3 Jul 2026 18:34:02 +0530 Subject: [PATCH 1/3] feat(tracing): export prompt/reply + native token/cost to Langfuse Adds opt-in content capture and promotes token/cost onto agent trace spans, plus two correctness fixes to the Langfuse exporter path. - Content (opt-in): the turn's prompt and the model's reply ride a new AgentProgress::TurnContent event onto the turn span; the exporter emits them as Langfuse `input`/`output` only when observability.agent_tracing.capture_content is set (default off; env toggle OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT). Preserves the metadata-only, PII-free default. - Native usage/cost: the turn's gen_ai.usage.* attributes are promoted into Langfuse's native generation / usageDetails / costDetails (they were buried in metadata, so cost always rendered 0). - Fix (span-id collision): span ids were a per-turn sequence (0000...0001) reset each turn; Langfuse dedupes observations by id globally, so later turns' observations silently bound to whichever trace first claimed each id (turns appeared empty). Ids now carry a per-turn nonce prefix. - One trace per turn: the trace id is unique per request; the thread id rides along as Langfuse `sessionId` (and client as `userId`) so a conversation's per-turn traces still group under one session. - Robustness: push_spans now inspects the 207 response body and warns on per-event rejections instead of swallowing them. Tests: 29 progress_tracing tests pass, including a new exporter test asserting usage->generation promotion and the content gate (on/off). --- .../agent/harness/session/turn/core.rs | 10 + src/openhuman/agent/progress.rs | 12 ++ src/openhuman/agent/progress_tracing.rs | 63 +++++- .../agent/progress_tracing/langfuse.rs | 183 ++++++++++++++++-- src/openhuman/agent/progress_tracing/tests.rs | 5 + .../channels/providers/web/progress_bridge.rs | 18 +- .../config/schema/load/env_overlay.rs | 15 ++ src/openhuman/config/schema/observability.rs | 8 + src/openhuman/threads/turn_state/mirror.rs | 5 + src/openhuman/workflows/run_log.rs | 3 +- 10 files changed, 294 insertions(+), 28 deletions(-) diff --git a/src/openhuman/agent/harness/session/turn/core.rs b/src/openhuman/agent/harness/session/turn/core.rs index 261f54391a..083dbb91cc 100644 --- a/src/openhuman/agent/harness/session/turn/core.rs +++ b/src/openhuman/agent/harness/session/turn/core.rs @@ -1063,6 +1063,16 @@ impl Agent { ) .await; + // Content (prompt + reply) rides its own event so a tracing consumer can + // attach it to the turn span. Transmission off-device stays gated by the + // exporter's opt-in `capture_content` flag; here it is only surfaced onto + // the in-memory span. + self.emit_progress(AgentProgress::TurnContent { + input: Some(user_message.to_string()), + output: Some(reply.clone()), + }) + .await; + self.emit_progress(AgentProgress::TurnCompleted { iterations: outcome.model_calls as u32, }) diff --git a/src/openhuman/agent/progress.rs b/src/openhuman/agent/progress.rs index 99e47aceb8..6d7dec0890 100644 --- a/src/openhuman/agent/progress.rs +++ b/src/openhuman/agent/progress.rs @@ -283,4 +283,16 @@ pub enum AgentProgress { /// Total iterations used. iterations: u32, }, + + /// The turn's content: the user's prompt and the model's final reply. + /// Emitted just before [`Self::TurnCompleted`] so a tracing consumer can + /// attach `input`/`output` to the turn span. Carries prompt/reply text, so + /// exporters must honor the opt-in `observability.agent_tracing.capture_content` + /// gate before transmitting it off-device. + TurnContent { + /// The user's prompt for this turn. + input: Option, + /// The model's final reply for this turn. + output: Option, + }, } diff --git a/src/openhuman/agent/progress_tracing.rs b/src/openhuman/agent/progress_tracing.rs index e599665664..1a0137d98b 100644 --- a/src/openhuman/agent/progress_tracing.rs +++ b/src/openhuman/agent/progress_tracing.rs @@ -52,11 +52,16 @@ pub(crate) mod langfuse; /// Trace-level correlation context, stamped onto the root span. #[derive(Debug, Clone)] pub struct TraceContext { - /// Trace id — the agent session id. All spans of a run share it. + /// Trace id — unique per turn. Every span of a single turn shares it, so + /// each turn becomes its own Langfuse trace. pub session_id: String, /// User attribution (e.g. the broadcast client id / "system" for /// autonomous runs). `None` when the caller is anonymous. pub user_id: Option, + /// Grouping key (the thread/conversation id) exported as the Langfuse + /// `sessionId` so per-turn traces still group under one session. `None` + /// leaves the trace ungrouped. + pub session_group: Option, } impl TraceContext { @@ -64,8 +69,16 @@ impl TraceContext { Self { session_id: session_id.into(), user_id, + session_group: None, } } + + /// Set the grouping key (thread/conversation id) for the Langfuse + /// `sessionId`, so a conversation's per-turn traces group together. + pub fn with_session_group(mut self, group: impl Into) -> Self { + self.session_group = Some(group.into()); + self + } } /// Derive the trace id (session id) for a run: prefer the UI session id when @@ -129,6 +142,15 @@ pub struct TraceSpan { pub status: SpanStatus, /// Metadata-only attributes (no secrets/PII). pub attributes: BTreeMap, + /// Optional prompt/input content. Populated only when content capture is on + /// (via `AgentProgress::TurnContent`); the exporter still gates transmission + /// behind `observability.agent_tracing.capture_content`. + #[serde(skip_serializing_if = "Option::is_none")] + pub input: Option, + /// Optional model-reply/output content. Same capture/gating rules as + /// [`Self::input`]. + #[serde(skip_serializing_if = "Option::is_none")] + pub output: Option, } impl TraceSpan { @@ -161,6 +183,12 @@ pub struct SpanCollector { ctx: TraceContext, spans: Vec, next_span_seq: u64, + /// Per-collector (per-turn) random prefix for minted span ids. Langfuse + /// dedupes observations by id **globally**, so a bare per-turn sequence + /// (`0000…0001`) collides across turns and silently binds later turns' + /// observations to whichever trace first claimed the id. Prefixing with a + /// fresh nonce makes every span id globally unique. + id_prefix: String, turn_span_id: Option, turn_span_index: Option, @@ -179,6 +207,7 @@ impl SpanCollector { ctx, spans: Vec::new(), next_span_seq: 0, + id_prefix: uuid::Uuid::new_v4().simple().to_string(), turn_span_id: None, turn_span_index: None, current_iteration_span_id: None, @@ -208,7 +237,9 @@ impl SpanCollector { /// and deterministic within a run, which keeps the tests reproducible. fn mint_span_id(&mut self) -> String { self.next_span_seq += 1; - format!("{:016x}", self.next_span_seq) + // Nonce prefix keeps the id globally unique across turns (Langfuse + // dedupes observations by id project-wide). + format!("{}-{:016x}", self.id_prefix, self.next_span_seq) } fn open_span( @@ -231,6 +262,8 @@ impl SpanCollector { end_unix_ms: None, status: SpanStatus::Unset, attributes, + input: None, + output: None, }); (span_id, index) } @@ -269,6 +302,12 @@ impl SpanCollector { serde_json::Value::String(user.clone()), ); } + if let Some(group) = &self.ctx.session_group { + attrs.insert( + "thread.id".to_string(), + serde_json::Value::String(group.clone()), + ); + } let (id, index) = self.open_span(SpanKind::Turn, "agent.turn", None, start_unix_ms, attrs); self.turn_span_id = Some(id.clone()); self.turn_span_index = Some(index); @@ -585,6 +624,26 @@ impl SpanCollector { } } + AgentProgress::TurnContent { input, output } => { + // Attach prompt/reply to the root turn span. Held in-memory only; + // the exporter decides whether to transmit it (opt-in gate). + let index = match self.turn_span_index { + Some(idx) => idx, + None => { + self.ensure_turn_span(now_unix_ms); + self.turn_span_index.expect("turn span just created") + } + }; + if let Some(span) = self.spans.get_mut(index) { + if let Some(text) = input { + span.input = Some(serde_json::Value::String(text.clone())); + } + if let Some(text) = output { + span.output = Some(serde_json::Value::String(text.clone())); + } + } + } + AgentProgress::TurnCompleted { iterations } => { self.close_current_iteration(now_unix_ms); if let Some(index) = self.turn_span_index { diff --git a/src/openhuman/agent/progress_tracing/langfuse.rs b/src/openhuman/agent/progress_tracing/langfuse.rs index 6e9f94b779..4347fcb60b 100644 --- a/src/openhuman/agent/progress_tracing/langfuse.rs +++ b/src/openhuman/agent/progress_tracing/langfuse.rs @@ -11,9 +11,12 @@ //! never hold Langfuse keys and never hit `/api/public/ingestion` directly. //! //! Best-effort: any failure is logged and swallowed by the caller so tracing -//! never breaks a turn. Spans carry only metadata (names, kinds, timings, -//! counts) — never prompt text, tool arguments, or PII — honoring the project's -//! "never log secrets or full PII" rule. +//! never breaks a turn. By default spans carry only metadata (names, kinds, +//! timings, and non-PII token/cost figures — the latter promoted into Langfuse's +//! native `usageDetails`/`costDetails`). Prompt text and the model's reply are +//! withheld unless the operator opts in via +//! `observability.agent_tracing.capture_content`, preserving the project's +//! "never log secrets or full PII" default. use std::time::Duration; @@ -83,7 +86,7 @@ fn langfuse_metadata(span: &TraceSpan) -> Value { /// a single `trace-create` for the shared trace id followed by one /// `span-create` observation per span. Field names are Langfuse's camelCase /// (`traceId`, `startTime`, `parentObservationId`); timestamps are ISO strings. -pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value { +pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan], include_content: bool) -> Value { let mut batch: Vec = Vec::with_capacity(spans.len() + 1); // One trace-create for the run, keyed by the shared trace id. Prefer the @@ -93,15 +96,25 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value { .find(|s| s.parent_span_id.is_none()) .or_else(|| spans.first()) { + let mut trace_body = json!({ + "id": root.trace_id, + "name": root.name, + "timestamp": iso_millis(root.start_unix_ms), + }); + // Attribute the trace to the user and group per-turn traces under the + // conversation via Langfuse's native `userId`/`sessionId` (read from the + // turn span's stamped attributes). + if let Some(user) = root.attributes.get("user.id").and_then(Value::as_str) { + trace_body["userId"] = json!(user); + } + if let Some(group) = root.attributes.get("thread.id").and_then(Value::as_str) { + trace_body["sessionId"] = json!(group); + } batch.push(json!({ "id": new_event_id(), "type": "trace-create", "timestamp": iso_millis(root.start_unix_ms), - "body": { - "id": root.trace_id, - "name": root.name, - "timestamp": iso_millis(root.start_unix_ms), - }, + "body": trace_body, })); } @@ -120,9 +133,29 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value { if let Some(parent) = &span.parent_span_id { body["parentObservationId"] = json!(parent); } + // Prompt/reply content is transmitted only when the caller opted in + // (`observability.agent_tracing.capture_content`); otherwise it never + // leaves the device even though it may sit on the in-memory span. + if include_content { + if let Some(input) = &span.input { + body["input"] = input.clone(); + } + if let Some(output) = &span.output { + body["output"] = output.clone(); + } + } + // A span carrying `gen_ai.usage.*` attributes (today only the root turn + // span) is emitted as a Langfuse `generation` so the UI renders native + // token usage + cost instead of burying them in metadata. Token counts + // and cost are non-PII, so this promotion is unconditional. + let event_type = if apply_usage_fields(&mut body, span) { + "generation-create" + } else { + "span-create" + }; batch.push(json!({ "id": new_event_id(), - "type": "span-create", + "type": event_type, "timestamp": iso_millis(span.start_unix_ms), "body": body, })); @@ -131,6 +164,46 @@ pub(crate) fn spans_to_langfuse_batch(spans: &[TraceSpan]) -> Value { json!({ "batch": batch }) } +/// Promote a span's `gen_ai.usage.*` / `gen_ai.request.model` attributes into +/// Langfuse's native `model` / `usageDetails` / `costDetails` fields so the +/// trace surfaces real token counts and cost (Langfuse only renders these on +/// `generation` observations). Returns `true` when usage was found, so the +/// caller emits the span as a `generation-create`. Only token/cost figures are +/// touched — never prompt text or PII. +fn apply_usage_fields(body: &mut Value, span: &TraceSpan) -> bool { + let attrs = &span.attributes; + let input = attrs + .get("gen_ai.usage.input_tokens") + .and_then(Value::as_u64); + let output = attrs + .get("gen_ai.usage.output_tokens") + .and_then(Value::as_u64); + if input.is_none() && output.is_none() { + return false; + } + let input = input.unwrap_or(0); + let output = output.unwrap_or(0); + let mut usage = Map::new(); + usage.insert("input".to_string(), json!(input)); + usage.insert("output".to_string(), json!(output)); + usage.insert("total".to_string(), json!(input.saturating_add(output))); + if let Some(cached) = attrs + .get("gen_ai.usage.cached_input_tokens") + .and_then(Value::as_u64) + .filter(|c| *c > 0) + { + usage.insert("cache_read_input_tokens".to_string(), json!(cached)); + } + body["usageDetails"] = Value::Object(usage); + if let Some(model) = attrs.get("gen_ai.request.model").and_then(Value::as_str) { + body["model"] = json!(model); + } + if let Some(cost) = attrs.get("gen_ai.usage.cost_usd").and_then(Value::as_f64) { + body["costDetails"] = json!({ "total": cost }); + } + true +} + /// Fresh per-event id. Langfuse dedupes ingestion events by this id, so it must /// be unique per event (distinct from the observation/trace id in `body`). fn new_event_id() -> String { @@ -152,7 +225,8 @@ pub(crate) async fn push_spans(config: &Config, spans: &[TraceSpan]) -> Result<( )); } let token = require_live_session_token(config)?; - let batch = spans_to_langfuse_batch(spans); + let include_content = config.observability.agent_tracing.capture_content; + let batch = spans_to_langfuse_batch(spans, include_content); let span_count = spans.len(); tracing::debug!( @@ -173,19 +247,36 @@ pub(crate) async fn push_spans(config: &Config, spans: &[TraceSpan]) -> Result<( .map_err(|err| format!("POST {url} failed: {err}"))?; let status = response.status(); - // Langfuse returns 207 Multi-Status on partial success; `is_success()` - // covers the whole 2xx range including that. - if status.is_success() { + let body = response.text().await.unwrap_or_default(); + if !status.is_success() { + let excerpt: String = body.chars().take(200).collect(); + return Err(format!("Langfuse ingestion returned {status}: {excerpt}")); + } + // Langfuse returns 207 Multi-Status even when individual events are rejected + // — the failures live in the response `errors` array, not the HTTP status. + // Surface them (a partial rejection is logged but never fails the turn). + let rejected = serde_json::from_str::(&body) + .ok() + .and_then(|v| v.get("errors").and_then(Value::as_array).cloned()) + .filter(|errs| !errs.is_empty()); + if let Some(errs) = rejected { + let excerpt: String = serde_json::to_string(&errs) + .unwrap_or_default() + .chars() + .take(400) + .collect(); + tracing::warn!( + target: LOG_TARGET, + "[agent-tracing] Langfuse ({status}) rejected {} of {span_count} span event(s): {excerpt}", + errs.len() + ); + } else { tracing::debug!( target: LOG_TARGET, "[agent-tracing] pushed {span_count} spans to Langfuse ({status})" ); - Ok(()) - } else { - let body = response.text().await.unwrap_or_default(); - let excerpt: String = body.chars().take(200).collect(); - Err(format!("Langfuse ingestion returned {status}: {excerpt}")) } + Ok(()) } #[cfg(test)] @@ -217,6 +308,8 @@ mod tests { end_unix_ms: end, status, attributes, + input: None, + output: None, } } @@ -273,7 +366,7 @@ mod tests { Some(1_500), ), ]; - let payload = spans_to_langfuse_batch(&spans); + let payload = spans_to_langfuse_batch(&spans, false); let batch = payload["batch"].as_array().expect("batch array"); assert_eq!(batch.len(), 3, "one trace-create + two span-create"); @@ -300,6 +393,56 @@ mod tests { assert_ne!(batch[1]["id"], batch[1]["body"]["id"]); } + #[test] + fn usage_span_becomes_generation_and_content_is_gated() { + let mut turn = span( + "trace-1", + "root", + None, + "agent.turn", + SpanKind::Turn, + SpanStatus::Ok, + 1_000, + Some(2_000), + ); + turn.attributes.clear(); + turn.attributes + .insert("gen_ai.request.model".into(), json!("claude-x")); + turn.attributes + .insert("gen_ai.usage.input_tokens".into(), json!(100)); + turn.attributes + .insert("gen_ai.usage.output_tokens".into(), json!(20)); + turn.attributes + .insert("gen_ai.usage.cost_usd".into(), json!(0.0123)); + turn.input = Some(json!("what is 2+2?")); + turn.output = Some(json!("4")); + let spans = vec![turn]; + + // Content OFF (default): span is promoted to a generation with native + // usage + cost, but prompt/reply are withheld. + let off = spans_to_langfuse_batch(&spans, false); + let obs = &off["batch"][1]; + assert_eq!(obs["type"], "generation-create"); + assert_eq!(obs["body"]["model"], "claude-x"); + assert_eq!(obs["body"]["usageDetails"]["input"], 100); + assert_eq!(obs["body"]["usageDetails"]["output"], 20); + assert_eq!(obs["body"]["usageDetails"]["total"], 120); + assert_eq!(obs["body"]["costDetails"]["total"], 0.0123); + assert!( + obs["body"].get("input").is_none(), + "prompt must be withheld when capture_content is off" + ); + assert!(obs["body"].get("output").is_none()); + + // Content ON: prompt/reply included, usage/cost unchanged. + let on = spans_to_langfuse_batch(&spans, true); + let obs = &on["batch"][1]; + assert_eq!(obs["type"], "generation-create"); + assert_eq!(obs["body"]["input"], "what is 2+2?"); + assert_eq!(obs["body"]["output"], "4"); + assert_eq!(obs["body"]["costDetails"]["total"], 0.0123); + } + #[tokio::test] async fn empty_spans_push_is_ok_noop() { let config = Config::default(); diff --git a/src/openhuman/agent/progress_tracing/tests.rs b/src/openhuman/agent/progress_tracing/tests.rs index ecceaad94e..704ddbd08e 100644 --- a/src/openhuman/agent/progress_tracing/tests.rs +++ b/src/openhuman/agent/progress_tracing/tests.rs @@ -52,6 +52,7 @@ fn tracing_config_round_trips_through_json() { enabled: true, backend: AgentTracingBackend::Langfuse, export_path: Some("/tmp/spans.ndjson".to_string()), + capture_content: false, }; let s = serde_json::to_string(&cfg).unwrap(); let back: AgentTracingConfig = serde_json::from_str(&s).unwrap(); @@ -570,6 +571,7 @@ fn export_disabled_is_a_noop_and_writes_nothing() { enabled: false, backend: AgentTracingBackend::Otel, export_path: Some(path.to_string_lossy().to_string()), + capture_content: false, }; export_spans(&cfg, &one_turn_spans()); assert!( @@ -588,6 +590,7 @@ fn export_appends_ndjson_to_the_configured_file() { enabled: true, backend: AgentTracingBackend::Otel, export_path: Some(path.to_string_lossy().to_string()), + capture_content: false, }; let spans = one_turn_spans(); export_spans(&cfg, &spans); @@ -611,6 +614,7 @@ fn export_with_no_path_does_not_panic() { enabled: true, backend: AgentTracingBackend::Langfuse, export_path: None, + capture_content: false, }; export_spans(&cfg, &one_turn_spans()); export_spans(&cfg, &[]); // empty slice short-circuits. @@ -643,6 +647,7 @@ async fn export_run_trace_otel_backend_uses_local_sink() { enabled: true, backend: AgentTracingBackend::Otel, export_path: Some(path.to_string_lossy().to_string()), + capture_content: false, }; export_run_trace(&config, &one_turn_spans()).await; diff --git a/src/openhuman/channels/providers/web/progress_bridge.rs b/src/openhuman/channels/providers/web/progress_bridge.rs index 7c3baf5dca..b27ea6c87d 100644 --- a/src/openhuman/channels/providers/web/progress_bridge.rs +++ b/src/openhuman/channels/providers/web/progress_bridge.rs @@ -160,11 +160,15 @@ pub(crate) fn spawn_progress_bridge( use crate::openhuman::agent::progress_tracing::{ trace_session_id, SpanCollector, TraceContext, }; - let session_id = trace_session_id(metadata.session_id, &thread_id); - Some(SpanCollector::new(TraceContext::new( - session_id, - Some(client_id.clone()), - ))) + // One trace per turn: the trace id is unique per request, while the + // thread id rides along as the Langfuse `sessionId` so a + // conversation's per-turn traces still group under one session. + let base = trace_session_id(metadata.session_id, &thread_id); + let trace_id = format!("{base}:{request_id}"); + Some(SpanCollector::new( + TraceContext::new(trace_id, Some(client_id.clone())) + .with_session_group(thread_id.clone()), + )) } else { None }; @@ -1080,6 +1084,10 @@ pub(crate) fn spawn_progress_bridge( total_usd={total_usd:.4} client_id={client_id} thread_id={thread_id}" ); } + AgentProgress::TurnContent { .. } => { + // Prompt/reply content is attached to the trace span by the + // span collector above; the ledger/telemetry bridge ignores it. + } } } turn_state.finish(); diff --git a/src/openhuman/config/schema/load/env_overlay.rs b/src/openhuman/config/schema/load/env_overlay.rs index 96f8ddff34..ca759edeaf 100644 --- a/src/openhuman/config/schema/load/env_overlay.rs +++ b/src/openhuman/config/schema/load/env_overlay.rs @@ -531,6 +531,21 @@ impl Config { _ => {} } } + + // Opt-in: export prompt/reply content on trace spans (default off — a + // deliberate PII reversal). Token/cost export is unaffected by this flag. + if let Some(flag) = env.get("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT") { + let normalized = flag.trim().to_ascii_lowercase(); + match normalized.as_str() { + "1" | "true" | "yes" | "on" => { + self.observability.agent_tracing.capture_content = true + } + "0" | "false" | "no" | "off" => { + self.observability.agent_tracing.capture_content = false + } + _ => {} + } + } } fn apply_learning_env(&mut self, env: &E) { diff --git a/src/openhuman/config/schema/observability.rs b/src/openhuman/config/schema/observability.rs index a132b067a5..82df7bd411 100644 --- a/src/openhuman/config/schema/observability.rs +++ b/src/openhuman/config/schema/observability.rs @@ -81,6 +81,13 @@ pub struct AgentTracingConfig { /// spans are emitted to the application log at `info` level instead, so /// the export still works on read-only or sandboxed deployments. pub export_path: Option, + + /// Opt-in: include the turn's prompt (`input`) and the model's reply + /// (`output`) on exported spans. Off by default — this reverses the + /// otherwise metadata-only, PII-free posture, so it must be enabled + /// deliberately. Token/cost figures are always exported (they carry no PII) + /// regardless of this flag. + pub capture_content: bool, } impl Default for AgentTracingConfig { @@ -89,6 +96,7 @@ impl Default for AgentTracingConfig { enabled: false, backend: AgentTracingBackend::Otel, export_path: None, + capture_content: false, } } } diff --git a/src/openhuman/threads/turn_state/mirror.rs b/src/openhuman/threads/turn_state/mirror.rs index 7431c6e506..dc5fd6c6bf 100644 --- a/src/openhuman/threads/turn_state/mirror.rs +++ b/src/openhuman/threads/turn_state/mirror.rs @@ -425,6 +425,11 @@ impl TurnStateMirror { // flush per LLM call — not worth it for telemetry. false } + AgentProgress::TurnContent { .. } => { + // Prompt/reply content is consumed by the tracing exporter, not + // the turn-state snapshot; nothing to mirror, no flush. + false + } } } diff --git a/src/openhuman/workflows/run_log.rs b/src/openhuman/workflows/run_log.rs index 89d26d522d..ca201988f1 100644 --- a/src/openhuman/workflows/run_log.rs +++ b/src/openhuman/workflows/run_log.rs @@ -232,7 +232,8 @@ pub fn format_event(ev: &AgentProgress) -> Option { | AgentProgress::TaskBoardUpdated { .. } | AgentProgress::SubagentTextDelta { .. } | AgentProgress::SubagentThinkingDelta { .. } - | AgentProgress::SubagentIterationStarted { .. } => return None, + | AgentProgress::SubagentIterationStarted { .. } + | AgentProgress::TurnContent { .. } => return None, }; Some(format!( "{} {}", From ab66c7f370397c3a56a8642813108af9b245769f Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 3 Jul 2026 18:47:36 +0530 Subject: [PATCH 2/3] test(tracing): cover content/grouping/span-id-uniqueness + capture_content toggle Adds targeted unit tests for the new tracing behavior to lift diff coverage on the changed lines: - collector: TurnContent attaches input/output to the turn span; session_group is stamped as thread.id and client as user.id; span ids are globally unique across turns (regression guard for the Langfuse dedup collision). - exporter: trace-create promotes user.id/thread.id to Langfuse userId/sessionId. - run_log: TurnContent produces no log line (content stays out of the run log). - config: OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT env toggle flips the flag on/off. All six pass (cargo test --lib). --- .../agent/progress_tracing/langfuse.rs | 25 ++++++++ src/openhuman/agent/progress_tracing/tests.rs | 58 +++++++++++++++++++ src/openhuman/config/schema/load_tests.rs | 19 ++++++ src/openhuman/workflows/run_log.rs | 6 ++ 4 files changed, 108 insertions(+) diff --git a/src/openhuman/agent/progress_tracing/langfuse.rs b/src/openhuman/agent/progress_tracing/langfuse.rs index 4347fcb60b..2772b3c8c6 100644 --- a/src/openhuman/agent/progress_tracing/langfuse.rs +++ b/src/openhuman/agent/progress_tracing/langfuse.rs @@ -443,6 +443,31 @@ mod tests { assert_eq!(obs["body"]["costDetails"]["total"], 0.0123); } + #[test] + fn trace_create_carries_user_and_session_grouping() { + // The turn span's user.id / thread.id attributes are promoted onto the + // trace-create as Langfuse userId / sessionId so per-turn traces group + // under one conversation and attribute to a user. + let mut turn = span( + "trace:req-1", + "root", + None, + "agent.turn", + SpanKind::Turn, + SpanStatus::Ok, + 1_000, + Some(2_000), + ); + turn.attributes.insert("user.id".into(), json!("client-7")); + turn.attributes + .insert("thread.id".into(), json!("thread-abc")); + let payload = spans_to_langfuse_batch(&[turn], false); + let trace = &payload["batch"][0]; + assert_eq!(trace["type"], "trace-create"); + assert_eq!(trace["body"]["userId"], "client-7"); + assert_eq!(trace["body"]["sessionId"], "thread-abc"); + } + #[tokio::test] async fn empty_spans_push_is_ok_noop() { let config = Config::default(); diff --git a/src/openhuman/agent/progress_tracing/tests.rs b/src/openhuman/agent/progress_tracing/tests.rs index 704ddbd08e..f1a3283466 100644 --- a/src/openhuman/agent/progress_tracing/tests.rs +++ b/src/openhuman/agent/progress_tracing/tests.rs @@ -658,3 +658,61 @@ async fn export_run_trace_otel_backend_uses_local_sink() { ); let _ = std::fs::remove_dir_all(&dir); } + +// ── Route A: content + grouping + span-id uniqueness ──────────────────────── + +#[test] +fn turn_content_attaches_input_output_to_turn_span() { + let c = collect(&[ + (AgentProgress::TurnStarted, 1_000), + ( + AgentProgress::TurnContent { + input: Some("what is your favorite color?".to_string()), + output: Some("i'm partial to teal".to_string()), + }, + 1_100, + ), + ]); + let turn = find(c.spans(), "agent.turn"); + assert_eq!( + turn.input.as_ref().and_then(|v| v.as_str()), + Some("what is your favorite color?"), + "TurnContent input must land on the turn span" + ); + assert_eq!( + turn.output.as_ref().and_then(|v| v.as_str()), + Some("i'm partial to teal") + ); +} + +#[test] +fn turn_span_stamps_user_and_thread_grouping_attributes() { + let mut c = SpanCollector::new( + TraceContext::new("trace:req-1", Some("client-7".to_string())) + .with_session_group("thread-abc"), + ); + c.record(&AgentProgress::TurnStarted, 1_000); + let turn = find(c.spans(), "agent.turn"); + assert_eq!( + turn.attributes.get("user.id").and_then(|v| v.as_str()), + Some("client-7") + ); + assert_eq!( + turn.attributes.get("thread.id").and_then(|v| v.as_str()), + Some("thread-abc"), + "session_group must be stamped as thread.id for the Langfuse sessionId" + ); +} + +#[test] +fn span_ids_are_unique_across_turns() { + // Two separate collectors (two turns) must not reuse span ids, or Langfuse + // dedupes their observations onto whichever trace claimed the id first. + let a = collect(&[(AgentProgress::TurnStarted, 1)]); + let b = collect(&[(AgentProgress::TurnStarted, 1)]); + assert_ne!( + find(a.spans(), "agent.turn").span_id, + find(b.spans(), "agent.turn").span_id, + "span ids must be globally unique across turns" + ); +} diff --git a/src/openhuman/config/schema/load_tests.rs b/src/openhuman/config/schema/load_tests.rs index 38893dfcc2..e93816127d 100644 --- a/src/openhuman/config/schema/load_tests.rs +++ b/src/openhuman/config/schema/load_tests.rs @@ -524,6 +524,25 @@ impl EnvLookup for HashMapEnv { } } +#[test] +fn env_overlay_toggles_agent_tracing_capture_content() { + // Off by default; the opt-in env var enables prompt/reply export. + let mut cfg = Config::default(); + assert!(!cfg.observability.agent_tracing.capture_content); + cfg.apply_env_overlay_with( + &HashMapEnv::new().with("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT", "true"), + ); + assert!(cfg.observability.agent_tracing.capture_content); + + // An explicit falsy value turns it back off. + let mut cfg = Config::default(); + cfg.observability.agent_tracing.capture_content = true; + cfg.apply_env_overlay_with( + &HashMapEnv::new().with("OPENHUMAN_AGENT_TRACING_CAPTURE_CONTENT", "off"), + ); + assert!(!cfg.observability.agent_tracing.capture_content); +} + #[test] fn env_overlay_model_only_honours_namespaced_var() { // Both set → OPENHUMAN_MODEL wins; bare MODEL is ignored even when diff --git a/src/openhuman/workflows/run_log.rs b/src/openhuman/workflows/run_log.rs index ca201988f1..f5b015138f 100644 --- a/src/openhuman/workflows/run_log.rs +++ b/src/openhuman/workflows/run_log.rs @@ -839,6 +839,12 @@ mod tests { iteration: 1 }) .is_none()); + // Content (prompt/reply) rides its own event and is never logged here. + assert!(format_event(&AgentProgress::TurnContent { + input: Some("secret prompt".into()), + output: Some("secret reply".into()), + }) + .is_none()); let line = format_event(&AgentProgress::ToolCallStarted { call_id: "c1".into(), tool_name: "codegraph_search".into(), From 18c116139df365c9e2940e11c1a0e19042e0eda8 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 3 Jul 2026 21:16:21 +0530 Subject: [PATCH 3/3] fix(bin): cover new AgentProgress::TurnContent in harness_subagent_audit match MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tracing feature adds the AgentProgress::TurnContent variant; the harness-subagent-audit bin exhaustively matches AgentProgress and didn't cover it, so `cargo check --all-targets` (CI's Rust Quality clippy) failed E0004 even though `cargo check --lib` passed (it doesn't build bins). Fold TurnContent into the ignored-variants arm — an audit tool has no use for prompt/reply content. --- src/bin/harness_subagent_audit.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/bin/harness_subagent_audit.rs b/src/bin/harness_subagent_audit.rs index 890b6dfbb2..3e1ce7a5a0 100644 --- a/src/bin/harness_subagent_audit.rs +++ b/src/bin/harness_subagent_audit.rs @@ -636,7 +636,8 @@ async fn drain_progress( | AgentProgress::ToolCallArgsDelta { .. } | AgentProgress::TaskBoardUpdated { .. } | AgentProgress::TurnCostUpdated { .. } - | AgentProgress::TurnStarted => {} + | AgentProgress::TurnStarted + | AgentProgress::TurnContent { .. } => {} } } }