diff --git a/docs/TEST-COVERAGE-MATRIX.md b/docs/TEST-COVERAGE-MATRIX.md index 761bb5a730..30b5422cdd 100644 --- a/docs/TEST-COVERAGE-MATRIX.md +++ b/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 diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs index a4d41cbc64..ec03b60a42 100644 --- a/src/openhuman/flows/bus.rs +++ b/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") } diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs index c11253ffac..dbeebe87d9 100644 --- a/src/openhuman/flows/mod.rs +++ b/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}; diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs index f4f577f08c..ecd4c9cfe2 100644 --- a/src/openhuman/flows/ops.rs +++ b/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, 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!( diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs index d782d49486..bd5ed76f73 100644 --- a/src/openhuman/flows/ops_tests.rs +++ b/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 = 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(); diff --git a/src/openhuman/flows/schemas.rs b/src/openhuman/flows/schemas.rs index 50c4212c62..21e0788fa8 100644 --- a/src/openhuman/flows/schemas.rs +++ b/src/openhuman/flows/schemas.rs @@ -413,7 +413,15 @@ fn handle_run(params: Map) -> ControllerFuture { let config = config_rpc::load_config_with_timeout().await?; let id = read_required::(¶ms, "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?, + ) }) } diff --git a/src/openhuman/flows/types.rs b/src/openhuman/flows/types.rs index 9251dcd2e7..c0047151f9 100644 --- a/src/openhuman/flows/types.rs +++ b/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:` 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)] diff --git a/src/openhuman/tinyagents/observability.rs b/src/openhuman/tinyagents/observability.rs index 04393b87a8..af700a7058 100644 --- a/src/openhuman/tinyagents/observability.rs +++ b/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), diff --git a/src/openhuman/tinyflows/langfuse_export.rs b/src/openhuman/tinyflows/langfuse_export.rs new file mode 100644 index 0000000000..6870d5bfd7 --- /dev/null +++ b/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:`) 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 { + 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; + } +} diff --git a/src/openhuman/tinyflows/mod.rs b/src/openhuman/tinyflows/mod.rs index 8d3b76bd54..3276521270 100644 --- a/src/openhuman/tinyflows/mod.rs +++ b/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; diff --git a/vendor/tinyagents b/vendor/tinyagents index ac7338241d..2f023e8abf 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit ac7338241d50f2ecbfe52e518626b31d52c16ce6 +Subproject commit 2f023e8abf88781d934272c989b7ed7cb28c3615 diff --git a/vendor/tinyflows b/vendor/tinyflows index 2dcddfd111..438f8fce2d 160000 --- a/vendor/tinyflows +++ b/vendor/tinyflows @@ -1 +1 @@ -Subproject commit 2dcddfd111709fbdfdfcc118d88bbc6b8ea1b496 +Subproject commit 438f8fce2dc54cfd98e57d9a2fe93d5e07aad27c