feat(c4): add journal Langfuse run telemetry (#4568)

docs/tinyagents-full-migration-plan/C4-journal-progress-parity-plan.md

@@ -165,7 +165,10 @@ span-projection slice.
    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
+   fallback until S3 shadow parity is release-proven. The OpenHuman wrapper also
+   injects aggregate `run.total` usage/cost from the durable run ledger so
+   journal-backed Langfuse export preserves the top-level cost signal while
+   per-call charged USD awaits a journal-native event. 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

src/openhuman/agent/progress_tracing.rs

@@ -1428,6 +1428,7 @@ pub(crate) async fn export_run_trace_from_journal(
     config: &Config,
     trace_ctx: &TraceContext,
     observations: &[tinyagents::harness::observability::AgentObservation],
+    run_telemetry: Option<&crate::openhuman::session_db::run_ledger::RunTelemetry>,
     live_spans: &[TraceSpan],
 ) {
     if observations.is_empty() && live_spans.is_empty() {
@@ -1436,7 +1437,9 @@ pub(crate) async fn export_run_trace_from_journal(
     let observability = &config.observability;
 
     if observability.share_usage_data && !observations.is_empty() {
-        if let Err(err) = langfuse::push_observations(config, trace_ctx, observations).await {
+        if let Err(err) =
+            langfuse::push_observations(config, trace_ctx, observations, run_telemetry).await
+        {
             log::warn!("[agent-tracing] Langfuse journal usage-data push failed ({err})");
         }
     } else if observability.share_usage_data {

src/openhuman/agent/progress_tracing/langfuse.rs

@@ -28,6 +28,7 @@ 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 crate::openhuman::session_db::run_ledger::RunTelemetry;
 
 use super::{SpanStatus, TraceContext, TraceSpan};
 
@@ -379,6 +380,86 @@ fn strip_observation_content(mut observation: AgentObservation) -> AgentObservat
     observation
 }
 
+fn insert_run_telemetry_generation(payload: &mut Value, telemetry: Option<&RunTelemetry>) -> bool {
+    let Some(telemetry) = telemetry else {
+        return false;
+    };
+    if telemetry.input_tokens == 0 && telemetry.output_tokens == 0 && telemetry.cost_usd == 0.0 {
+        return false;
+    }
+
+    let Some(batch) = payload.get_mut("batch").and_then(Value::as_array_mut) else {
+        return false;
+    };
+    let Some(trace_id) = batch
+        .first()
+        .and_then(|event| event.get("body"))
+        .and_then(|body| body.get("id"))
+        .and_then(Value::as_str)
+        .map(str::to_string)
+    else {
+        return false;
+    };
+    let start_time = batch
+        .first()
+        .and_then(|event| event.get("timestamp"))
+        .and_then(Value::as_str)
+        .unwrap_or_default()
+        .to_string();
+    let end_time = batch
+        .last()
+        .and_then(|event| event.get("timestamp"))
+        .and_then(Value::as_str)
+        .unwrap_or(start_time.as_str())
+        .to_string();
+
+    let non_cached_input = telemetry
+        .input_tokens
+        .saturating_sub(telemetry.cached_input_tokens);
+    let mut body = json!({
+        "id": format!("{trace_id}:openhuman-run-telemetry"),
+        "traceId": trace_id,
+        "name": "run.total",
+        "startTime": start_time,
+        "endTime": end_time,
+        "usageDetails": {
+            "input": non_cached_input,
+            "output": telemetry.output_tokens,
+            "total": telemetry.input_tokens.saturating_add(telemetry.output_tokens),
+            "cache_read_input_tokens": telemetry.cached_input_tokens,
+        },
+        "costDetails": {
+            "total": telemetry.cost_usd,
+        },
+        "metadata": {
+            "source": "openhuman.run_telemetry",
+            "run_id": telemetry.run_id.as_str(),
+            "tool_count": telemetry.tool_count,
+        },
+    });
+    if let Some(model) = &telemetry.model {
+        body["model"] = json!(model);
+    }
+    if let Some(provider) = &telemetry.provider {
+        body["metadata"]["provider"] = json!(provider);
+    }
+    if let Some(error) = &telemetry.error {
+        body["level"] = json!("ERROR");
+        body["statusMessage"] = json!(error);
+    }
+
+    batch.insert(
+        1,
+        json!({
+            "id": new_event_id(),
+            "type": "generation-create",
+            "timestamp": body["startTime"].clone(),
+            "body": body,
+        }),
+    );
+    true
+}
+
 /// 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`
@@ -387,6 +468,7 @@ pub(crate) async fn push_observations(
     config: &Config,
     trace_ctx: &TraceContext,
     observations: &[AgentObservation],
+    run_telemetry: Option<&RunTelemetry>,
 ) -> Result<(), String> {
     if observations.is_empty() {
         return Ok(());
@@ -410,13 +492,24 @@ pub(crate) async fn push_observations(
 
     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}"))?;
+    let mut payload = client
+        .build_ingestion_batch(trace, observations.as_ref())
+        .map_err(|err| format!("Langfuse journal batch build failed: {err}"))?;
+    if insert_run_telemetry_generation(&mut payload, run_telemetry) {
+        tracing::debug!(
+            target: LOG_TARGET,
+            "[agent-tracing] added run telemetry aggregate to Langfuse journal batch"
+        );
+    } else {
+        tracing::debug!(
+            target: LOG_TARGET,
+            "[agent-tracing] no run telemetry aggregate added to Langfuse journal batch"
+        );
+    }
+    tokio::time::timeout(PUSH_TIMEOUT, client.send_batch(payload))
+        .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,
@@ -654,6 +747,61 @@ mod tests {
         }
     }
 
+    #[test]
+    fn run_telemetry_inserts_aggregate_generation() {
+        let observations = vec![obs(
+            1,
+            AgentEvent::ModelCompleted {
+                call_id: CallId::new("model-1"),
+                started_at_ms: Some(1_000),
+                usage: Some(Usage::new(100, 20)),
+                input: None,
+                output: None,
+            },
+        )];
+        let client =
+            LangfuseClient::proxy("https://api.tinyhumans.ai", "token").expect("proxy client");
+        let trace =
+            trace_config_from_context(&TraceContext::new("trace:req-1", None), "production");
+        let mut payload = client
+            .build_ingestion_batch(trace, &observations)
+            .expect("batch");
+        let telemetry = RunTelemetry {
+            run_id: "req-1".to_string(),
+            input_tokens: 120,
+            output_tokens: 30,
+            cached_input_tokens: 40,
+            cost_usd: 0.0123,
+            elapsed_ms: Some(900),
+            tool_count: 2,
+            model: Some("managed.chat-v1".to_string()),
+            provider: Some("managed".to_string()),
+            error: None,
+            updated_at: None,
+        };
+
+        assert!(insert_run_telemetry_generation(
+            &mut payload,
+            Some(&telemetry)
+        ));
+        let batch = payload["batch"].as_array().expect("batch array");
+        assert_eq!(batch[1]["type"], "generation-create");
+        let body = &batch[1]["body"];
+        assert_eq!(body["id"], "trace:req-1:openhuman-run-telemetry");
+        assert_eq!(body["name"], "run.total");
+        assert_eq!(body["traceId"], "trace:req-1");
+        assert_eq!(body["model"], "managed.chat-v1");
+        assert_eq!(body["usageDetails"]["input"], 80);
+        assert_eq!(body["usageDetails"]["output"], 30);
+        assert_eq!(body["usageDetails"]["total"], 150);
+        assert_eq!(body["usageDetails"]["cache_read_input_tokens"], 40);
+        assert_eq!(body["costDetails"]["total"], 0.0123);
+        assert_eq!(body["metadata"]["source"], "openhuman.run_telemetry");
+        assert_eq!(body["metadata"]["run_id"], "req-1");
+        assert_eq!(body["metadata"]["tool_count"], 2);
+        assert_eq!(body["metadata"]["provider"], "managed");
+    }
+
     #[test]
     fn iso_millis_formats_epoch_as_rfc3339() {
         // 2021-01-01T00:00:00Z = 1_609_459_200_000 ms.

src/openhuman/channels/providers/web/progress_bridge.rs

@@ -90,6 +90,37 @@ pub(super) fn ledger_upsert_telemetry(
     }
 }
 
+pub(super) fn ledger_get_telemetry(
+    config: &crate::openhuman::config::Config,
+    run_id: &str,
+) -> Option<crate::openhuman::session_db::run_ledger::RunTelemetry> {
+    match crate::openhuman::session_db::run_ledger::get_agent_run(config, run_id) {
+        Ok(Some(run)) => {
+            let telemetry = run.telemetry;
+            log::debug!(
+                "[run_ledger][web_channel] read telemetry run_id={} present={}",
+                run_id,
+                telemetry.is_some()
+            );
+            telemetry
+        }
+        Ok(None) => {
+            log::debug!(
+                "[run_ledger][web_channel] telemetry unavailable; run missing run_id={}",
+                run_id
+            );
+            None
+        }
+        Err(err) => {
+            log::warn!(
+                "[run_ledger][web_channel] failed to read telemetry run_id={} err={err}",
+                run_id
+            );
+            None
+        }
+    }
+}
+
 /// Build the worktree-isolation slice of a `subagent_completed`
 /// [`SubagentProgressDetail`] (#3376). An empty `changed_files` collapses to
 /// `None` so the renderer omits an empty "changed files" list rather than
@@ -1330,10 +1361,12 @@ pub(crate) fn spawn_progress_bridge(
                 None
             };
             if let Some((trace_ctx, observations)) = journal_export {
+                let run_telemetry = ledger_get_telemetry(&config, &request_id);
                 crate::openhuman::agent::progress_tracing::export_run_trace_from_journal(
                     &config,
                     &trace_ctx,
                     &observations,
+                    run_telemetry.as_ref(),
                     &live_spans,
                 )
                 .await;