feat(observability): export flow/graph runs to Langfuse with Agent-Graph-view metadata (#4499)

docs/TEST-COVERAGE-MATRIX.md

@@ -416,7 +416,7 @@ End-to-end coverage of the agent harness via the web-chat RPC surface against an
 | ------ | ------------------------- | ----- | ------------------------------------------------------------------------------------------------- | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------- |
 | 10.3.1 | Incoming Message Sync     | RU+WD | `src/openhuman/channels/tests/`, `gmail-flow.spec.ts`                                             | ✅     |                                                                                                                                                 |
 | 10.3.2 | Message Deduplication     | RU    | `src/openhuman/channels/tests/`                                                                   | ✅     |                                                                                                                                                 |
-| 10.3.3 | WhatsApp Agent Retrieval  | RU    | `src/openhuman/whatsapp_data/tools/`, `tests/json_rpc_e2e.rs::whatsapp_data_agent_tools_e2e_1341` | ✅     | Three read-only agent tools wrap the local SQLite store; ingest stays internal-only. See [`docs/whatsapp-data-flow.md`](whatsapp-data-flow.md). |
+| 10.3.3 | WhatsApp Agent Retrieval  | RU    | `src/openhuman/whatsapp_data/tools/`, `tests/json_rpc_e2e.rs::whatsapp_data_agent_tools_e2e_1341` | ✅     | Three read-only agent tools wrap the local SQLite store; ingest stays internal-only. See [`src/openhuman/whatsapp_data/README.md`](../src/openhuman/whatsapp_data/README.md). |
 | 10.3.4 | Real-Time vs Delayed Sync | RU    | `src/openhuman/channels/tests/runtime_dispatch.rs`                                                | ✅     |                                                                                                                                                 |
 
 ### 10.4 Messaging Operations

src/openhuman/flows/bus.rs

@@ -120,7 +120,11 @@ impl FlowTriggerSubscriber {
             tracing::debug!(target: "flows", %flow_id, "[flows] schedule tick for flow whose trigger is no longer `schedule` — ignoring");
             return;
         }
-        self.spawn_run(flow_id.to_string(), Value::Null);
+        self.spawn_run(
+            flow_id.to_string(),
+            Value::Null,
+            crate::openhuman::flows::FlowRunTrigger::Schedule,
+        );
     }
 
     /// `DomainEvent::ComposioTriggerReceived` — scans every enabled flow for
