From 6e0b3f480d29739cc36b2a60c6a60932e7d8dd17 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Sun, 5 Jul 2026 03:15:33 -0700 Subject: [PATCH] feat(c4): export Langfuse usage data from journal observations (#4562) --- .../C4-journal-progress-parity-plan.md | 6 +- src/openhuman/agent/progress_tracing.rs | 32 ++- .../agent/progress_tracing/langfuse.rs | 225 +++++++++++++++++- .../channels/providers/web/progress_bridge.rs | 29 ++- 4 files changed, 279 insertions(+), 13 deletions(-) diff --git a/docs/tinyagents-full-migration-plan/C4-journal-progress-parity-plan.md b/docs/tinyagents-full-migration-plan/C4-journal-progress-parity-plan.md index 73fd65eb3..fac22dc7f 100644 --- a/docs/tinyagents-full-migration-plan/C4-journal-progress-parity-plan.md +++ b/docs/tinyagents-full-migration-plan/C4-journal-progress-parity-plan.md @@ -161,7 +161,11 @@ span-projection slice. (log divergences), matching the C1 `session_shadow_reads` pattern. **STARTED** — the web bridge now keeps live export as-is and logs a structural journal-projection shadow comparison keyed by the durable journal run id. -4. **S4 — Langfuse swap** to the crate exporter (§3); delete +4. **S4 — Langfuse swap** to the crate exporter (§3). **STARTED** — when the + web bridge can read a durable run journal for S3 shadow comparison, the + remote `share_usage_data` push now sends those `AgentObservation`s through + the crate `LangfuseClient`; live spans remain the local tracing sink and + fallback until S3 shadow parity is release-proven. Later delete `progress_tracing/langfuse.rs` (~825 + tests). 5. **S5 — delete `progress_tracing.rs` + `SpanCollector`** once S3 shadow shows no divergence for one release **and** V3 projection parity holds (the doc 07 diff --git a/src/openhuman/agent/progress_tracing.rs b/src/openhuman/agent/progress_tracing.rs index 779ae9239..c13da876d 100644 --- a/src/openhuman/agent/progress_tracing.rs +++ b/src/openhuman/agent/progress_tracing.rs @@ -1395,7 +1395,9 @@ pub(crate) fn export_spans(config: &AgentTracingConfig, spans: &[TraceSpan]) { /// the current backend host, authed with the session bearer (see /// [`langfuse::push_spans`]). A failure (no live session, network, rejected /// batch) just logs; there is no local fallback, since sharing and local -/// export are distinct opt-ins. +/// export are distinct opt-ins. Web-channel turns that successfully read a +/// durable tinyagents journal should call [`export_run_trace_from_journal`] +/// instead, so the remote push uses the crate-owned observation exporter. /// 2. **Local exporter** (`observability.agent_tracing.enabled`, opt-in): append /// OTel/Langfuse-format NDJSON to the configured file or the app log via /// [`export_spans`]. @@ -1418,5 +1420,33 @@ pub(crate) async fn export_run_trace(config: &Config, spans: &[TraceSpan]) { } } +/// Export a completed run when durable tinyagents observations are available. +/// Remote usage-data sharing uses the crate Langfuse exporter over the journal; +/// local tracing still writes the live spans until the migration deletes the +/// legacy span collector/exporter path. +pub(crate) async fn export_run_trace_from_journal( + config: &Config, + trace_ctx: &TraceContext, + observations: &[tinyagents::harness::observability::AgentObservation], + live_spans: &[TraceSpan], +) { + if observations.is_empty() && live_spans.is_empty() { + return; + } + let observability = &config.observability; + + if observability.share_usage_data && !observations.is_empty() { + if let Err(err) = langfuse::push_observations(config, trace_ctx, observations).await { + log::warn!("[agent-tracing] Langfuse journal usage-data push failed ({err})"); + } + } else if observability.share_usage_data { + log::debug!("[agent-tracing] no journal observations for Langfuse usage-data push"); + } + + if observability.agent_tracing.enabled && !live_spans.is_empty() { + export_spans(&observability.agent_tracing, live_spans); + } +} + #[cfg(test)] mod tests; diff --git a/src/openhuman/agent/progress_tracing/langfuse.rs b/src/openhuman/agent/progress_tracing/langfuse.rs index e669cab7e..d908e0046 100644 --- a/src/openhuman/agent/progress_tracing/langfuse.rs +++ b/src/openhuman/agent/progress_tracing/langfuse.rs @@ -14,20 +14,22 @@ //! never breaks a turn. Spans always carry metadata (names, kinds, timings, //! and non-PII token/cost figures — the latter promoted into Langfuse's native //! `usageDetails`/`costDetails`). Prompt/reply text and truncated tool I/O -//! ride along while `observability.agent_tracing.capture_content` is on (its -//! default); setting it to `false` withholds all content and falls back to -//! the metadata-only posture. +//! ride along only while `observability.agent_tracing.capture_content` is on; +//! with the default off, content is withheld and export stays metadata-only. +use std::borrow::Cow; use std::time::Duration; use serde_json::{json, Map, Value}; +use tinyagents::harness::events::AgentEvent; +use tinyagents::harness::observability::{AgentObservation, LangfuseClient, LangfuseTraceConfig}; use crate::api::config::effective_backend_api_url; use crate::api::jwt::bearer_authorization_value; use crate::openhuman::config::Config; use crate::openhuman::credentials::session_support::require_live_session_token; -use super::{SpanStatus, TraceSpan}; +use super::{SpanStatus, TraceContext, TraceSpan}; const LOG_TARGET: &str = "agent-tracing::langfuse"; /// Backend proxy route for Langfuse ingestion (relative to the backend origin). @@ -310,6 +312,119 @@ fn new_event_id() -> String { uuid::Uuid::new_v4().to_string() } +fn trace_config_from_context(trace_ctx: &TraceContext, environment: &str) -> LangfuseTraceConfig { + let mut metadata = Map::new(); + if let Some(client_id) = &trace_ctx.client_id { + metadata.insert("client.id".into(), json!(client_id)); + } + if let Some(agent_id) = &trace_ctx.agent_id { + metadata.insert("agent.id".into(), json!(agent_id)); + } + if let Some(source) = &trace_ctx.channel_source { + metadata.insert("channel.source".into(), json!(source)); + } + metadata.insert("run_type".into(), json!(trace_ctx.run_type.as_str())); + metadata.insert("app.version".into(), json!(env!("CARGO_PKG_VERSION"))); + + let mut tags = vec![format!("run:{}", trace_ctx.run_type.as_str())]; + if let Some(source) = &trace_ctx.channel_source { + tags.push(format!("source:{source}")); + } + + LangfuseTraceConfig { + trace_id: Some(trace_ctx.session_id.clone()), + name: Some(match &trace_ctx.agent_id { + Some(agent_id) => format!("agent.turn:{agent_id}"), + None => "agent.turn".to_string(), + }), + user_id: trace_ctx.user_id.clone(), + session_id: trace_ctx + .session_group + .clone() + .or_else(|| Some(trace_ctx.session_id.clone())), + environment: Some(environment.to_string()), + release: Some(env!("CARGO_PKG_VERSION").to_string()), + version: Some(env!("CARGO_PKG_VERSION").to_string()), + tags, + metadata: Value::Object(metadata), + } +} + +fn observations_for_export<'a>( + trace_ctx: &TraceContext, + observations: &'a [AgentObservation], +) -> Cow<'a, [AgentObservation]> { + if trace_ctx.capture_content { + return Cow::Borrowed(observations); + } + + Cow::Owned( + observations + .iter() + .cloned() + .map(strip_observation_content) + .collect(), + ) +} + +fn strip_observation_content(mut observation: AgentObservation) -> AgentObservation { + match &mut observation.event { + AgentEvent::ModelCompleted { input, output, .. } + | AgentEvent::ToolCompleted { input, output, .. } => { + *input = None; + *output = None; + } + _ => {} + } + observation +} + +/// Push durable journal observations through the tinyagents crate Langfuse +/// exporter. The journal is already redacted before persistence, and this +/// exporter additionally strips model/tool payloads unless `capture_content` +/// is explicitly enabled. +pub(crate) async fn push_observations( + config: &Config, + trace_ctx: &TraceContext, + observations: &[AgentObservation], +) -> Result<(), String> { + if observations.is_empty() { + return Ok(()); + } + let url = ingestion_url(config); + if !url.starts_with("http") { + return Err(format!( + "could not resolve Langfuse ingestion URL from backend host (got {url:?})" + )); + } + let token = require_live_session_token(config)?; + let environment = environment_for_base(&url); + let trace = trace_config_from_context(trace_ctx, environment); + let observation_count = observations.len(); + let observations = observations_for_export(trace_ctx, observations); + + tracing::debug!( + target: LOG_TARGET, + "[agent-tracing] pushing {observation_count} journal observations to Langfuse at {url}" + ); + + let client = LangfuseClient::proxy(url, token) + .map_err(|err| format!("Langfuse client setup failed: {err}"))?; + tokio::time::timeout( + PUSH_TIMEOUT, + client.send_observations(trace, observations.as_ref()), + ) + .await + .map_err(|_| format!("Langfuse journal push timed out after {PUSH_TIMEOUT:?}"))? + .map_err(|err| format!("Langfuse journal ingestion failed: {err}"))?; + + tracing::debug!( + target: LOG_TARGET, + "[agent-tracing] pushed {observation_count} journal observations to Langfuse" + ); + Ok(()) +} + /// Push `spans` to the co-hosted Langfuse server. Resolves the endpoint from the /// current backend host and authenticates with the live session bearer. Returns /// `Err` (for the caller to log + fall back) when there is no live session, the @@ -386,6 +501,8 @@ mod tests { use std::collections::BTreeMap; use crate::openhuman::agent::progress_tracing::SpanKind; + use tinyagents::harness::ids::{CallId, EventId, RunId}; + use tinyagents::harness::usage::Usage; fn span( trace: &str, @@ -414,6 +531,18 @@ mod tests { } } + fn obs(offset: u64, event: AgentEvent) -> AgentObservation { + AgentObservation { + event_id: EventId::new(format!("run-1-evt-{offset}")), + run_id: RunId::new("run-1"), + parent_run_id: None, + root_run_id: RunId::new("run-1"), + offset, + ts_ms: 1_000 + offset, + event, + } + } + #[test] fn ingestion_url_uses_backend_origin_and_ingestion_path() { let mut config = Config::default(); @@ -437,6 +566,94 @@ mod tests { ); } + #[test] + fn trace_config_from_context_matches_span_trace_attribution() { + let ctx = TraceContext::new("trace:req-1", Some("user-1".to_string())) + .with_session_group("thread-abc") + .with_client_id("socket-abc") + .with_agent_id("researcher") + .with_channel_source("chat") + .with_run_type(crate::openhuman::agent::progress_tracing::RunType::InteractiveChat); + + let trace = trace_config_from_context(&ctx, "staging"); + assert_eq!(trace.trace_id.as_deref(), Some("trace:req-1")); + assert_eq!(trace.name.as_deref(), Some("agent.turn:researcher")); + assert_eq!(trace.user_id.as_deref(), Some("user-1")); + assert_eq!(trace.session_id.as_deref(), Some("thread-abc")); + assert_eq!(trace.environment.as_deref(), Some("staging")); + assert_eq!(trace.tags, vec!["run:interactive_chat", "source:chat"]); + assert_eq!(trace.metadata["client.id"], "socket-abc"); + assert_eq!(trace.metadata["agent.id"], "researcher"); + assert_eq!(trace.metadata["channel.source"], "chat"); + assert_eq!(trace.metadata["run_type"], "interactive_chat"); + assert_eq!(trace.metadata["app.version"], env!("CARGO_PKG_VERSION")); + } + + #[test] + fn journal_observation_content_follows_capture_gate() { + let observations = vec![ + obs( + 1, + AgentEvent::ModelCompleted { + call_id: CallId::new("model-1"), + started_at_ms: Some(1_000), + usage: Some(Usage::new(10, 3)), + input: Some(json!([{"role": "user", "content": "secret prompt"}])), + output: Some(json!({"role": "assistant", "content": "secret reply"})), + }, + ), + obs( + 2, + AgentEvent::ToolCompleted { + call_id: CallId::new("tool-1"), + tool_name: "search".to_string(), + started_at_ms: Some(1_010), + input: Some(json!({"query": "secret"})), + output: Some(json!("secret result")), + duration_ms: Some(20), + output_bytes: Some(13), + error: None, + }, + ), + ]; + + let off_ctx = TraceContext::new("trace:req-1", None); + let filtered = observations_for_export(&off_ctx, &observations); + assert!(matches!(filtered, Cow::Owned(_))); + match &filtered[0].event { + AgentEvent::ModelCompleted { input, output, .. } => { + assert!(input.is_none()); + assert!(output.is_none()); + } + other => panic!("unexpected event: {other:?}"), + } + match &filtered[1].event { + AgentEvent::ToolCompleted { input, output, .. } => { + assert!(input.is_none()); + assert!(output.is_none()); + } + other => panic!("unexpected event: {other:?}"), + } + match &observations[0].event { + AgentEvent::ModelCompleted { input, output, .. } => { + assert!(input.is_some(), "source journal observation stays intact"); + assert!(output.is_some(), "source journal observation stays intact"); + } + other => panic!("unexpected event: {other:?}"), + } + + let on_ctx = TraceContext::new("trace:req-1", None).with_capture_content(true); + let passthrough = observations_for_export(&on_ctx, &observations); + assert!(matches!(passthrough, Cow::Borrowed(_))); + match &passthrough[1].event { + AgentEvent::ToolCompleted { input, output, .. } => { + assert_eq!(input.as_ref(), Some(&json!({"query": "secret"}))); + assert_eq!(output.as_ref(), Some(&json!("secret result"))); + } + other => panic!("unexpected event: {other:?}"), + } + } + #[test] fn iso_millis_formats_epoch_as_rfc3339() { // 2021-01-01T00:00:00Z = 1_609_459_200_000 ms. diff --git a/src/openhuman/channels/providers/web/progress_bridge.rs b/src/openhuman/channels/providers/web/progress_bridge.rs index 8a49c1b31..e2f2de89d 100644 --- a/src/openhuman/channels/providers/web/progress_bridge.rs +++ b/src/openhuman/channels/providers/web/progress_bridge.rs @@ -154,7 +154,7 @@ async fn shadow_compare_journal_projection( trace_ctx: crate::openhuman::agent::progress_tracing::TraceContext, max_iterations: u32, live_spans: &[crate::openhuman::agent::progress_tracing::TraceSpan], -) { +) -> Option> { let Some(journal_run_id) = crate::openhuman::tinyagents::journal::take_request_journal_run(request_id) else { @@ -162,7 +162,7 @@ async fn shadow_compare_journal_projection( "[agent-tracing][journal-shadow] no journal run registered request_id={}", request_id ); - return; + return None; }; let observations = match crate::openhuman::tinyagents::journal::read_run_events( @@ -178,7 +178,7 @@ async fn shadow_compare_journal_projection( request_id, journal_run_id ); - return; + return None; } }; if observations.is_empty() { @@ -187,7 +187,7 @@ async fn shadow_compare_journal_projection( request_id, journal_run_id ); - return; + return None; } let projected = @@ -218,6 +218,7 @@ async fn shadow_compare_journal_projection( projected_sig ); } + Some(observations) } /// Spawn a background task that reads [`AgentProgress`] events from the @@ -1316,16 +1317,30 @@ pub(crate) fn spawn_progress_bridge( if let Some(mut collector) = span_collector.take() { collector.finish(unix_epoch_ms()); let live_spans = collector.spans().to_vec(); - if let Some(trace_ctx) = journal_trace_ctx.take() { + let journal_export = if let Some(trace_ctx) = journal_trace_ctx.take() { shadow_compare_journal_projection( &request_id, - trace_ctx, + trace_ctx.clone(), parent_max_iterations, &live_spans, ) + .await + .map(|observations| (trace_ctx, observations)) + } else { + None + }; + if let Some((trace_ctx, observations)) = journal_export { + crate::openhuman::agent::progress_tracing::export_run_trace_from_journal( + &config, + &trace_ctx, + &observations, + &live_spans, + ) .await; + } else { + crate::openhuman::agent::progress_tracing::export_run_trace(&config, &live_spans) + .await; } - crate::openhuman::agent::progress_tracing::export_run_trace(&config, &live_spans).await; } log::debug!(