@@ -140,7 +144,11 @@ impl FlowTriggerSubscriber {
         for flow in flows {
             if matches_app_event(&flow, toolkit, trigger_slug) {
                 matched += 1;
-                self.spawn_run(flow.id.clone(), payload.clone());
+                self.spawn_run(
+                    flow.id.clone(),
+                    payload.clone(),
+                    crate::openhuman::flows::FlowRunTrigger::AppEvent,
+                );
             }
         }
         tracing::debug!(target: "flows", %toolkit, %trigger_slug, matched, "[flows] app_event trigger matching complete");
@@ -154,7 +162,12 @@ impl FlowTriggerSubscriber {
     /// Skips the dispatch (see [`try_acquire_dispatch`]) if a trigger-driven
     /// run for this `flow_id` is already in flight, so a fast schedule or a
     /// burst of matching `app_event`s cannot run the same flow concurrently.
-    fn spawn_run(&self, flow_id: String, input: Value) {
+    fn spawn_run(
+        &self,
+        flow_id: String,
+        input: Value,
+        trigger: crate::openhuman::flows::FlowRunTrigger,
+    ) {
         let Some(guard) = self.try_acquire_dispatch(&flow_id) else {
             tracing::debug!(target: "flows", %flow_id, "[flows] trigger: flow already running — skipping this tick");
             return;
@@ -166,7 +179,7 @@ impl FlowTriggerSubscriber {
             // on panic) by `InFlightGuard`.
             let _guard = guard;
             tracing::info!(target: "flows", %flow_id, "[flows] trigger fired — starting run");
-            match crate::openhuman::flows::ops::flows_run(&config, &flow_id, input).await {
+            match crate::openhuman::flows::ops::flows_run(&config, &flow_id, input, trigger).await {
                 Ok(_) => {
                     tracing::info!(target: "flows", %flow_id, "[flows] trigger-driven run finished")
                 }

src/openhuman/flows/mod.rs

@@ -24,4 +24,4 @@ pub use schemas::{
 // them to implement `tinyflows::caps::StateStore` without duplicating the
 // `flow_state` table's persistence logic.
 pub use store::{kv_get, kv_set};
-pub use types::{Flow, FlowRun, FlowRunStep};
+pub use types::{Flow, FlowRun, FlowRunStep, FlowRunTrigger};

src/openhuman/flows/ops.rs

@@ -13,7 +13,7 @@ use crate::openhuman::agent::turn_origin::{with_origin, AgentTurnOrigin, Trusted
 use crate::openhuman::config::Config;
 use crate::openhuman::flows::bus;
 use crate::openhuman::flows::store;
-use crate::openhuman::flows::types::FlowRunStep;
+use crate::openhuman::flows::types::{FlowRunStep, FlowRunTrigger};
 use crate::openhuman::flows::{Flow, FlowRun};
 use crate::rpc::RpcOutcome;
 
@@ -335,6 +335,65 @@ pub async fn reconcile_schedule_triggers_on_boot(config: &Config) -> Result<(),
     Ok(())
 }
 
+/// Reads a settled run's durable [`tinyflows::engine::GraphObservation`]
+/// slice back out of the per-run journal (keyed by the tinyagents-minted
+/// `graph_run_id`) and exports it to Langfuse as one trace. Best-effort by
+/// construction: any journal read failure is logged and swallowed, and the
+/// exporter itself never fails the run. Skips the journal read entirely when
+/// `observability.share_usage_data` is off.
+async fn export_run_to_langfuse(
+    config: &Config,
+    flow_name: &str,
+    flow_id: &str,
+    thread_id: &str,
+    status: &str,
+    trigger: FlowRunTrigger,
+    journal: &tinyflows::engine::InMemoryGraphEventJournal,
+    graph_run_id: &str,
+) {
+    if !config.observability.share_usage_data {
+        tracing::debug!(
+            target: "flows",
+            flow_id = %flow_id,
+            "[flows] langfuse export skipped: observability.share_usage_data is off"
+        );
+        return;
+    }
+    use tinyflows::engine::GraphEventJournal as _;
+    let observations = match journal.read_from(graph_run_id, 0).await {
+        Ok(observations) => observations,
+        Err(e) => {
+            tracing::warn!(
+                target: "flows",
+                flow_id = %flow_id,
+                %thread_id,
+                graph_run_id = %graph_run_id,
+                error = %e,
+                "[flows] langfuse export skipped: could not read run journal"
+            );
+            return;
+        }
+    };
+    tracing::debug!(
+        target: "flows",
+        flow_id = %flow_id,
+        %thread_id,
+        graph_run_id = %graph_run_id,
+        observation_count = observations.len(),
+        "[flows] exporting flow run trace to Langfuse"
+    );
+    crate::openhuman::tinyflows::langfuse_export::export_flow_run_trace(
+        config,
+        flow_name,
+        flow_id,
+        thread_id,
+        status,
+        trigger,
+        &observations,
+    )
+    .await;
+}
+
 /// Runs a saved flow end-to-end: compile → build capabilities → durable
 /// checkpointed run → record the outcome onto the flow's summary fields and
 /// into a `flow_runs` history row.
@@ -356,6 +415,7 @@ pub async fn flows_run(
     config: &Config,
     flow_id: &str,
     input: Value,
+    trigger: FlowRunTrigger,
 ) -> Result<RpcOutcome<Value>, String> {
     let flow = store::get_flow(config, flow_id)
         .map_err(|e| e.to_string())?
@@ -400,17 +460,29 @@ pub async fn flows_run(
     };
 
     let origin = workflow_origin(flow_id, flow.require_approval);
+    // Per-run in-memory journal: tinyflows records every graph event as a
+    // durable GraphObservation under the run's tinyagents run id, which the
+    // post-run Langfuse export reads back. Process-local and dropped with the
+    // run — never persisted.
+    let journal = Arc::new(tinyflows::engine::InMemoryGraphEventJournal::new());
     let run = with_origin(
         origin,
-        tinyflows::engine::run_with_checkpointer(&compiled, input, &caps, checkpointer, &thread_id),
+        tinyflows::engine::run_with_checkpointer_journaled(
+            &compiled,
+            input,
+            &caps,
+            checkpointer,
+            &thread_id,
+            journal.clone(),
+        ),
     );
-    let outcome = match tokio::time::timeout(
+    let journaled = match tokio::time::timeout(
         std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS),
         run,
     )
     .await
     {
-        Ok(Ok(outcome)) => outcome,
+        Ok(Ok(journaled)) => journaled,
         Ok(Err(e)) => {
             record_failed(&e.to_string());
             tracing::warn!(target: "flows", flow_id = %flow_id, error = %e, "[flows] flows_run: run failed");
@@ -423,6 +495,7 @@ pub async fn flows_run(
             return Err(msg);
         }
     };
+    let outcome = journaled.outcome;
 
     let status = if outcome.pending_approvals.is_empty() {
         "completed"
@@ -438,6 +511,17 @@ pub async fn flows_run(
         &outcome.pending_approvals,
         None,
     );
+    export_run_to_langfuse(
+        config,
+        &flow.name,
+        flow_id,
+        &thread_id,
+        status,
+        trigger,
+        &journal,
+        &journaled.graph_run_ids.run_id,
+    )
+    .await;
     notify_pending_approval(&flow, &thread_id, &outcome.pending_approvals);
 
     tracing::info!(
@@ -537,24 +621,28 @@ pub async fn flows_resume(
     );
 
     let origin = workflow_origin(flow_id, flow.require_approval);
+    // Same per-run journal as `flows_run`: the resumed execution mints a new
+    // tinyagents run id, so its observation slice is read under that id.
+    let journal = Arc::new(tinyflows::engine::InMemoryGraphEventJournal::new());
     let run = with_origin(
         origin,
-        tinyflows::engine::resume_with_checkpointer(
+        tinyflows::engine::resume_with_checkpointer_journaled(
             &compiled,
             &caps,
             checkpointer,
             thread_id,
             approvals,
+            journal.clone(),
         ),
     );
 
-    let outcome = match tokio::time::timeout(
+    let journaled = match tokio::time::timeout(
         std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS),
         run,
     )
     .await
     {
-        Ok(Ok(outcome)) => outcome,
+        Ok(Ok(journaled)) => journaled,
         Ok(Err(e)) => {
             let _ = store::record_run(config, flow_id, "failed");
             finish_flow_run_row(config, thread_id, "failed", &[], &[], Some(&e.to_string()));
@@ -569,6 +657,7 @@ pub async fn flows_resume(
             return Err(msg);
         }
     };
+    let outcome = journaled.outcome;
 
     let status = if outcome.pending_approvals.is_empty() {
         "completed"
@@ -584,6 +673,17 @@ pub async fn flows_resume(
         &outcome.pending_approvals,
         None,
     );
+    export_run_to_langfuse(
+        config,
+        &flow.name,
+        flow_id,
+        thread_id,
+        status,
+        FlowRunTrigger::Resume,
+        &journal,
+        &journaled.graph_run_ids.run_id,
+    )
+    .await;
     notify_pending_approval(&flow, thread_id, &outcome.pending_approvals);
 
     tracing::info!(

src/openhuman/flows/ops_tests.rs

@@ -160,9 +160,14 @@ async fn flows_run_completes_trigger_only_graph() {
         .await
         .unwrap();
 
-    let outcome = flows_run(&config, &created.value.id, json!({ "hello": "world" }))
-        .await
-        .unwrap();
+    let outcome = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "hello": "world" }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
 
     assert_eq!(outcome.value["pending_approvals"], json!([]));
     assert_eq!(
@@ -197,9 +202,14 @@ async fn flows_run_reports_pending_approval_and_blocks_downstream() {
         .await
         .unwrap();
 
-    let outcome = flows_run(&config, &created.value.id, json!({ "x": 1 }))
-        .await
-        .unwrap();
+    let outcome = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "x": 1 }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
 
     let pending = outcome.value["pending_approvals"].as_array().unwrap();
     assert!(pending.iter().any(|v| v == "gate"));
@@ -224,7 +234,7 @@ async fn flows_get_missing_flow_errors() {
 async fn flows_run_missing_flow_errors() {
     let tmp = TempDir::new().unwrap();
     let config = test_config(&tmp);
-    let err = flows_run(&config, "missing", json!({}))
+    let err = flows_run(&config, "missing", json!({}), FlowRunTrigger::Rpc)
         .await
         .expect_err("must error");
     assert!(err.contains("not found"));
@@ -251,7 +261,7 @@ async fn flows_run_records_failed_status_when_a_node_errors() {
         .await
         .unwrap();
 
-    let err = flows_run(&config, &created.value.id, json!({}))
+    let err = flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
         .await
         .expect_err("a run whose node errors under on_error:stop must fail");
     assert!(!err.is_empty());
@@ -452,9 +462,14 @@ async fn flows_resume_continues_a_paused_run_to_completion() {
         .await
         .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({ "x": 1 }))
-        .await
-        .unwrap();
+    let run = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "x": 1 }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
     let pending: Vec<String> =
         serde_json::from_value(run.value["pending_approvals"].clone()).unwrap();
@@ -516,9 +531,14 @@ async fn flows_resume_with_empty_approvals_is_rejected_and_does_not_complete_the
         .await
         .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({ "x": 1 }))
-        .await
-        .unwrap();
+    let run = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "x": 1 }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
 
     let err = flows_resume(&config, &created.value.id, &thread_id, vec![])
@@ -550,9 +570,14 @@ async fn flows_resume_with_mismatched_approvals_is_rejected() {
         .await
         .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({ "x": 1 }))
-        .await
-        .unwrap();
+    let run = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "x": 1 }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
 
     // Names a node id that is not actually pending for this run.
@@ -575,9 +600,14 @@ async fn flows_resume_with_the_correct_gate_completes_and_runs_downstream() {
         .await
         .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({ "x": 1 }))
-        .await
-        .unwrap();
+    let run = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "x": 1 }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
 
     let resumed = flows_resume(
@@ -608,7 +638,7 @@ async fn flows_resume_of_a_non_paused_run_errors_clearly() {
 
     // This run completes outright (no approval gate) — its recorded status
     // is "completed", not "pending_approval".
-    let run = flows_run(&config, &created.value.id, json!({}))
+    let run = flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
         .await
         .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
@@ -651,9 +681,14 @@ async fn flows_run_persists_a_flow_run_row_queryable_via_list_and_get() {
         .await
         .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({ "hello": "world" }))
-        .await
-        .unwrap();
+    let run = flows_run(
+        &config,
+        &created.value.id,
+        json!({ "hello": "world" }),
+        FlowRunTrigger::Rpc,
+    )
+    .await
+    .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
 
     let runs = flows_list_runs(&config, &created.value.id, 20)
@@ -699,7 +734,7 @@ async fn flows_run_emits_pending_approval_notification() {
     .await
     .unwrap();
 
-    let run = flows_run(&config, &created.value.id, json!({}))
+    let run = flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
         .await
         .unwrap();
     let thread_id = run.value["thread_id"].as_str().unwrap().to_string();
@@ -752,7 +787,7 @@ async fn flows_run_does_not_notify_when_run_completes_without_pending_approvals(
         .unwrap();
     let created_id = created.value.id.clone();
 
-    flows_run(&config, &created.value.id, json!({}))
+    flows_run(&config, &created.value.id, json!({}), FlowRunTrigger::Rpc)
         .await
         .unwrap();
 

src/openhuman/flows/schemas.rs

@@ -413,7 +413,15 @@ fn handle_run(params: Map<String, Value>) -> ControllerFuture {
         let config = config_rpc::load_config_with_timeout().await?;
         let id = read_required::<String>(&params, "id")?;
         let input = params.get("input").cloned().unwrap_or(Value::Null);
-        to_json(ops::flows_run(&config, id.trim(), input).await?)
+        to_json(
+            ops::flows_run(
+                &config,
+                id.trim(),
+                input,
+                crate::openhuman::flows::FlowRunTrigger::Rpc,
+            )
+            .await?,
+        )
     })
 }
 

src/openhuman/flows/types.rs

@@ -8,6 +8,33 @@
 use serde::{Deserialize, Serialize};
 use tinyflows::model::WorkflowGraph;
 
+/// How a flow run was started. Stamped onto the run's Langfuse trace as a
+/// `trigger:<kind>` tag plus `trigger` metadata so runs can be filtered by
+/// origin in the Langfuse UI.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FlowRunTrigger {
+    /// An explicit run request over RPC/CLI (the Workflows UI "Run" button).
+    Rpc,
+    /// A `FlowScheduleTick` cron dispatch (`schedule` trigger node).
+    Schedule,
+    /// A `ComposioTriggerReceived` dispatch (`app_event` trigger node).
+    AppEvent,
+    /// A human-in-the-loop resume of a paused run (`flows_resume`).
+    Resume,
+}
+
+impl FlowRunTrigger {
+    /// Stable snake_case identifier used in Langfuse tags/metadata.
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            FlowRunTrigger::Rpc => "rpc",
+            FlowRunTrigger::Schedule => "schedule",
+            FlowRunTrigger::AppEvent => "app_event",
+            FlowRunTrigger::Resume => "resume",
+        }
+    }
+}
+
 /// A saved automation workflow: a `tinyflows` graph plus OpenHuman-side
 /// bookkeeping (enablement, run history summary).
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/openhuman/tinyagents/observability.rs

@@ -631,7 +631,9 @@ impl EventListener for OpenhumanEventBridge {
                     }),
                 }
             }
-            AgentEvent::ToolCompleted { call_id, tool_name } => {
+            AgentEvent::ToolCompleted {
+                call_id, tool_name, ..
+            } => {
                 let iteration = self.iteration();
                 // The crate event carries no success/error, so read what the
                 // outcome-capture middleware classified for this call. Absent →
@@ -775,6 +777,8 @@ mod tests {
         sink.emit(AgentEvent::ToolCompleted {
             call_id: "c1".into(),
             tool_name: "echo".to_string(),
+            input: None,
+            output: None,
         });
         sink.emit(AgentEvent::UsageRecorded {
             usage: Usage::new(100, 40),

src/openhuman/tinyflows/langfuse_export.rs

@@ -0,0 +1,419 @@
+//! Langfuse export for `flows::` graph runs.
+//!
+//! After a `flows_run` / `flows_resume` settles, the `flows::` domain hands
+//! this module the run's durable [`GraphObservation`] slice (captured by the
+//! per-run in-memory journal that `tinyflows`' journaled entry points fill)
+//! and it exports the run as one Langfuse trace via the backend's Langfuse
+//! proxy route, `/telemetry/langfuse/ingestion`.
+//!
+//! The batch is built by `tinyagents`' [`GraphLangfuseExporter`], which turns
+//! each superstep and node into a timed span and stamps the Langfuse **Agent
+//! Graph view** keys (`langgraph_node` / `langgraph_step`) on node spans, so
+//! the Langfuse UI can render the flow run as a graph. A host span-metadata
+//! injector additionally stamps `flow_id` on every span for filtering.
+//!
+//! Transport mirrors the agent-turn tracing path
+//! (`agent::progress_tracing::langfuse::push_spans`): the endpoint is derived
+//! from the **current backend hostname** (`effective_backend_api_url`), auth
+//! is the live OpenHuman session bearer (the backend injects the real
+//! Langfuse keys server-side), the send is capped at 10s, `207 Multi-Status`
+//! is tolerated, and every failure is logged and swallowed — exporting is
+//! best-effort and never fails the run. Gated on
+//! `observability.share_usage_data`.
+
+use std::time::Duration;
+
+use serde_json::{json, Map};
+use tinyagents::{
+    GraphLangfuseExporter, GraphObservation, LangfuseAuth, LangfuseClient, LangfuseTraceConfig,
+};
+
+use crate::api::config::effective_backend_api_url;
+use crate::openhuman::config::Config;
+use crate::openhuman::credentials::session_support::require_live_session_token;
+use crate::openhuman::flows::FlowRunTrigger;
+
+const LOG_TARGET: &str = "flows::langfuse";
+/// Backend proxy route for Langfuse ingestion (relative to the backend
+/// origin). The backend authenticates the session JWT, injects the Langfuse
+/// project keys, and forwards to Langfuse's real `/api/public/ingestion`.
+const INGESTION_PATH: &str = "/telemetry/langfuse/ingestion";
+/// Cap the push so a slow/hung Langfuse never stalls run teardown (same
+/// posture as the agent-turn exporter).
+const PUSH_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Resolve the Langfuse ingestion URL from the current backend host — the
+/// exact base-server resolution every other backend call uses, so the host
+/// always matches wherever the app's domain calls go (staging, prod, or a
+/// custom `api_url` override).
+fn ingestion_url(config: &Config) -> String {
+    let base = effective_backend_api_url(&config.api_url);
+    crate::api::config::api_url(&base, INGESTION_PATH)
+}
+
+/// The OpenHuman core crate version (e.g. `0.58.0`), stamped onto every flow
+/// trace as the Langfuse `release` field plus an `app_version` metadata key so
+/// traces can be correlated with the app build that produced them.
+const APP_VERSION: &str = env!("CARGO_PKG_VERSION");
+
+/// Builds the [`LangfuseTraceConfig`] for one flow run: the trace id **and**
+/// session id are the run's `thread_id` (`flow:{flow_id}:{uuid}`), the trace
+/// is named `flow.run:{flow_name}`, run-type tags (`run:flow` +
+/// `trigger:<kind>`) mark how the run started, and flow coordinates plus the
+/// app version ride on the trace metadata. No content — ids, name, status,
+/// trigger, and version only.
+fn build_flow_trace_config(
+    flow_name: &str,
+    flow_id: &str,
+    thread_id: &str,
+    status: &str,
+    trigger: FlowRunTrigger,
+) -> LangfuseTraceConfig {
+    LangfuseTraceConfig {
+        trace_id: Some(thread_id.to_string()),
+        name: Some(format!("flow.run:{flow_name}")),
+        session_id: Some(thread_id.to_string()),
+        release: Some(APP_VERSION.to_string()),
+        tags: vec![
+            "run:flow".to_string(),
+            format!("trigger:{}", trigger.as_str()),
+        ],
+        metadata: json!({
+            "flow_id": flow_id,
+            "status": status,
+            "source": "flows",
+            "run_type": "flow",
+            "trigger": trigger.as_str(),
+            "app_version": APP_VERSION,
+        }),
+        ..Default::default()
+    }
+}
+
+/// Builds the graph exporter for a flow run: the batch builder plus a host
+/// span-metadata injector that stamps `flow_id` on every span (node spans
+/// already carry `langgraph_node`/`langgraph_step` from `tinyagents`).
+fn build_flow_exporter(client: LangfuseClient, flow_id: &str) -> GraphLangfuseExporter {
+    let flow_id = flow_id.to_string();
+    GraphLangfuseExporter::new(client).with_span_metadata_fn(move |_obs| {
+        let mut extra = Map::new();
+        extra.insert("flow_id".to_string(), json!(flow_id));
+        Some(extra)
+    })
+}
+
+/// Exports one settled flow run to Langfuse as a single trace. Best-effort:
+/// every failure path logs a `[flows]`-prefixed warning and returns — a
+/// Langfuse outage can never fail or delay-fail the run. No-op when
+/// `observability.share_usage_data` is off or there is nothing to send.
+pub async fn export_flow_run_trace(
+    config: &Config,
+    flow_name: &str,
+    flow_id: &str,
+    thread_id: &str,
+    status: &str,
+    trigger: FlowRunTrigger,
+    journal_observations: &[GraphObservation],
+) {
+    if !config.observability.share_usage_data {
+        tracing::debug!(
+            target: LOG_TARGET,
+            flow_id = %flow_id,
+            "[flows] langfuse export skipped: observability.share_usage_data is off"
+        );
+        return;
+    }
+    if journal_observations.is_empty() {
+        tracing::debug!(
+            target: LOG_TARGET,
+            flow_id = %flow_id,
+            thread_id = %thread_id,
+            "[flows] langfuse export skipped: run journal is empty"
+        );
+        return;
+    }
+
+    let url = ingestion_url(config);
+    if !url.starts_with("http") {
+        tracing::warn!(
+            target: LOG_TARGET,
+            flow_id = %flow_id,
+            "[flows] langfuse export skipped: could not resolve ingestion URL from backend host (got {url:?})"
+        );
+        return;
+    }
+    let token = match require_live_session_token(config) {
+        Ok(token) => token,
+        Err(err) => {
+            tracing::warn!(
+                target: LOG_TARGET,
+                flow_id = %flow_id,
+                error = %err,
+                "[flows] langfuse export skipped: no live session token"
+            );
+            return;
+        }
+    };
+    let client = match LangfuseClient::new(url.clone(), LangfuseAuth::Bearer { token }) {
+        Ok(client) => client,
+        Err(err) => {
+            tracing::warn!(
+                target: LOG_TARGET,
+                flow_id = %flow_id,
+                error = %err,
+                "[flows] langfuse export skipped: could not build client for {url}"
+            );
+            return;
+        }
+    };
+
+    let exporter = build_flow_exporter(client, flow_id);
+    let trace = build_flow_trace_config(flow_name, flow_id, thread_id, status, trigger);
+    let observation_count = journal_observations.len();
+    tracing::debug!(
+        target: LOG_TARGET,
+        flow_id = %flow_id,
+        thread_id = %thread_id,
+        status = %status,
+        trigger = %trigger.as_str(),
+        observation_count,
+        endpoint = %exporter.endpoint(),
+        "[flows] pushing flow run trace to Langfuse"
+    );
+
+    // `send_observations` already tolerates 207 Multi-Status; the outer
+    // timeout caps a hung connection the same way the agent exporter does.
+    match tokio::time::timeout(
+        PUSH_TIMEOUT,
+        exporter.send_observations(trace, journal_observations),
+    )
+    .await
+    {
+        Ok(Ok(_)) => {
+            tracing::debug!(
+                target: LOG_TARGET,
+                flow_id = %flow_id,
+                thread_id = %thread_id,
+                observation_count,
+                "[flows] pushed flow run trace to Langfuse"
+            );
+        }
+        Ok(Err(err)) => {
+            tracing::warn!(
+                target: LOG_TARGET,
+                flow_id = %flow_id,
+                thread_id = %thread_id,
+                error = %err,
+                "[flows] langfuse export failed (run unaffected)"
+            );
+        }
+        Err(_elapsed) => {
+            tracing::warn!(
+                target: LOG_TARGET,
+                flow_id = %flow_id,
+                thread_id = %thread_id,
+                timeout_secs = PUSH_TIMEOUT.as_secs(),
+                "[flows] langfuse export timed out (run unaffected)"
+            );
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::Value;
+    use tinyagents::harness::ids;
+    use tinyagents::GraphEvent;
+
+    /// Builds a minimal observation stream for one node under run/thread ids
+    /// shaped like a real `flows_run` (`thread_id = flow:{id}:{uuid}`).
+    fn sample_observations(thread_id: &str) -> Vec<GraphObservation> {
+        let node = ids::NodeId::new("fetch");
+        let mk = |offset: u64, step: usize, ts_ms: u64, event: GraphEvent| GraphObservation {
+            event_id: ids::EventId::new(format!("evt-{offset}")),
+            run_id: ids::RunId::new("run-9"),
+            root_run_id: ids::RunId::new("run-9"),
+            parent_run_id: None,
+            thread_id: Some(ids::ThreadId::new(thread_id)),
+            graph_id: ids::GraphId::new("workflow"),
+            checkpoint_id: None,
+            namespace: Vec::new(),
+            step,
+            offset,
+            ts_ms,
+            event,
+        };
+        vec![
+            mk(
+                0,
+                0,
+                1_000,
+                GraphEvent::RunStarted {
+                    run_id: ids::RunId::new("run-9"),
+                },
+            ),
+            mk(
+                1,
+                1,
+                1_010,
+                GraphEvent::StepStarted {
+                    step: 1,
+                    active: vec![node.clone()],
+                },
+            ),
+            mk(
+                2,
+                1,
+                1_020,
+                GraphEvent::NodeStarted {
+                    node: node.clone(),
+                    step: 1,
+                },
+            ),
+            mk(
+                3,
+                1,
+                1_050,
+                GraphEvent::NodeCompleted {
+                    node: node.clone(),
+                    step: 1,
+                },
+            ),
+            mk(4, 1, 1_060, GraphEvent::StepCompleted { step: 1 }),
+        ]
+    }
+
+    /// Finds the first span-create whose body id matches.
+    fn find_span<'a>(batch: &'a [Value], id: &str) -> Option<&'a Value> {
+        batch
+            .iter()
+            .find(|e| e["type"] == "span-create" && e["body"]["id"] == id)
+    }
+
+    #[test]
+    fn ingestion_url_targets_backend_proxy_route() {
+        let mut config = Config::default();
+        config.api_url = Some("https://staging-api.tinyhumans.ai/api/v1".to_string());
+        assert_eq!(
+            ingestion_url(&config),
+            "https://staging-api.tinyhumans.ai/telemetry/langfuse/ingestion"
+        );
+    }
+
+    #[test]
+    fn flow_trace_config_uses_thread_id_and_flow_coordinates() {
+        let trace = build_flow_trace_config(
+            "Daily digest",
+            "flow-1",
+            "flow:flow-1:uuid-1",
+            "completed",
+            FlowRunTrigger::Schedule,
+        );
+        assert_eq!(trace.trace_id.as_deref(), Some("flow:flow-1:uuid-1"));
+        assert_eq!(trace.session_id.as_deref(), Some("flow:flow-1:uuid-1"));
+        assert_eq!(trace.name.as_deref(), Some("flow.run:Daily digest"));
+        assert_eq!(trace.tags, vec!["run:flow", "trigger:schedule"]);
+        assert_eq!(trace.metadata["flow_id"], "flow-1");
+        assert_eq!(trace.metadata["status"], "completed");
+        assert_eq!(trace.metadata["source"], "flows");
+        assert_eq!(trace.metadata["run_type"], "flow");
+        assert_eq!(trace.metadata["trigger"], "schedule");
+        assert_eq!(trace.release.as_deref(), Some(APP_VERSION));
+        assert_eq!(trace.metadata["app_version"], APP_VERSION);
+        assert!(!APP_VERSION.is_empty(), "crate version must be baked in");
+    }
+
+    #[test]
+    fn batch_carries_flow_trace_and_langgraph_keys_on_node_spans() {
+        let thread_id = "flow:flow-1:uuid-1";
+        let observations = sample_observations(thread_id);
+        let client = LangfuseClient::new(
+            "https://backend.test/telemetry/langfuse/ingestion",
+            LangfuseAuth::Bearer {
+                token: "tok".to_string(),
+            },
+        )
+        .expect("client");
+        let exporter = build_flow_exporter(client, "flow-1");
+        let trace = build_flow_trace_config(
+            "Daily digest",
+            "flow-1",
+            thread_id,
+            "completed",
+            FlowRunTrigger::Rpc,
+        );
+        let payload = exporter
+            .build_ingestion_batch(trace, &observations)
+            .expect("batch");
+        let batch = payload["batch"].as_array().expect("batch array");
+
+        // Trace: id + sessionId are the run thread id; name and flow
+        // coordinates as configured.
+        let trace_event = &batch[0];
+        assert_eq!(trace_event["type"], "trace-create");
+        assert_eq!(trace_event["body"]["id"], thread_id);
+        assert_eq!(trace_event["body"]["sessionId"], thread_id);
+        assert_eq!(trace_event["body"]["name"], "flow.run:Daily digest");
+        assert_eq!(trace_event["body"]["metadata"]["flow_id"], "flow-1");
+        assert_eq!(trace_event["body"]["metadata"]["status"], "completed");
+        assert_eq!(trace_event["body"]["metadata"]["source"], "flows");
+        assert_eq!(trace_event["body"]["metadata"]["run_type"], "flow");
+        assert_eq!(trace_event["body"]["metadata"]["trigger"], "rpc");
+        assert_eq!(
+            trace_event["body"]["tags"],
+            json!(["run:flow", "trigger:rpc"]),
+            "run-type tags must ride on the trace-create"
+        );
+        assert_eq!(
+            trace_event["body"]["release"], APP_VERSION,
+            "app version must ride on the trace-create as release"
+        );
+        assert_eq!(trace_event["body"]["metadata"]["app_version"], APP_VERSION);
+
+        // Node span: Agent-Graph-view keys + the injected flow_id, under the
+        // overridden trace id.
+        let node = find_span(batch, &format!("{thread_id}:node:fetch:1")).expect("node span");
+        assert_eq!(node["body"]["traceId"], thread_id);
+        assert_eq!(node["body"]["metadata"]["langgraph_node"], "fetch");
+        assert_eq!(node["body"]["metadata"]["langgraph_step"], 1);
+        assert_eq!(node["body"]["metadata"]["flow_id"], "flow-1");
+
+        // Step span: superstep index + injected flow_id.
+        let step = find_span(batch, &format!("{thread_id}:step:1")).expect("step span");
+        assert_eq!(step["body"]["metadata"]["langgraph_step"], 1);
+        assert_eq!(step["body"]["metadata"]["flow_id"], "flow-1");
+    }
+
+    #[tokio::test]
+    async fn export_is_a_noop_when_share_usage_data_is_off() {
+        let mut config = Config::default();
+        config.observability.share_usage_data = false;
+        // Must return without any host/token resolution or network.
+        export_flow_run_trace(
+            &config,
+            "Daily digest",
+            "flow-1",
+            "flow:flow-1:uuid-1",
+            "completed",
+            FlowRunTrigger::Rpc,
+            &sample_observations("flow:flow-1:uuid-1"),
+        )
+        .await;
+    }
+
+    #[tokio::test]
+    async fn export_with_empty_observations_is_a_noop() {
+        let config = Config::default();
+        export_flow_run_trace(
+            &config,
+            "Daily digest",
+            "flow-1",
+            "flow:flow-1:uuid-1",
+            "completed",
+            FlowRunTrigger::AppEvent,
+            &[],
+        )
+        .await;
+    }
+}

src/openhuman/tinyflows/mod.rs

@@ -5,10 +5,14 @@
 //! This module is export-focused. The five capability adapters plus the two
 //! run entry points — [`build_capabilities`] and [`open_flow_checkpointer`],
 //! re-exported below — live in [`caps`]; run observability logging lives in
-//! [`observability`]. The `flows::` domain (`src/openhuman/flows/ops.rs`) calls
-//! [`build_capabilities`] / [`open_flow_checkpointer`] to drive a run.
+//! [`observability`]; post-run Langfuse export of a run's durable graph
+//! observations lives in [`langfuse_export`]. The `flows::` domain
+//! (`src/openhuman/flows/ops.rs`) calls [`build_capabilities`] /
+//! [`open_flow_checkpointer`] to drive a run and
+//! [`langfuse_export::export_flow_run_trace`] after it settles.
 
 pub mod caps;
+pub mod langfuse_export;
 pub mod observability;
 #[cfg(test)]
 mod tests;