From 8aab529deff46ac818b7f57f597d432d016ca33c Mon Sep 17 00:00:00 2001 From: Cyrus Gray <144336577+graycyrus@users.noreply.github.com> Date: Fri, 3 Jul 2026 19:31:43 +0530 Subject: [PATCH] =?UTF-8?q?feat(flows):=20tinyflows=20Workflows=20(B1)=20?= =?UTF-8?q?=E2=80=94=20engine=20seam=20+=20flows=20domain=20(#4433)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 138 ++++++ Cargo.toml | 5 + src/core/all.rs | 4 + src/openhuman/agent_registry/agents/loader.rs | 4 +- src/openhuman/channels/runtime/startup.rs | 5 + src/openhuman/flows/bus.rs | 81 +++ src/openhuman/flows/mod.rs | 26 + src/openhuman/flows/ops.rs | 213 ++++++++ src/openhuman/flows/ops_tests.rs | 248 +++++++++ src/openhuman/flows/schemas.rs | 405 +++++++++++++++ src/openhuman/flows/store.rs | 301 +++++++++++ src/openhuman/flows/store_tests.rs | 169 +++++++ src/openhuman/flows/types.rs | 71 +++ src/openhuman/mcp_server/resources.rs | 12 + src/openhuman/mod.rs | 2 + src/openhuman/orchestration/graph/mod.rs | 13 +- src/openhuman/orchestration/ops.rs | 10 +- src/openhuman/tinyflows/caps.rs | 469 ++++++++++++++++++ src/openhuman/tinyflows/mod.rs | 16 + src/openhuman/tinyflows/observability.rs | 70 +++ src/openhuman/tinyflows/tests.rs | 214 ++++++++ 21 files changed, 2470 insertions(+), 6 deletions(-) create mode 100644 src/openhuman/flows/bus.rs create mode 100644 src/openhuman/flows/mod.rs create mode 100644 src/openhuman/flows/ops.rs create mode 100644 src/openhuman/flows/ops_tests.rs create mode 100644 src/openhuman/flows/schemas.rs create mode 100644 src/openhuman/flows/store.rs create mode 100644 src/openhuman/flows/store_tests.rs create mode 100644 src/openhuman/flows/types.rs create mode 100644 src/openhuman/tinyflows/caps.rs create mode 100644 src/openhuman/tinyflows/mod.rs create mode 100644 src/openhuman/tinyflows/observability.rs create mode 100644 src/openhuman/tinyflows/tests.rs diff --git a/Cargo.lock b/Cargo.lock index d030ea443..c0b7809f8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -657,6 +657,17 @@ dependencies = [ "tinyvec", ] +[[package]] +name = "bstr" +version = "1.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5cee35f73844aa3014bb606320a6c1f010249dbdf43342fe54b5a4f6a8ed4b79" +dependencies = [ + "memchr", + "regex-automata", + "serde_core", +] + [[package]] name = "bumpalo" version = "3.20.2" @@ -2345,6 +2356,12 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" +[[package]] +name = "futures-timer" +version = "3.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968" + [[package]] name = "futures-util" version = "0.3.32" @@ -2604,6 +2621,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3011d1213f159867b13cfd6ac92d2cd5f1345762c63be3554e84092d85a50bbd" +[[package]] +name = "hifijson" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "242402749acf71e6f32f5857598b7002c4058a4e3c3b22b4c7d51cab9aea754e" + [[package]] name = "hkdf" version = "0.12.4" @@ -3122,6 +3145,54 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jaq-core" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7561783b20275a6c9cb576e39208b0c635f34ef14357f1f05a2927a774f3adec" +dependencies = [ + "dyn-clone", + "once_cell", + "typed-arena", +] + +[[package]] +name = "jaq-json" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4ec9aaad7340e6990c6c1878ef3b46dbec624e535d7f786cc9ddcf94f773d33" +dependencies = [ + "bstr", + "bytes", + "foldhash 0.1.5", + "hifijson", + "indexmap", + "jaq-core", + "jaq-std", + "num-bigint", + "num-traits", + "ryu", + "self_cell", + "serde_core", +] + +[[package]] +name = "jaq-std" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bdc5a74b0feeb5e6a1dc2dd08c34280a61e37668d10a6a3b27ad69d0fb9ce2e" +dependencies = [ + "aho-corasick", + "base64 0.22.1", + "bstr", + "jaq-core", + "jiff", + "libm", + "log", + "regex-bites", + "urlencoding", +] + [[package]] name = "jiff" version = "0.2.24" @@ -3129,10 +3200,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f00b5dbd620d61dfdcb6007c9c1f6054ebd75319f163d886a9055cec1155073d" dependencies = [ "jiff-static", + "jiff-tzdb-platform", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "windows-sys 0.61.2", ] [[package]] @@ -3146,6 +3219,21 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "jiff-tzdb" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6142247df1a93c2b3587402a19710be3e6e942f1581a1702e76408f2c21d6590" + +[[package]] +name = "jiff-tzdb-platform" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8" +dependencies = [ + "jiff-tzdb", +] + [[package]] name = "jni" version = "0.21.1" @@ -3760,6 +3848,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-bigint" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c863e9ab5e7bf9c99ba75e1050f1e4d624ae87ed3532d6238ffbdc7b585dbbe6" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-conv" version = "0.2.1" @@ -3777,6 +3875,15 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "num-integer" +version = "0.1.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f" +dependencies = [ + "num-traits", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -4285,6 +4392,7 @@ dependencies = [ "tempfile", "thiserror 2.0.18", "tinyagents", + "tinyflows", "tinyplace", "tokio", "tokio-rustls", @@ -5306,6 +5414,12 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "regex-bites" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6a15a2fa0bfda9361941c45550896ae87b15cc6c8c939ea350079670332e211" + [[package]] name = "regex-syntax" version = "0.8.10" @@ -6577,6 +6691,24 @@ dependencies = [ "tokio", ] +[[package]] +name = "tinyflows" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11d5c7fbe95186c2a2b29cbfbcdde6c5241977261a55fc50bbdce203c1b568d2" +dependencies = [ + "async-trait", + "futures-timer", + "jaq-core", + "jaq-json", + "jaq-std", + "serde", + "serde_json", + "thiserror 2.0.18", + "tinyagents", + "tracing", +] + [[package]] name = "tinyplace" version = "1.0.1" @@ -7066,6 +7198,12 @@ dependencies = [ "pom", ] +[[package]] +name = "typed-arena" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6af6ae20167a9ece4bcb41af5b80f8a1f1df981f6391189ce00fd257af04126a" + [[package]] name = "typed-builder" version = "0.23.2" diff --git a/Cargo.toml b/Cargo.toml index f90daaa82..cbd25ca86 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -40,6 +40,11 @@ crate-type = ["rlib"] [dependencies] # tiny.place A2A social network SDK — published on crates.io (tinyhumansai/tiny.place) tinyplace = "1.0.1" +# tinyflows — host-agnostic workflow engine (typed node graph → validate → compile → +# run on tinyagents). Powers the "Workflows" feature via the seam in +# `src/openhuman/tinyflows/` + the `flows::` domain. Pulls tinyagents 1.3 transitively +# (same version openhuman already uses — no conflict). Published on crates.io. +tinyflows = "0.2" # TinyAgents — Rust LLM orchestration framework (LangGraph/LangChain-style): # durable state graphs, agent-loop harness, model/tool registries, REPL + # `.rag` workflow language. openhuman's agent engine + orchestration run on this diff --git a/src/core/all.rs b/src/core/all.rs index 6f120e4b6..ab5eece79 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -119,6 +119,8 @@ fn build_registered_controllers() -> Vec { .extend(crate::openhuman::recall_calendar::all_recall_calendar_registered_controllers()); // Scheduled job management controllers.extend(crate::openhuman::cron::all_cron_registered_controllers()); + // Saved automation workflows (tinyflows graphs): create/get/list/update/delete/run + controllers.extend(crate::openhuman::flows::all_flows_registered_controllers()); // Proactive task ingestion from external tools (github/notion/linear/clickup) controllers.extend(crate::openhuman::task_sources::all_task_sources_registered_controllers()); controllers.extend(crate::openhuman::dashboard::all_dashboard_registered_controllers()); @@ -375,6 +377,7 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::composio::all_composio_controller_schemas()); schemas.extend(crate::openhuman::recall_calendar::all_recall_calendar_controller_schemas()); schemas.extend(crate::openhuman::cron::all_cron_controller_schemas()); + schemas.extend(crate::openhuman::flows::all_flows_controller_schemas()); schemas.extend(crate::openhuman::task_sources::all_task_sources_controller_schemas()); schemas.extend(crate::openhuman::dashboard::all_dashboard_controller_schemas()); schemas.extend(crate::openhuman::mcp_registry::all_mcp_registry_controller_schemas()); @@ -536,6 +539,7 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "Connectivity diagnostics for the local sidecar, listening port, and backend Socket.IO state.", ), "cron" => Some("Manage scheduled jobs and run history."), + "flows" => Some("Create, store, and run automation workflows."), "dashboard" => Some( "Operator-facing dashboard aggregations: per-model health comparison rows.", ), diff --git a/src/openhuman/agent_registry/agents/loader.rs b/src/openhuman/agent_registry/agents/loader.rs index c2eb06dc7..fa1f30a34 100644 --- a/src/openhuman/agent_registry/agents/loader.rs +++ b/src/openhuman/agent_registry/agents/loader.rs @@ -291,13 +291,13 @@ pub const BUILTINS: &[BuiltinAgent] = &[ id: "frontend_agent", toml: include_str!("../../orchestration/frontend_agent/agent.toml"), prompt_fn: crate::openhuman::orchestration::frontend_agent::prompt::build, - graph_fn: crate::openhuman::orchestration::frontend_agent::graph::graph, + graph_fn: Some(crate::openhuman::orchestration::frontend_agent::graph::graph), }, BuiltinAgent { id: "reasoning_agent", toml: include_str!("../../orchestration/reasoning_agent/agent.toml"), prompt_fn: crate::openhuman::orchestration::reasoning_agent::prompt::build, - graph_fn: crate::openhuman::orchestration::reasoning_agent::graph::graph, + graph_fn: Some(crate::openhuman::orchestration::reasoning_agent::graph::graph), }, ]; diff --git a/src/openhuman/channels/runtime/startup.rs b/src/openhuman/channels/runtime/startup.rs index d29b290b6..accd31783 100644 --- a/src/openhuman/channels/runtime/startup.rs +++ b/src/openhuman/channels/runtime/startup.rs @@ -702,6 +702,11 @@ pub async fn start_channels(mut config: Config) -> Result<()> { let _cron_delivery_handle = bus.subscribe(Arc::new( crate::openhuman::cron::bus::CronDeliverySubscriber::new(Arc::clone(&channels_by_name)), )); + // Register the flows trigger subscriber (B1: observes matched events and + // logs them; B2 maps them onto enabled flows and calls `flows_run`). + let _flows_trigger_handle = bus.subscribe(Arc::new( + crate::openhuman::flows::bus::FlowTriggerSubscriber::new(), + )); // Register the proactive message subscriber so morning briefings, // welcome messages, and other proactive agent output gets routed to // the user's active channel (+ always to web). diff --git a/src/openhuman/flows/bus.rs b/src/openhuman/flows/bus.rs new file mode 100644 index 000000000..1b0c9032a --- /dev/null +++ b/src/openhuman/flows/bus.rs @@ -0,0 +1,81 @@ +//! Event bus handlers for the `flows::` domain. +//! +//! B1 scope: this subscriber only **observes** and logs events that will later +//! drive automatic flow runs (B2). It does not yet map a matched event onto an +//! enabled [`crate::openhuman::flows::Flow`] or call `flows_run` — that bridge, +//! plus `TrustedAutomationSource::Workflow` for externally-triggered runs, is +//! B2+ (see `my_docs/ohxtf/b1-engine-seam-domain/07-execution-and-hitl.md`). + +use crate::core::event_bus::{DomainEvent, EventHandler}; +use async_trait::async_trait; + +/// Listens for events that a saved flow's trigger node might match +/// (`cron`, `composio`, `system` domains) and logs them. +/// +/// A future revision resolves matched events to enabled flows and invokes +/// `flows::ops::flows_run` for each match. +pub struct FlowTriggerSubscriber; + +impl FlowTriggerSubscriber { + pub fn new() -> Self { + Self + } +} + +impl Default for FlowTriggerSubscriber { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl EventHandler for FlowTriggerSubscriber { + fn name(&self) -> &str { + "flows::trigger" + } + + fn domains(&self) -> Option<&[&str]> { + Some(&["cron", "composio", "system"]) + } + + async fn handle(&self, event: &DomainEvent) { + // B1: log-only. Once trigger binding lands (B2), this maps the event + // onto enabled flows whose trigger node kind/config matches it, and + // runs each match through `flows::ops::flows_run`. + tracing::debug!( + target: "flows", + ?event, + "[flows] trigger subscriber observed event (B1: not yet dispatched to flows)" + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn name_and_domains_are_stable() { + let sub = FlowTriggerSubscriber::new(); + assert_eq!(sub.name(), "flows::trigger"); + assert_eq!(sub.domains(), Some(&["cron", "composio", "system"][..])); + } + + #[tokio::test] + async fn handle_does_not_panic_on_arbitrary_events() { + let sub = FlowTriggerSubscriber; + sub.handle(&DomainEvent::CronJobTriggered { + job_id: "j1".into(), + job_name: "test".into(), + job_type: "shell".into(), + }) + .await; + } + + #[test] + fn default_constructs_the_same_as_new() { + let a = FlowTriggerSubscriber::new(); + let b = FlowTriggerSubscriber::default(); + assert_eq!(a.name(), b.name()); + } +} diff --git a/src/openhuman/flows/mod.rs b/src/openhuman/flows/mod.rs new file mode 100644 index 000000000..f1973add6 --- /dev/null +++ b/src/openhuman/flows/mod.rs @@ -0,0 +1,26 @@ +//! The `flows::` domain: saved automation workflows (tinyflows graphs) — +//! create/get/list/update/delete/enable/run, backed by SQLite. Mirrors +//! `src/openhuman/cron/`'s module shape. +//! +//! Business logic lives in [`ops`]; persistence in `store` (private, with a +//! handful of functions re-exported below for the capability seam's +//! [`crate::openhuman::tinyflows::caps::FlowStateStore`]); the RPC/CLI +//! controller surface in `schemas` (private, re-exported below). + +pub mod bus; +pub mod ops; +mod schemas; +mod store; +mod types; + +pub use schemas::{ + all_controller_schemas as all_flows_controller_schemas, + all_registered_controllers as all_flows_registered_controllers, +}; +// `kv_get`/`kv_set` are re-exported (not just `pub(crate)`-visible within this +// domain's own module tree) because `tinyflows::caps::FlowStateStore` +// (`src/openhuman/tinyflows/caps.rs`) lives in a sibling domain and needs +// 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; diff --git a/src/openhuman/flows/ops.rs b/src/openhuman/flows/ops.rs new file mode 100644 index 000000000..d3c945427 --- /dev/null +++ b/src/openhuman/flows/ops.rs @@ -0,0 +1,213 @@ +//! Business logic for the `flows::` domain: validate-on-save CRUD plus the +//! end-to-end `flows_run` path. Delegated to from `schemas.rs`'s `handle_*` +//! RPC/CLI handlers, mirroring `src/openhuman/cron/ops.rs`. + +use std::sync::Arc; + +use serde_json::{json, Value}; +use tinyflows::model::WorkflowGraph; + +use crate::openhuman::config::Config; +use crate::openhuman::flows::store; +use crate::openhuman::flows::Flow; +use crate::rpc::RpcOutcome; + +/// Overall safety bound on a single `flows_run`. Individual capabilities have +/// their own timeouts (HTTP, sandbox), but a hung LLM/tool call must never let +/// the RPC block indefinitely — this caps the whole run. +const FLOW_RUN_TIMEOUT_SECS: u64 = 600; + +/// Runs a raw graph JSON value through `tinyflows::migrate::migrate` (upgrade +/// an older-schema definition to current), deserializes it, and rejects a +/// structurally invalid graph via `tinyflows::validate::validate` — so a bad +/// graph is caught at the door, before it's ever persisted. +fn validate_and_migrate_graph(graph_json: Value) -> Result { + let migrated = tinyflows::migrate::migrate(graph_json).map_err(|e| e.to_string())?; + let graph: WorkflowGraph = serde_json::from_value(migrated).map_err(|e| e.to_string())?; + tinyflows::validate::validate(&graph).map_err(|e| e.to_string())?; + Ok(graph) +} + +/// Creates a new flow from a name and a raw graph JSON value. +pub async fn flows_create( + config: &Config, + name: String, + graph_json: Value, +) -> Result, String> { + let graph = validate_and_migrate_graph(graph_json)?; + tracing::debug!(target: "flows", %name, node_count = graph.nodes.len(), "[flows] flows_create: persisting new flow"); + let flow = store::create_flow(config, name, graph).map_err(|e| e.to_string())?; + Ok(RpcOutcome::single_log(flow, "flow created")) +} + +/// Loads one flow by id. +pub async fn flows_get(config: &Config, id: &str) -> Result, String> { + let flow = store::get_flow(config, id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("flow '{id}' not found"))?; + Ok(RpcOutcome::single_log(flow, format!("flow loaded: {id}"))) +} + +/// Lists every saved flow. +pub async fn flows_list(config: &Config) -> Result>, String> { + let flows = store::list_flows(config).map_err(|e| e.to_string())?; + Ok(RpcOutcome::single_log(flows, "flows listed")) +} + +/// Updates a flow's name and/or graph. Re-validates the graph (whether newly +/// supplied or the existing one) before persisting, same as `flows_create`. +pub async fn flows_update( + config: &Config, + id: &str, + name: Option, + graph_json: Option, +) -> Result, String> { + let existing = store::get_flow(config, id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("flow '{id}' not found"))?; + + let new_name = name.unwrap_or(existing.name); + let graph = match graph_json { + Some(raw) => validate_and_migrate_graph(raw)?, + None => { + tinyflows::validate::validate(&existing.graph).map_err(|e| e.to_string())?; + existing.graph + } + }; + + tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_update: persisting changes"); + let updated = + store::update_flow_graph(config, id, new_name, graph).map_err(|e| e.to_string())?; + Ok(RpcOutcome::single_log( + updated, + format!("flow updated: {id}"), + )) +} + +/// Deletes a flow by id. +pub async fn flows_delete(config: &Config, id: &str) -> Result, String> { + store::remove_flow(config, id).map_err(|e| e.to_string())?; + tracing::debug!(target: "flows", flow_id = %id, "[flows] flows_delete: removed"); + Ok(RpcOutcome::new( + json!({ "id": id, "removed": true }), + vec![format!("flow removed: {id}")], + )) +} + +/// Enables or disables a flow. B1 note: this does not yet gate anything at +/// run time (`flows_run` still runs a disabled flow on demand, mirroring +/// `cron::rpc::cron_run`'s "Run Now always works" behavior) — `enabled` will +/// gate automatic trigger-driven dispatch once `FlowTriggerSubscriber` +/// (`src/openhuman/flows/bus.rs`) is wired up to actually invoke flows (B2). +pub async fn flows_set_enabled( + config: &Config, + id: &str, + enabled: bool, +) -> Result, String> { + let flow = store::set_enabled(config, id, enabled).map_err(|e| e.to_string())?; + Ok(RpcOutcome::single_log( + flow, + format!("flow {id} enabled={enabled}"), + )) +} + +/// Runs a flow end-to-end: compile → build capabilities → durable +/// checkpointed run → record the outcome onto the flow's summary fields. +/// +/// Uses `tinyflows::engine::run_with_checkpointer` (not the simpler `run`) so +/// a run that pauses at a human-in-the-loop approval gate is durably +/// checkpointed and can survive a process restart (resumed later via a +/// `flows_resume` RPC — B2+; see +/// `my_docs/ohxtf/b1-engine-seam-domain/07-execution-and-hitl.md`). +pub async fn flows_run( + config: &Config, + flow_id: &str, + input: Value, +) -> Result, String> { + let flow = store::get_flow(config, flow_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("flow '{flow_id}' not found"))?; + + // `store::get_flow` already ran the stored `graph_json` through + // `tinyflows::migrate::migrate` before deserializing, so `flow.graph` is + // always on the current schema here. + let compiled = tinyflows::compiler::compile(&flow.graph).map_err(|e| e.to_string())?; + + let config_arc = Arc::new(config.clone()); + // Scope the state store per-flow so two flows never collide on a state key. + let caps = + crate::openhuman::tinyflows::build_capabilities(config_arc, format!("flow:{flow_id}")); + let checkpointer = + crate::openhuman::tinyflows::open_flow_checkpointer(config).map_err(|e| e.to_string())?; + let thread_id = format!("flow:{flow_id}:{}", uuid::Uuid::new_v4()); + + tracing::debug!( + target: "flows", + flow_id = %flow_id, + thread_id = %thread_id, + "[flows] flows_run: starting checkpointed run" + ); + + // Record a failed attempt so `last_run_at`/`last_status` reflect reality + // (a stop-policy engine/capability failure or a timeout) rather than + // leaving the prior success/pending state on the flow. + let record_failed = || { + if let Err(rec_err) = store::record_run(config, flow_id, "failed") { + tracing::warn!( + target: "flows", + flow_id = %flow_id, + error = %rec_err, + "[flows] flows_run: failed to record failed run" + ); + } + }; + + let run = + tinyflows::engine::run_with_checkpointer(&compiled, input, &caps, checkpointer, &thread_id); + let outcome = match tokio::time::timeout( + std::time::Duration::from_secs(FLOW_RUN_TIMEOUT_SECS), + run, + ) + .await + { + Ok(Ok(outcome)) => outcome, + Ok(Err(e)) => { + record_failed(); + tracing::warn!(target: "flows", flow_id = %flow_id, error = %e, "[flows] flows_run: run failed"); + return Err(e.to_string()); + } + Err(_elapsed) => { + record_failed(); + tracing::warn!(target: "flows", flow_id = %flow_id, timeout_secs = FLOW_RUN_TIMEOUT_SECS, "[flows] flows_run: run timed out"); + return Err(format!("flow run timed out after {FLOW_RUN_TIMEOUT_SECS}s")); + } + }; + + let status = if outcome.pending_approvals.is_empty() { + "completed" + } else { + "pending_approval" + }; + store::record_run(config, flow_id, status).map_err(|e| e.to_string())?; + + tracing::info!( + target: "flows", + flow_id = %flow_id, + status, + pending_approvals = outcome.pending_approvals.len(), + "[flows] flows_run: finished" + ); + + Ok(RpcOutcome::single_log( + json!({ + "output": outcome.output, + "pending_approvals": outcome.pending_approvals, + "thread_id": thread_id, + }), + format!("flow run {status}"), + )) +} + +#[cfg(test)] +#[path = "ops_tests.rs"] +mod tests; diff --git a/src/openhuman/flows/ops_tests.rs b/src/openhuman/flows/ops_tests.rs new file mode 100644 index 000000000..a21670d26 --- /dev/null +++ b/src/openhuman/flows/ops_tests.rs @@ -0,0 +1,248 @@ +use super::*; +use crate::openhuman::config::Config; +use serde_json::json; +use tempfile::TempDir; + +fn test_config(tmp: &TempDir) -> Config { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + config +} + +fn trigger_only_graph() -> Value { + json!({ + "name": "trigger-only", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" } + ], + "edges": [] + }) +} + +#[tokio::test] +async fn flows_create_rejects_graph_without_trigger() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let graph_without_trigger = json!({ + "name": "bad", + "nodes": [ { "id": "a", "kind": "output_parser", "name": "A" } ], + "edges": [] + }); + + let err = flows_create(&config, "bad".to_string(), graph_without_trigger) + .await + .expect_err("graph without a trigger must be rejected"); + assert!( + err.contains("trigger"), + "expected a MissingTrigger-style error, got: {err}" + ); +} + +#[tokio::test] +async fn flows_create_get_list_delete_roundtrip() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + .await + .unwrap(); + let flow_id = created.value.id.clone(); + + let fetched = flows_get(&config, &flow_id).await.unwrap(); + assert_eq!(fetched.value.id, flow_id); + assert_eq!(fetched.value.name, "demo"); + + let listed = flows_list(&config).await.unwrap(); + assert_eq!(listed.value.len(), 1); + + flows_delete(&config, &flow_id).await.unwrap(); + assert!(flows_get(&config, &flow_id).await.is_err()); + assert!(flows_list(&config).await.unwrap().value.is_empty()); +} + +#[tokio::test] +async fn flows_set_enabled_toggles() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + .await + .unwrap(); + assert!(created.value.enabled); + + let disabled = flows_set_enabled(&config, &created.value.id, false) + .await + .unwrap(); + assert!(!disabled.value.enabled); + + let enabled = flows_set_enabled(&config, &created.value.id, true) + .await + .unwrap(); + assert!(enabled.value.enabled); +} + +#[tokio::test] +async fn flows_update_replaces_name_and_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + .await + .unwrap(); + + let mut new_graph = trigger_only_graph(); + new_graph["name"] = json!("renamed-graph"); + + let updated = flows_update( + &config, + &created.value.id, + Some("renamed".to_string()), + Some(new_graph), + ) + .await + .unwrap(); + + assert_eq!(updated.value.name, "renamed"); + assert_eq!(updated.value.graph.name, "renamed-graph"); +} + +#[tokio::test] +async fn flows_update_rejects_invalid_replacement_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + .await + .unwrap(); + + let invalid_graph = json!({ + "name": "no-trigger", + "nodes": [ { "id": "a", "kind": "output_parser", "name": "A" } ], + "edges": [] + }); + + let err = flows_update(&config, &created.value.id, None, Some(invalid_graph)) + .await + .expect_err("invalid replacement graph must be rejected"); + assert!(err.contains("trigger")); +} + +#[tokio::test] +async fn flows_run_completes_trigger_only_graph() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let created = flows_create(&config, "demo".to_string(), trigger_only_graph()) + .await + .unwrap(); + + let outcome = flows_run(&config, &created.value.id, json!({ "hello": "world" })) + .await + .unwrap(); + + assert_eq!(outcome.value["pending_approvals"], json!([])); + assert_eq!( + outcome.value["output"]["run"]["trigger"], + json!({ "hello": "world" }) + ); + + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!(reloaded.value.last_status.as_deref(), Some("completed")); + assert!(reloaded.value.last_run_at.is_some()); +} + +#[tokio::test] +async fn flows_run_reports_pending_approval_and_blocks_downstream() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let graph = json!({ + "name": "approval-gated", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "gate", "kind": "output_parser", "name": "Gate", "config": { "requires_approval": true } }, + { "id": "downstream", "kind": "output_parser", "name": "Downstream" } + ], + "edges": [ + { "from_node": "t", "to_node": "gate" }, + { "from_node": "gate", "to_node": "downstream" } + ] + }); + + let created = flows_create(&config, "gated".to_string(), graph) + .await + .unwrap(); + + let outcome = flows_run(&config, &created.value.id, json!({ "x": 1 })) + .await + .unwrap(); + + let pending = outcome.value["pending_approvals"].as_array().unwrap(); + assert!(pending.iter().any(|v| v == "gate")); + assert!(outcome.value["output"]["nodes"]["downstream"].is_null()); + + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!( + reloaded.value.last_status.as_deref(), + Some("pending_approval") + ); +} + +#[tokio::test] +async fn flows_get_missing_flow_errors() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = flows_get(&config, "missing").await.expect_err("must error"); + assert!(err.contains("not found")); +} + +#[tokio::test] +async fn flows_run_missing_flow_errors() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = flows_run(&config, "missing", json!({})) + .await + .expect_err("must error"); + assert!(err.contains("not found")); +} + +#[tokio::test] +async fn flows_run_records_failed_status_when_a_node_errors() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // A `tool_call` with no `slug` errors in the node executor before reaching + // any external service; with the default `on_error: stop` the whole run + // fails deterministically — no network/credentials needed. + let graph = json!({ + "name": "boom", + "nodes": [ + { "id": "t", "kind": "trigger", "name": "Trigger" }, + { "id": "x", "kind": "tool_call", "name": "X" } + ], + "edges": [ { "from_node": "t", "to_node": "x" } ] + }); + + let created = flows_create(&config, "boom".to_string(), graph) + .await + .unwrap(); + + let err = flows_run(&config, &created.value.id, json!({})) + .await + .expect_err("a run whose node errors under on_error:stop must fail"); + assert!(!err.is_empty()); + + // The failed attempt must be recorded, not left on the prior state. + let reloaded = flows_get(&config, &created.value.id).await.unwrap(); + assert_eq!( + reloaded.value.last_status.as_deref(), + Some("failed"), + "a failed run must record last_status=failed" + ); + assert!( + reloaded.value.last_run_at.is_some(), + "a failed run must stamp last_run_at" + ); +} diff --git a/src/openhuman/flows/schemas.rs b/src/openhuman/flows/schemas.rs new file mode 100644 index 000000000..f369e0a27 --- /dev/null +++ b/src/openhuman/flows/schemas.rs @@ -0,0 +1,405 @@ +//! RPC/CLI controller surface for the `flows::` domain. Mirrors +//! `src/openhuman/cron/schemas.rs`'s shape exactly: `schemas(function)` builds +//! one `ControllerSchema`, `all_controller_schemas()`/ +//! `all_registered_controllers()` aggregate them, and each `handle_*` loads +//! config, reads params, awaits the matching `ops::flows_*` fn, and converts +//! the `RpcOutcome` to CLI-compatible JSON. + +use serde::de::DeserializeOwned; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::flows::ops; +use crate::rpc::RpcOutcome; + +fn id_input(comment: &'static str) -> FieldSchema { + FieldSchema { + name: "id", + ty: TypeSchema::String, + comment, + required: true, + } +} + +fn flow_output() -> FieldSchema { + FieldSchema { + name: "flow", + ty: TypeSchema::Ref("Flow"), + comment: "The flow definition.", + required: true, + } +} + +pub fn all_controller_schemas() -> Vec { + vec![ + schemas("create"), + schemas("get"), + schemas("list"), + schemas("update"), + schemas("delete"), + schemas("set_enabled"), + schemas("run"), + ] +} + +pub fn all_registered_controllers() -> Vec { + vec![ + RegisteredController { + schema: schemas("create"), + handler: handle_create, + }, + RegisteredController { + schema: schemas("get"), + handler: handle_get, + }, + RegisteredController { + schema: schemas("list"), + handler: handle_list, + }, + RegisteredController { + schema: schemas("update"), + handler: handle_update, + }, + RegisteredController { + schema: schemas("delete"), + handler: handle_delete, + }, + RegisteredController { + schema: schemas("set_enabled"), + handler: handle_set_enabled, + }, + RegisteredController { + schema: schemas("run"), + handler: handle_run, + }, + ] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "create" => ControllerSchema { + namespace: "flows", + function: "create", + description: "Create a new saved automation workflow from a tinyflows graph.", + inputs: vec![ + FieldSchema { + name: "name", + ty: TypeSchema::String, + comment: "Human-readable flow name.", + required: true, + }, + FieldSchema { + name: "graph", + ty: TypeSchema::Json, + comment: + "A tinyflows WorkflowGraph (nodes + edges); validated and migrated on save.", + required: true, + }, + ], + outputs: vec![flow_output()], + }, + "get" => ControllerSchema { + namespace: "flows", + function: "get", + description: "Load one saved flow by id.", + inputs: vec![id_input("Identifier of the flow to load.")], + outputs: vec![flow_output()], + }, + "list" => ControllerSchema { + namespace: "flows", + function: "list", + description: "List all saved flows.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "flows", + ty: TypeSchema::Array(Box::new(TypeSchema::Ref("Flow"))), + comment: "Flows currently stored in the workspace.", + required: true, + }], + }, + "update" => ControllerSchema { + namespace: "flows", + function: "update", + description: "Update a saved flow's name and/or graph; re-validates before persisting.", + inputs: vec![ + id_input("Identifier of the flow to update."), + FieldSchema { + name: "name", + ty: TypeSchema::Option(Box::new(TypeSchema::String)), + comment: "New name, if changing it.", + required: false, + }, + FieldSchema { + name: "graph", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Replacement WorkflowGraph, if changing it.", + required: false, + }, + ], + outputs: vec![flow_output()], + }, + "delete" => ControllerSchema { + namespace: "flows", + function: "delete", + description: "Delete a saved flow by id.", + inputs: vec![id_input("Identifier of the flow to delete.")], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Object { + fields: vec![ + FieldSchema { + name: "id", + ty: TypeSchema::String, + comment: "Identifier that was requested for removal.", + required: true, + }, + FieldSchema { + name: "removed", + ty: TypeSchema::Bool, + comment: "True when the flow was removed.", + required: true, + }, + ], + }, + comment: "Removal result payload.", + required: true, + }], + }, + "set_enabled" => ControllerSchema { + namespace: "flows", + function: "set_enabled", + description: "Enable or disable a saved flow.", + inputs: vec![ + id_input("Identifier of the flow to toggle."), + FieldSchema { + name: "enabled", + ty: TypeSchema::Bool, + comment: "New enabled state.", + required: true, + }, + ], + outputs: vec![flow_output()], + }, + "run" => ControllerSchema { + namespace: "flows", + function: "run", + description: + "Run a saved flow to completion (or until it pauses on a human-approval gate).", + inputs: vec![ + id_input("Identifier of the flow to run."), + FieldSchema { + name: "input", + ty: TypeSchema::Option(Box::new(TypeSchema::Json)), + comment: "Trigger payload seeded into the run; defaults to null.", + required: false, + }, + ], + outputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::Object { + fields: vec![ + FieldSchema { + name: "output", + ty: TypeSchema::Json, + comment: "The run's final state (per-node items, trigger payload).", + required: true, + }, + FieldSchema { + name: "pending_approvals", + ty: TypeSchema::Array(Box::new(TypeSchema::String)), + comment: + "Node ids paused awaiting human approval; empty once completed.", + required: true, + }, + FieldSchema { + name: "thread_id", + ty: TypeSchema::String, + comment: + "Durable checkpoint thread id for this run (needed to resume).", + required: true, + }, + ], + }, + comment: "Run outcome payload.", + required: true, + }], + }, + _other => ControllerSchema { + namespace: "flows", + function: "unknown", + description: "Unknown flows controller function.", + inputs: vec![FieldSchema { + name: "function", + ty: TypeSchema::String, + comment: "Unknown function requested for schema lookup.", + required: true, + }], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +fn handle_create(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let name = read_required::(¶ms, "name")?; + let graph = read_required::(¶ms, "graph")?; + to_json(ops::flows_create(&config, name, graph).await?) + }) +} + +fn handle_get(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + to_json(ops::flows_get(&config, id.trim()).await?) + }) +} + +fn handle_list(_params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + to_json(ops::flows_list(&config).await?) + }) +} + +fn handle_update(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + let name = params + .get("name") + .filter(|v| !v.is_null()) + .map(|v| serde_json::from_value(v.clone())) + .transpose() + .map_err(|e| format!("invalid 'name': {e}"))?; + let graph = params.get("graph").filter(|v| !v.is_null()).cloned(); + to_json(ops::flows_update(&config, id.trim(), name, graph).await?) + }) +} + +fn handle_delete(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + to_json(ops::flows_delete(&config, id.trim()).await?) + }) +} + +fn handle_set_enabled(params: Map) -> ControllerFuture { + Box::pin(async move { + let config = config_rpc::load_config_with_timeout().await?; + let id = read_required::(¶ms, "id")?; + let enabled = params + .get("enabled") + .and_then(Value::as_bool) + .ok_or_else(|| "missing required param 'enabled'".to_string())?; + to_json(ops::flows_set_enabled(&config, id.trim(), enabled).await?) + }) +} + +fn handle_run(params: Map) -> ControllerFuture { + Box::pin(async move { + 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?) + }) +} + +fn read_required(params: &Map, key: &str) -> Result { + let value = params + .get(key) + .cloned() + .ok_or_else(|| format!("missing required param '{key}'"))?; + serde_json::from_value(value).map_err(|e| format!("invalid '{key}': {e}")) +} + +fn to_json(outcome: RpcOutcome) -> Result { + outcome.into_cli_compatible_json() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn all_controller_schemas_covers_every_supported_function() { + let names: Vec<_> = all_controller_schemas() + .into_iter() + .map(|s| s.function) + .collect(); + assert_eq!( + names, + vec![ + "create", + "get", + "list", + "update", + "delete", + "set_enabled", + "run" + ] + ); + } + + #[test] + fn all_registered_controllers_has_handler_per_schema() { + let controllers = all_registered_controllers(); + assert_eq!(controllers.len(), 7); + let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect(); + assert_eq!( + names, + vec![ + "create", + "get", + "list", + "update", + "delete", + "set_enabled", + "run" + ] + ); + } + + #[test] + fn schemas_create_requires_name_and_graph() { + let s = schemas("create"); + assert_eq!(s.namespace, "flows"); + let required: Vec<_> = s + .inputs + .iter() + .filter(|f| f.required) + .map(|f| f.name) + .collect(); + assert_eq!(required, vec!["name", "graph"]); + } + + #[test] + fn schemas_run_input_is_optional() { + let s = schemas("run"); + let input = s.inputs.iter().find(|f| f.name == "input").unwrap(); + assert!(!input.required); + } + + #[test] + fn schemas_unknown_function_returns_placeholder() { + let s = schemas("does-not-exist"); + assert_eq!(s.function, "unknown"); + assert_eq!(s.outputs[0].name, "error"); + } + + #[test] + fn read_required_errors_when_missing() { + let params = Map::new(); + let err = read_required::(¶ms, "id").unwrap_err(); + assert!(err.contains("missing required param 'id'")); + } +} diff --git a/src/openhuman/flows/store.rs b/src/openhuman/flows/store.rs new file mode 100644 index 000000000..d62365b2d --- /dev/null +++ b/src/openhuman/flows/store.rs @@ -0,0 +1,301 @@ +//! SQLite persistence for the `flows::` domain. +//! +//! Mirrors `src/openhuman/cron/store.rs`'s idiom: a `with_connection` helper +//! opens (and migrates) a dedicated SQLite database under the workspace, and +//! every public function takes `&Config` first and returns `anyhow::Result`. +//! +//! Two tables: +//! - `flow_definitions` — one row per saved [`Flow`], with the graph stored as +//! JSON text (`graph_json`). +//! - `flow_state` — a generic namespaced key/value table backing +//! `tinyflows::caps::StateStore` (see `src/openhuman/tinyflows/caps.rs`). +//! +//! There is deliberately **no** `flow_checkpoints` table here: the crate's own +//! `tinyagents::SqliteCheckpointer` owns checkpoint persistence in a separate +//! `checkpoints.db` (see `src/openhuman/tinyflows/mod.rs::open_flow_checkpointer`). + +use crate::openhuman::config::Config; +use crate::openhuman::flows::Flow; +use anyhow::{Context, Result}; +use chrono::Utc; +use rusqlite::{params, Connection}; +use uuid::Uuid; + +/// Opens (creating/migrating as needed) the flows SQLite database and runs `f` +/// against the connection. +fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) -> Result { + let db_path = config.workspace_dir.join("flows").join("flows.db"); + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("Failed to create flows directory: {}", parent.display()))?; + } + + let conn = Connection::open(&db_path) + .with_context(|| format!("Failed to open flows DB: {}", db_path.display()))?; + + conn.execute_batch( + // `busy_timeout` retries (rather than immediately erroring + // `SQLITE_BUSY`) when a concurrent run/state write holds the lock; WAL + // lets readers and a writer proceed together. Both are safe to re-issue + // on every open (WAL is a persistent db-file setting; busy_timeout is + // per-connection). + "PRAGMA busy_timeout = 5000; + PRAGMA journal_mode = WAL; + PRAGMA foreign_keys = ON; + CREATE TABLE IF NOT EXISTS flow_definitions ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + graph_json TEXT NOT NULL, + enabled INTEGER NOT NULL DEFAULT 1, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + last_run_at TEXT, + last_status TEXT + ); + CREATE INDEX IF NOT EXISTS idx_flow_definitions_enabled ON flow_definitions(enabled); + + CREATE TABLE IF NOT EXISTS flow_state ( + namespace TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT NOT NULL, + PRIMARY KEY (namespace, key) + );", + ) + .context("Failed to initialize flows schema")?; + + tracing::debug!(db = %db_path.display(), "[flows] store opened"); + + f(&conn) +} + +/// Inserts or fully replaces a flow definition row. +pub fn upsert_flow(config: &Config, flow: &Flow) -> Result<()> { + let graph_json = serde_json::to_string(&flow.graph).context("Failed to serialize graph")?; + with_connection(config, |conn| { + conn.execute( + "INSERT INTO flow_definitions + (id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(id) DO UPDATE SET + name = excluded.name, + graph_json = excluded.graph_json, + enabled = excluded.enabled, + updated_at = excluded.updated_at, + last_run_at = excluded.last_run_at, + last_status = excluded.last_status", + params![ + flow.id, + flow.name, + graph_json, + if flow.enabled { 1 } else { 0 }, + flow.created_at, + flow.updated_at, + flow.last_run_at, + flow.last_status, + ], + ) + .context("Failed to upsert flow definition")?; + tracing::debug!(flow_id = %flow.id, "[flows] upserted flow definition"); + Ok(()) + }) +} + +/// Creates a brand-new [`Flow`] row from a name + validated graph, stamping +/// fresh id/timestamps, and returns the persisted record. +pub fn create_flow( + config: &Config, + name: String, + graph: tinyflows::model::WorkflowGraph, +) -> Result { + let now = Utc::now().to_rfc3339(); + let flow = Flow { + id: Uuid::new_v4().to_string(), + name, + enabled: true, + graph, + created_at: now.clone(), + updated_at: now, + last_run_at: None, + last_status: None, + }; + upsert_flow(config, &flow)?; + Ok(flow) +} + +/// Loads one flow by id, running its stored `graph_json` through +/// `tinyflows::migrate::migrate` before deserializing so a graph persisted +/// under an older `schema_version` is upgraded on read. +pub fn get_flow(config: &Config, id: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status + FROM flow_definitions WHERE id = ?1", + )?; + let mut rows = stmt.query(params![id])?; + match rows.next()? { + Some(row) => Ok(Some(map_flow_row(row)?)), + None => Ok(None), + } + }) +} + +/// Lists all saved flows, migrating each graph on read (see [`get_flow`]). +pub fn list_flows(config: &Config) -> Result> { + with_connection(config, |conn| { + let mut stmt = conn.prepare( + "SELECT id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status + FROM flow_definitions ORDER BY created_at ASC", + )?; + let rows = stmt.query_map([], map_flow_row)?; + let mut flows = Vec::new(); + for row in rows { + flows.push(row?); + } + Ok(flows) + }) +} + +/// Deletes a flow by id. Returns an error if no such flow exists. +pub fn remove_flow(config: &Config, id: &str) -> Result<()> { + let changed = with_connection(config, |conn| { + conn.execute("DELETE FROM flow_definitions WHERE id = ?1", params![id]) + .context("Failed to delete flow definition") + })?; + if changed == 0 { + anyhow::bail!("flow '{id}' not found"); + } + tracing::debug!(flow_id = %id, "[flows] removed flow definition"); + Ok(()) +} + +/// Toggles a flow's `enabled` flag, returning the updated record. +pub fn set_enabled(config: &Config, id: &str, enabled: bool) -> Result { + let now = Utc::now().to_rfc3339(); + let changed = with_connection(config, |conn| { + conn.execute( + "UPDATE flow_definitions SET enabled = ?1, updated_at = ?2 WHERE id = ?3", + params![if enabled { 1 } else { 0 }, now, id], + ) + .context("Failed to update flow enabled state") + })?; + if changed == 0 { + anyhow::bail!("flow '{id}' not found"); + } + tracing::debug!(flow_id = %id, enabled, "[flows] set_enabled"); + get_flow(config, id)?.ok_or_else(|| anyhow::anyhow!("flow '{id}' not found after update")) +} + +/// Replaces a flow's name/graph (re-validated by the caller before this is +/// invoked) in place, bumping `updated_at`. +pub fn update_flow_graph( + config: &Config, + id: &str, + name: String, + graph: tinyflows::model::WorkflowGraph, +) -> Result { + let graph_json = serde_json::to_string(&graph).context("Failed to serialize graph")?; + let now = Utc::now().to_rfc3339(); + // Targeted UPDATE of only the editable columns, so a concurrent + // `set_enabled` / `record_run` isn't clobbered by writing back a stale + // `enabled` / `last_run_at` / `last_status` from a read-modify-write. + let changed = with_connection(config, |conn| { + conn.execute( + "UPDATE flow_definitions SET name = ?1, graph_json = ?2, updated_at = ?3 WHERE id = ?4", + params![name, graph_json, now, id], + ) + .context("Failed to update flow") + })?; + if changed == 0 { + anyhow::bail!("flow '{id}' not found"); + } + get_flow(config, id)?.ok_or_else(|| anyhow::anyhow!("flow '{id}' not found")) +} + +/// Records the outcome of a `flows_run` invocation onto the flow's summary +/// fields (`last_run_at` / `last_status`). +pub fn record_run(config: &Config, id: &str, status: &str) -> Result<()> { + let now = Utc::now().to_rfc3339(); + let changed = with_connection(config, |conn| { + conn.execute( + "UPDATE flow_definitions SET last_run_at = ?1, last_status = ?2 WHERE id = ?3", + params![now, status, id], + ) + .context("Failed to record flow run") + })?; + if changed == 0 { + anyhow::bail!("flow '{id}' not found"); + } + tracing::debug!(flow_id = %id, status, "[flows] recorded run"); + Ok(()) +} + +fn map_flow_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let graph_raw: String = row.get(2)?; + let raw_value: serde_json::Value = + serde_json::from_str(&graph_raw).map_err(sql_conversion_error)?; + let migrated = tinyflows::migrate::migrate(raw_value).map_err(sql_conversion_error)?; + let graph: tinyflows::model::WorkflowGraph = + serde_json::from_value(migrated).map_err(sql_conversion_error)?; + + Ok(Flow { + id: row.get(0)?, + name: row.get(1)?, + graph, + enabled: row.get::<_, i64>(3)? != 0, + created_at: row.get(4)?, + updated_at: row.get(5)?, + last_run_at: row.get(6)?, + last_status: row.get(7)?, + }) +} + +fn sql_conversion_error(err: E) -> rusqlite::Error { + rusqlite::Error::ToSqlConversionFailure(Box::new(err)) +} + +/// Loads a value from the `flow_state` KV table, scoped to `namespace`. +/// +/// Backs `tinyflows::caps::StateStore::load` via +/// `src/openhuman/tinyflows/caps.rs::FlowStateStore`. +pub fn kv_get(config: &Config, namespace: &str, key: &str) -> Result> { + with_connection(config, |conn| { + let mut stmt = + conn.prepare("SELECT value FROM flow_state WHERE namespace = ?1 AND key = ?2")?; + let mut rows = stmt.query(params![namespace, key])?; + match rows.next()? { + Some(row) => { + let raw: String = row.get(0)?; + let value: serde_json::Value = + serde_json::from_str(&raw).map_err(sql_conversion_error)?; + Ok(Some(value)) + } + None => Ok(None), + } + }) +} + +/// Stores a value into the `flow_state` KV table, scoped to `namespace`. +/// +/// Backs `tinyflows::caps::StateStore::store` via +/// `src/openhuman/tinyflows/caps.rs::FlowStateStore`. +pub fn kv_set( + config: &Config, + namespace: &str, + key: &str, + value: &serde_json::Value, +) -> Result<()> { + let raw = serde_json::to_string(value).context("Failed to serialize flow state value")?; + with_connection(config, |conn| { + conn.execute( + "INSERT INTO flow_state (namespace, key, value) VALUES (?1, ?2, ?3) + ON CONFLICT(namespace, key) DO UPDATE SET value = excluded.value", + params![namespace, key, raw], + ) + .context("Failed to store flow state value")?; + Ok(()) + }) +} + +#[cfg(test)] +#[path = "store_tests.rs"] +mod tests; diff --git a/src/openhuman/flows/store_tests.rs b/src/openhuman/flows/store_tests.rs new file mode 100644 index 000000000..249a5f5ea --- /dev/null +++ b/src/openhuman/flows/store_tests.rs @@ -0,0 +1,169 @@ +use super::*; +use crate::openhuman::config::Config; +use tempfile::TempDir; +use tinyflows::model::{Node, NodeKind, WorkflowGraph}; + +fn test_config(tmp: &TempDir) -> Config { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + config +} + +fn trigger_graph() -> WorkflowGraph { + WorkflowGraph { + nodes: vec![Node { + id: "t".to_string(), + kind: NodeKind::Trigger, + type_version: 1, + name: "Trigger".to_string(), + config: serde_json::Value::Null, + ports: Vec::new(), + position: None, + }], + ..Default::default() + } +} + +#[test] +fn create_get_list_delete_roundtrip() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + assert_eq!(flow.name, "demo"); + assert!(flow.enabled); + + let fetched = get_flow(&config, &flow.id).unwrap().expect("flow present"); + assert_eq!(fetched.id, flow.id); + assert_eq!(fetched.graph, flow.graph); + + let listed = list_flows(&config).unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].id, flow.id); + + remove_flow(&config, &flow.id).unwrap(); + assert!(get_flow(&config, &flow.id).unwrap().is_none()); + assert!(list_flows(&config).unwrap().is_empty()); +} + +#[test] +fn get_flow_returns_none_for_unknown_id() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + assert!(get_flow(&config, "missing").unwrap().is_none()); +} + +#[test] +fn remove_flow_errors_when_not_found() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let err = remove_flow(&config, "missing").unwrap_err(); + assert!(err.to_string().contains("not found")); +} + +#[test] +fn set_enabled_toggles_and_persists() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + assert!(flow.enabled); + + let disabled = set_enabled(&config, &flow.id, false).unwrap(); + assert!(!disabled.enabled); + + let reloaded = get_flow(&config, &flow.id).unwrap().unwrap(); + assert!(!reloaded.enabled); + + let enabled = set_enabled(&config, &flow.id, true).unwrap(); + assert!(enabled.enabled); +} + +#[test] +fn update_flow_graph_bumps_updated_at_and_preserves_created_at() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + + let mut new_graph = trigger_graph(); + new_graph.name = "renamed-graph".to_string(); + let updated = update_flow_graph(&config, &flow.id, "renamed".to_string(), new_graph).unwrap(); + + assert_eq!(updated.name, "renamed"); + assert_eq!(updated.created_at, flow.created_at); + assert_eq!(updated.graph.name, "renamed-graph"); +} + +#[test] +fn record_run_sets_last_run_fields() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let flow = create_flow(&config, "demo".to_string(), trigger_graph()).unwrap(); + assert!(flow.last_run_at.is_none()); + + record_run(&config, &flow.id, "completed").unwrap(); + let reloaded = get_flow(&config, &flow.id).unwrap().unwrap(); + assert!(reloaded.last_run_at.is_some()); + assert_eq!(reloaded.last_status.as_deref(), Some("completed")); +} + +#[test] +fn stored_graph_older_than_current_schema_is_migrated_on_read() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // Insert a raw, versionless graph row directly (bypassing create_flow's + // typed path) to simulate a definition persisted by an older crate build. + let legacy_graph_json = serde_json::json!({ + "name": "legacy", + "nodes": [{ "id": "t", "kind": "trigger", "name": "Trigger" }], + "edges": [] + }) + .to_string(); + + with_connection(&config, |conn| { + conn.execute( + "INSERT INTO flow_definitions + (id, name, graph_json, enabled, created_at, updated_at, last_run_at, last_status) + VALUES ('legacy-1', 'legacy', ?1, 1, '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', NULL, NULL)", + rusqlite::params![legacy_graph_json], + )?; + Ok(()) + }) + .unwrap(); + + let loaded = get_flow(&config, "legacy-1").unwrap().expect("row present"); + assert_eq!( + loaded.graph.schema_version, + tinyflows::model::CURRENT_SCHEMA_VERSION + ); + assert_eq!(loaded.graph.nodes.len(), 1); +} + +#[test] +fn kv_get_set_round_trips_and_is_namespace_scoped() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + assert!(kv_get(&config, "ns1", "k").unwrap().is_none()); + + kv_set(&config, "ns1", "k", &serde_json::json!({"v": 1})).unwrap(); + assert_eq!( + kv_get(&config, "ns1", "k").unwrap(), + Some(serde_json::json!({"v": 1})) + ); + + // A different namespace does not see ns1's value. + assert!(kv_get(&config, "ns2", "k").unwrap().is_none()); + + // Overwrite. + kv_set(&config, "ns1", "k", &serde_json::json!(2)).unwrap(); + assert_eq!( + kv_get(&config, "ns1", "k").unwrap(), + Some(serde_json::json!(2)) + ); +} diff --git a/src/openhuman/flows/types.rs b/src/openhuman/flows/types.rs new file mode 100644 index 000000000..c3b13e67e --- /dev/null +++ b/src/openhuman/flows/types.rs @@ -0,0 +1,71 @@ +//! The [`Flow`] entity: a saved automation workflow definition. +//! +//! Wraps `tinyflows::model::WorkflowGraph` with the metadata OpenHuman needs to +//! store, list, and track runs for a saved flow. The graph itself is the +//! portable, tinyflows-owned contract (validated + migrated on load); this +//! struct is the OpenHuman-side record around it. + +use serde::{Deserialize, Serialize}; +use tinyflows::model::WorkflowGraph; + +/// A saved automation workflow: a `tinyflows` graph plus OpenHuman-side +/// bookkeeping (enablement, run history summary). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Flow { + /// Stable identifier (UUID) for this flow. + pub id: String, + /// Human-readable name shown in the Workflows UI. + pub name: String, + /// Whether this flow may currently be triggered (B2) / run. + pub enabled: bool, + /// The validated, migrated workflow graph. + pub graph: WorkflowGraph, + /// RFC3339 creation timestamp. + pub created_at: String, + /// RFC3339 last-update timestamp. + pub updated_at: String, + /// RFC3339 timestamp of the most recent run, if any. + pub last_run_at: Option, + /// Outcome of the most recent run: `"completed"` | `"pending_approval"` | `"failed"`. + pub last_status: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + use tinyflows::model::{Node, NodeKind}; + + fn sample_graph() -> WorkflowGraph { + WorkflowGraph { + nodes: vec![Node { + id: "t".to_string(), + kind: NodeKind::Trigger, + type_version: 1, + name: "Trigger".to_string(), + config: serde_json::Value::Null, + ports: Vec::new(), + position: None, + }], + ..Default::default() + } + } + + #[test] + fn flow_round_trips_through_json() { + let flow = Flow { + id: "flow_1".to_string(), + name: "demo".to_string(), + enabled: true, + graph: sample_graph(), + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + last_run_at: None, + last_status: None, + }; + let json = serde_json::to_string(&flow).expect("serialize"); + let back: Flow = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back.id, flow.id); + assert_eq!(back.graph, flow.graph); + assert!(back.last_run_at.is_none()); + } +} diff --git a/src/openhuman/mcp_server/resources.rs b/src/openhuman/mcp_server/resources.rs index 2a08f744e..a289ad296 100644 --- a/src/openhuman/mcp_server/resources.rs +++ b/src/openhuman/mcp_server/resources.rs @@ -259,6 +259,18 @@ const RESOURCE_CATALOG: &[PromptResource] = &[ description: "Background awareness agent: diffs the user's world, prepares context, and decides what to do.", content: include_str!("../subconscious/agent/prompt.md"), }, + PromptResource { + uri: "openhuman://prompts/agents/frontend_agent", + name: "frontend_agent", + description: "Fast, always-on front end of the split-brain orchestration loop: triages incoming sessions and routes macro-instructions to the reasoning core.", + content: include_str!("../orchestration/frontend_agent/prompt.md"), + }, + PromptResource { + uri: "openhuman://prompts/agents/reasoning_agent", + name: "reasoning_agent", + description: "Reasoning core of the split-brain orchestration loop: executes the front end's macro-instructions and compiles the channel reply.", + content: include_str!("../orchestration/reasoning_agent/prompt.md"), + }, PromptResource { uri: "openhuman://prompts/agents/skill_setup", name: "skill_setup", diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 511ab1b93..5daa41d89 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -50,6 +50,7 @@ pub mod doctor; pub mod embeddings; pub mod encryption; pub mod file_state; +pub mod flows; pub mod harness_init; pub mod health; pub mod heartbeat; @@ -123,6 +124,7 @@ pub mod text_input; pub mod thread_goals; pub mod threads; pub mod tinyagents; +pub mod tinyflows; pub mod tinyplace; pub mod tls; pub mod todos; diff --git a/src/openhuman/orchestration/graph/mod.rs b/src/openhuman/orchestration/graph/mod.rs index b3b1273a7..dc03027b4 100644 --- a/src/openhuman/orchestration/graph/mod.rs +++ b/src/openhuman/orchestration/graph/mod.rs @@ -39,7 +39,7 @@ use tinyagents::graph::{ use crate::openhuman::config::Config; use crate::openhuman::tinyagents::observability::GraphTracingSink; -use crate::openhuman::tinyagents::SqlRunLedgerCheckpointer; +use tinyagents::graph::SqliteCheckpointer; pub use state::{CompressedEntry, OrchestrationState, WorldDiff, WorldDiffEntry}; @@ -381,7 +381,16 @@ pub async fn run_orchestration_graph( let threshold = config.orchestration.effective_evict_threshold(); let thread_id = format!("orchestration:{}", state.session_id); let label = thread_id.clone(); - let checkpointer = Arc::new(SqlRunLedgerCheckpointer::::new(config)); + // `SqlRunLedgerCheckpointer` was retired in favor of the crate's own + // `SqliteCheckpointer` (see `agent_orchestration/delegation.rs`); mirrors + // that swap here with a dedicated `orchestration_graph_checkpoints.db`. + let checkpoint_db = config + .workspace_dir + .join("orchestration_graph_checkpoints.db"); + let checkpointer = Arc::new( + SqliteCheckpointer::::open(&checkpoint_db) + .map_err(|e| anyhow::anyhow!("open durable orchestration checkpoint store: {e}"))?, + ); tracing::debug!( target: LOG, session_id = %state.session_id, %thread_id, diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index 5cc46d01c..d39948cac 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -727,9 +727,9 @@ impl OrchestrationRuntime for ProductionRuntime { mod tests { use super::*; use crate::openhuman::orchestration::types::OrchestrationMessage; - use crate::openhuman::tinyagents::SqlRunLedgerCheckpointer; use std::sync::atomic::{AtomicUsize, Ordering}; use tinyagents::graph::checkpoint::Checkpointer; + use tinyagents::graph::SqliteCheckpointer; fn test_config(tmp: &tempfile::TempDir) -> Config { Config { @@ -910,7 +910,13 @@ mod tests { .unwrap(); // Checkpoints persisted → kill/restart could resume without re-sending. - let cp = SqlRunLedgerCheckpointer::::new(config); + // Same `orchestration_graph_checkpoints.db` path `run_orchestration_graph` + // opens (see `orchestration/graph/mod.rs`). + let checkpoint_db = config + .workspace_dir + .join("orchestration_graph_checkpoints.db"); + let cp = SqliteCheckpointer::::open(&checkpoint_db) + .expect("open checkpoint store"); let list = cp.list("orchestration:h1").await.expect("list checkpoints"); assert!(!list.is_empty(), "wake cycle persisted checkpoints"); } diff --git a/src/openhuman/tinyflows/caps.rs b/src/openhuman/tinyflows/caps.rs new file mode 100644 index 000000000..2655df474 --- /dev/null +++ b/src/openhuman/tinyflows/caps.rs @@ -0,0 +1,469 @@ +//! The capability seam: five adapters implementing `tinyflows::caps` traits +//! over real OpenHuman services. +//! +//! Each tinyflows integration node hands its **whole** `node.config` to the +//! matching trait method — the adapter interprets a free-form JSON value the +//! flow author wrote, pulling a connection ref out of `config["connection_ref"]` +//! where relevant. See `my_docs/ohxtf/b1-engine-seam-domain/04-capability-seam.md` +//! for the source-verified node → trait contract this mirrors. +//! +//! All host errors are mapped to `tinyflows::error::EngineError::Capability`, +//! per the crate's contract (`caps` traits return `tinyflows::error::Result`). + +use std::sync::Arc; + +use anyhow::Context; +use async_trait::async_trait; +use serde_json::{json, Value}; +use tinyagents::graph::SqliteCheckpointer; +use tinyflows::caps::{ + Capabilities, CodeLanguage, CodeRunner, HttpClient, LlmProvider, StateStore, ToolInvoker, +}; +use tinyflows::error::{EngineError, Result}; + +use crate::openhuman::agent::harness::definition::SandboxMode; +use crate::openhuman::composio::client::{ + create_composio_client, direct_execute, ComposioClientKind, +}; +use crate::openhuman::config::{Config, HttpRequestConfig}; +use crate::openhuman::flows; +use crate::openhuman::inference::provider::{ + create_chat_provider, ChatMessage, ChatRequest, UsageInfo, +}; +use crate::openhuman::sandbox::{execute_in_sandbox, resolve_sandbox_policy}; +use crate::openhuman::security::SecurityPolicy; +use crate::openhuman::tools::traits::Tool as _; +use crate::openhuman::tools::HttpRequestTool; + +/// Maps a `UsageInfo` (not `Serialize`) into a JSON value field-by-field, so +/// [`OpenHumanLlm::complete`] can surface it in its response `Value` without +/// requiring an upstream `Serialize` impl change. +fn usage_to_json(usage: &Option) -> Value { + match usage { + None => Value::Null, + Some(u) => json!({ + "input_tokens": u.input_tokens, + "output_tokens": u.output_tokens, + "context_window": u.context_window, + "cached_input_tokens": u.cached_input_tokens, + "cache_creation_tokens": u.cache_creation_tokens, + "reasoning_tokens": u.reasoning_tokens, + "charged_amount_usd": u.charged_amount_usd, + }), + } +} + +/// [`LlmProvider`] adapter over OpenHuman's inference stack +/// (`src/openhuman/inference/provider/`). +/// +/// The `agent` node is single-completion in tinyflows 0.2 (no tool-calling +/// loop, no sub-ports), so `complete` performs exactly one `provider.chat` +/// call and returns its result — no agent loop is driven here. +pub struct OpenHumanLlm { + pub config: Arc, +} + +#[async_trait] +impl LlmProvider for OpenHumanLlm { + async fn complete(&self, request: Value, conn: Option<&str>) -> Result { + if let Some(c) = conn { + // B1 does not resolve `connection_ref` to a specific BYOK account — + // `create_chat_provider` picks the configured provider for `role`. + tracing::debug!(target: "flows", conn = %c, "[flows] llm conn (not resolved in B1)"); + } + + let role = request + .get("role") + .and_then(Value::as_str) + .unwrap_or("summarization"); + let temperature = request + .get("temperature") + .and_then(Value::as_f64) + .unwrap_or(0.7); + let max_tokens = request + .get("max_tokens") + .and_then(Value::as_u64) + .and_then(|n| u32::try_from(n).ok()); + + let messages: Vec = match request.get("messages").and_then(Value::as_array) { + Some(entries) if !entries.is_empty() => entries + .iter() + .filter_map(|entry| { + let content = entry.get("content").and_then(Value::as_str)?.to_string(); + let role = entry.get("role").and_then(Value::as_str).unwrap_or("user"); + Some(match role { + "system" => ChatMessage::system(content), + "assistant" => ChatMessage::assistant(content), + "tool" => ChatMessage::tool(content), + _ => ChatMessage::user(content), + }) + }) + .collect(), + _ => { + let prompt = request + .get("prompt") + .and_then(Value::as_str) + .unwrap_or_default(); + vec![ChatMessage::user(prompt)] + } + }; + + tracing::debug!( + target: "flows", + role, + message_count = messages.len(), + "[flows] llm.complete: dispatching agent-node completion" + ); + + let (provider, model) = create_chat_provider(role, &self.config) + .map_err(|e| EngineError::Capability(e.to_string()))?; + + let response = provider + .chat( + ChatRequest { + messages: &messages, + tools: None, + stream: None, + max_tokens, + }, + &model, + temperature, + ) + .await + .map_err(|e| EngineError::Capability(e.to_string()))?; + + Ok(json!({ + "text": response.text, + "tool_calls": response.tool_calls, + "usage": usage_to_json(&response.usage), + "reasoning_content": response.reasoning_content, + })) + } +} + +/// [`ToolInvoker`] adapter over Composio (`src/openhuman/composio/client.rs`). +/// +/// **B1 deviation (tracked, see `my_docs/ohxtf/commons/11-gotchas-and-decisions.md`):** +/// `connection_ref` is logged but not forwarded — `execute_tool` (backend mode) +/// takes no connection id and resolves the ambient signed-in account; direct +/// mode uses `config.composio.entity_id`. Fine for a single-account desktop +/// user; must be resolved before multi-account or B2 trigger runs. There is +/// also no curated-tool-set / scope filter yet — `invoke` will call any slug. +pub struct OpenHumanTools { + pub config: Arc, +} + +#[async_trait] +impl ToolInvoker for OpenHumanTools { + async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result { + if let Some(c) = conn { + tracing::debug!( + target: "flows", + %slug, + conn = %c, + "[flows] tool conn (backend resolves ambient account; not forwarded in B1)" + ); + } + + let kind = create_composio_client(&self.config) + .map_err(|e| EngineError::Capability(e.to_string()))?; + let args_opt = if args.is_null() { None } else { Some(args) }; + + tracing::debug!(target: "flows", %slug, mode = kind.mode(), "[flows] tool_call: invoking composio tool"); + + let response = match kind { + ComposioClientKind::Backend(client) => client + .execute_tool(slug, args_opt) + .await + .map_err(|e| EngineError::Capability(e.to_string()))?, + ComposioClientKind::Direct(tool) => { + direct_execute(&tool, slug, args_opt, &self.config.composio.entity_id, None) + .await + .map_err(|e| EngineError::Capability(e.to_string()))? + } + }; + + serde_json::to_value(response).map_err(|e| EngineError::Capability(e.to_string())) + } +} + +/// [`HttpClient`] adapter over `HttpRequestTool` +/// (`src/openhuman/tools/impl/network/http_request.rs`). Allowlist + DNS-rebind +/// guard + Network gating live inside `execute`, so this adapter gets them for +/// free. +pub struct OpenHumanHttp { + pub security: Arc, + pub http_config: HttpRequestConfig, +} + +#[async_trait] +impl HttpClient for OpenHumanHttp { + async fn request(&self, request: Value, conn: Option<&str>) -> Result { + if let Some(c) = conn { + tracing::debug!(target: "flows", conn = %c, "[flows] http conn (not resolved in B1)"); + } + + let tool = HttpRequestTool::new( + self.security.clone(), + self.http_config.allowed_domains.clone(), + self.http_config.max_response_size, + self.http_config.timeout_secs, + ); + + tracing::debug!( + target: "flows", + method = ?request.get("method"), + url = ?request.get("url"), + "[flows] http_request: dispatching outbound request" + ); + + // `request` is already `{ method, url, headers?, body? }` — the node's + // config is the request descriptor; `HttpRequestTool::execute` reads + // only those keys and ignores the rest (e.g. `connection_ref`, + // `on_error`), so passing the whole config through is safe. + let result = tool + .execute(request) + .await + .map_err(|e| EngineError::Capability(e.to_string()))?; + + // `HttpRequestTool::execute` always returns `Ok`, using `is_error` to + // signal a failed request (non-2xx, DNS/allowlist rejection, timeout, + // …) — surface that as a capability error so the engine's + // `on_error`/`retry` policy can act on it. + if result.is_error { + return Err(EngineError::Capability(result.text())); + } + + Ok(json!({ "text": result.text() })) + } +} + +/// [`CodeRunner`] adapter running sandboxed user code via +/// `src/openhuman/sandbox/ops.rs` (`resolve_sandbox_policy` + +/// `execute_in_sandbox`), modeled on +/// `src/openhuman/tools/impl/system/node_exec.rs::run_sandboxed`. +/// +/// **Mismatch handled here:** the sandbox runs a shell command string, not a +/// `(language, source, input)` triple. `source` is treated as a function body +/// receiving the serialized `input` items array and returning the node's +/// output — this convention is a B1 design choice (not specified by the +/// crate), matching the mock's "function body" tests +/// (`tinyflows::nodes::integration::code` — e.g. `"source": "return 1;"`). +/// +/// Requires `node`/`python3` on the `PATH` the sandbox backend runs under; +/// there is no managed toolchain wiring here (unlike `node_exec`'s +/// `NodeBootstrap`). +pub struct OpenHumanCode { + pub config: Arc, +} + +const CODE_RUN_TIMEOUT_SECS: u64 = 60; + +#[async_trait] +impl CodeRunner for OpenHumanCode { + async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result { + let policy = resolve_sandbox_policy( + SandboxMode::Sandboxed, + &self.config.action_dir, + &self.config.runtime, + false, + ); + + // Work dir lives under `action_dir` (the sandbox workspace root). We keep + // its path *relative* to `action_dir` so the run command works on every + // backend: for Local, `execute_in_sandbox`'s `working_dir` is the host + // cwd; for Docker, `action_dir` is bind-mounted at `/workspace` with + // `-w /workspace`. Host-absolute paths would not exist inside the + // container, so we pass `action_dir` as the working dir and reference the + // script/input by their `action_dir`-relative paths. + let rel_dir = std::path::Path::new(".flows_code").join(uuid::Uuid::new_v4().to_string()); + let work_dir = self.config.action_dir.join(&rel_dir); + tokio::fs::create_dir_all(&work_dir) + .await + .map_err(|e| EngineError::Capability(format!("failed to create code work dir: {e}")))?; + + let (script_name, interpreter, script_body) = match language { + CodeLanguage::JavaScript => ("script.js", "node", js_harness(source)), + CodeLanguage::Python => ("script.py", "python3", python_harness(source)), + }; + let script_path = work_dir.join(script_name); + let input_path = work_dir.join("input.json"); + + let input_json = serde_json::to_string(&input) + .map_err(|e| EngineError::Capability(format!("failed to serialize code input: {e}")))?; + tokio::fs::write(&script_path, script_body) + .await + .map_err(|e| EngineError::Capability(format!("failed to write code script: {e}")))?; + tokio::fs::write(&input_path, input_json) + .await + .map_err(|e| EngineError::Capability(format!("failed to write code input: {e}")))?; + + // Backend-agnostic, `action_dir`-relative command paths (see above). + let rel_script = rel_dir.join(script_name); + let rel_input = rel_dir.join("input.json"); + let command = format!( + "{} {} {}", + shell_quote(interpreter), + shell_quote(&rel_script.to_string_lossy()), + shell_quote(&rel_input.to_string_lossy()), + ); + + let mut extra_env = std::collections::HashMap::new(); + if let Ok(host_path) = std::env::var("PATH") { + extra_env.insert("PATH".to_string(), host_path); + } + + tracing::debug!( + target: "flows", + ?language, + work_dir = %work_dir.display(), + "[flows] code: running sandboxed script" + ); + + let exec_result = execute_in_sandbox( + &policy, + &command, + &self.config.action_dir, + extra_env, + std::time::Duration::from_secs(CODE_RUN_TIMEOUT_SECS), + ) + .await; + + // Always clean up the work dir — even when `execute_in_sandbox` itself + // errors (e.g. a spawn failure) — so temp scripts never leak. + if let Err(e) = tokio::fs::remove_dir_all(&work_dir).await { + tracing::debug!(target: "flows", error = %e, "[flows] code: failed to clean up work dir (non-fatal)"); + } + + let result = exec_result + .map_err(|e| EngineError::Capability(format!("sandbox execution failed: {e}")))?; + + if !result.success() { + return Err(EngineError::Capability(format!( + "code node exited non-zero (timed_out={}): {}", + result.timed_out, result.stderr + ))); + } + + serde_json::from_str(result.stdout.trim()) + .map_err(|e| EngineError::Capability(format!("code output was not valid JSON: {e}"))) + } +} + +/// Wraps user `source` as a function body receiving `input`, executed by Node, +/// printing the JSON result (or `null`) to stdout. +fn js_harness(source: &str) -> String { + format!( + "const fs = require('fs');\n\ + const input = JSON.parse(fs.readFileSync(process.argv[2], 'utf8'));\n\ + const __result__ = (function(input) {{\n{source}\n}})(input);\n\ + process.stdout.write(JSON.stringify(__result__ === undefined ? null : __result__));\n" + ) +} + +/// Wraps user `source` as a function body receiving `input`, executed by +/// Python, printing the JSON result (or `null`) to stdout. +fn python_harness(source: &str) -> String { + let indented: String = if source.trim().is_empty() { + " pass".to_string() + } else { + source + .lines() + .map(|line| format!(" {line}")) + .collect::>() + .join("\n") + }; + format!( + "import sys, json\n\ + with open(sys.argv[1]) as __f__:\n input = json.load(__f__)\n\ + def __user_fn__(input):\n{indented}\n return None\n\ + __result__ = __user_fn__(input)\n\ + print(json.dumps(__result__))\n" + ) +} + +/// POSIX single-quote shell escaping, mirroring +/// `tools/impl/system/node_exec.rs::shell_quote`. +fn shell_quote(s: &str) -> String { + let escaped = s.replace('\'', "'\\''"); + format!("'{escaped}'") +} + +/// [`StateStore`] adapter over the `flows::` domain's `flow_state` KV table. +pub struct FlowStateStore { + pub config: Arc, + pub namespace: String, +} + +#[async_trait] +impl StateStore for FlowStateStore { + async fn load(&self, key: &str) -> Result> { + flows::kv_get(&self.config, &self.namespace, key) + .map_err(|e| EngineError::Capability(e.to_string())) + } + + async fn store(&self, key: &str, value: Value) -> Result<()> { + flows::kv_set(&self.config, &self.namespace, key, &value) + .map_err(|e| EngineError::Capability(e.to_string())) + } +} + +/// Builds the [`Capabilities`] bundle for one run, wiring each of the five +/// host-injected traits to a real OpenHuman adapter (see each adapter above for +/// its contract). +/// +/// `state_namespace` scopes the [`FlowStateStore`] KV so two saved flows that +/// use the same state key never read or overwrite each other — callers pass a +/// per-flow namespace (e.g. `"flow:"`). +pub fn build_capabilities(config: Arc, state_namespace: impl Into) -> Capabilities { + let security = Arc::new(SecurityPolicy::from_config( + &config.autonomy, + &config.workspace_dir, + &config.action_dir, + )); + let http_config = config.http_request.clone(); + + Capabilities { + llm: Arc::new(OpenHumanLlm { + config: config.clone(), + }), + tools: Arc::new(OpenHumanTools { + config: config.clone(), + }), + http: Arc::new(OpenHumanHttp { + security, + http_config, + }), + code: Arc::new(OpenHumanCode { + config: config.clone(), + }), + state: Arc::new(FlowStateStore { + config, + namespace: state_namespace.into(), + }), + } +} + +/// Opens the durable, cross-process checkpointer a `flows_run` uses via +/// `tinyflows::engine::run_with_checkpointer` — the crate's own +/// `tinyagents::graph::SqliteCheckpointer`, stored under +/// `/flows/checkpoints.db`. +/// +/// Deliberately **not** a bespoke checkpointer: the crate ships its own +/// SQLite-backed `Checkpointer` impl (feature `sqlite`, already enabled +/// on the `tinyagents` dependency), so the seam just opens it — mirrors the +/// construction in `src/openhuman/agent_orchestration/delegation.rs`. +pub fn open_flow_checkpointer( + config: &Config, +) -> anyhow::Result>> { + let db_path = config.workspace_dir.join("flows").join("checkpoints.db"); + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent) + .with_context(|| format!("Failed to create flows directory: {}", parent.display()))?; + } + tracing::debug!(target: "flows", db = %db_path.display(), "[flows] opening checkpointer"); + Ok(Arc::new( + SqliteCheckpointer::::open(&db_path) + .with_context(|| format!("Failed to open flows checkpointer: {}", db_path.display()))?, + )) +} diff --git a/src/openhuman/tinyflows/mod.rs b/src/openhuman/tinyflows/mod.rs new file mode 100644 index 000000000..8d3b76bd5 --- /dev/null +++ b/src/openhuman/tinyflows/mod.rs @@ -0,0 +1,16 @@ +//! The `tinyflows` capability seam: wires the `tinyflows` workflow engine +//! (an external, host-agnostic crate — validate → compile → run on +//! `tinyagents`) to real OpenHuman services. +//! +//! 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. + +pub mod caps; +pub mod observability; +#[cfg(test)] +mod tests; + +pub use caps::{build_capabilities, open_flow_checkpointer}; diff --git a/src/openhuman/tinyflows/observability.rs b/src/openhuman/tinyflows/observability.rs new file mode 100644 index 000000000..8d97229b7 --- /dev/null +++ b/src/openhuman/tinyflows/observability.rs @@ -0,0 +1,70 @@ +//! Log-only [`tinyflows::observability::RunObserver`] for the `flows::` +//! domain. +//! +//! `tinyflows` emits structured run/step records; the host decides what to do +//! with them. B1 only logs — persisting steps/runs for a run-history view is +//! B2+ (see `my_docs/ohxtf/b1-engine-seam-domain/07-execution-and-hitl.md`). +//! Note the durable path (`engine::run_with_checkpointer`, what `flows_run` +//! uses) installs a `NoopObserver` internally in 0.2, so this observer is +//! wired for `run_with_observer` call sites, not the durable run path itself. + +use tinyflows::observability::{ExecutionStep, Run, RunObserver}; + +/// Logs run/step lifecycle events with grep-friendly `[flows]` prefixes. +pub struct TracingRunObserver { + pub run_label: String, +} + +impl RunObserver for TracingRunObserver { + fn on_run_start(&self, run_id: &str) { + tracing::info!(target: "flows", run_label = %self.run_label, %run_id, "[flows] run start"); + } + + fn on_step_finish(&self, step: &ExecutionStep) { + tracing::debug!( + target: "flows", + run_label = %self.run_label, + node = %step.node_id, + status = ?step.status, + duration_ms = step.duration_ms, + "[flows] step finished" + ); + } + + fn on_run_finish(&self, run: &Run) { + tracing::info!( + target: "flows", + run_label = %self.run_label, + id = %run.id, + status = ?run.status, + steps = run.steps.len(), + "[flows] run finish" + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::Value; + use tinyflows::observability::{RunStatus, StepStatus}; + + #[test] + fn callbacks_do_not_panic() { + let observer = TracingRunObserver { + run_label: "test".to_string(), + }; + observer.on_run_start("run-1"); + observer.on_step_finish(&ExecutionStep { + node_id: "n".to_string(), + status: StepStatus::Success, + output: Value::Null, + duration_ms: 5, + }); + observer.on_run_finish(&Run { + id: "run-1".to_string(), + status: RunStatus::Completed, + steps: Vec::new(), + }); + } +} diff --git a/src/openhuman/tinyflows/tests.rs b/src/openhuman/tinyflows/tests.rs new file mode 100644 index 000000000..4c4d2a5b2 --- /dev/null +++ b/src/openhuman/tinyflows/tests.rs @@ -0,0 +1,214 @@ +//! Seam tests for `src/openhuman/tinyflows/`. +//! +//! **Deviation from the original test plan** (see +//! `my_docs/ohxtf/b1-engine-seam-domain/09-testing-and-verification.md` item 2 +//! and commons/11): the plan called for pointing `HttpRequestTool` at a local +//! mock HTTP server and asserting a success round-trip. That is not possible +//! against the REAL `HttpRequestTool` — unlike `tinyflows`' own mock +//! `HttpClient`, OpenHuman's `url_guard` unconditionally blocks +//! loopback/private hosts as an SSRF guard (`is_private_or_local_host`), +//! before the allowlist is even consulted, and any locally-hosted mock server +//! is necessarily loopback. So instead: +//! - the HTTP adapter tests assert the SSRF guard and the strict-allowlist +//! rejection both surface as `EngineError::Capability` (proving the adapter +//! correctly propagates `HttpRequestTool`'s real security behavior), and +//! - the engine smoke test drives `trigger -> http_request` against a +//! deterministically-blocked loopback URL with `on_error: continue`, which +//! exercises the full real stack (build_capabilities -> engine -> compiled +//! graph -> `OpenHumanHttp` -> real `HttpRequestTool` -> SSRF guard -> +//! `EngineError::Capability` -> the crate's `on_error: continue` policy -> +//! error item) without any network dependency. + +use std::sync::Arc; + +use serde_json::json; +use tempfile::TempDir; +use tinyflows::caps::{CodeLanguage, CodeRunner, HttpClient, StateStore}; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; + +use crate::openhuman::config::Config; +use crate::openhuman::security::SecurityPolicy; + +use super::build_capabilities; +use super::caps::{FlowStateStore, OpenHumanCode, OpenHumanHttp}; + +fn test_config(tmp: &TempDir) -> Arc { + let config = Config { + workspace_dir: tmp.path().join("workspace"), + action_dir: tmp.path().join("workspace"), + config_path: tmp.path().join("config.toml"), + ..Config::default() + }; + std::fs::create_dir_all(&config.workspace_dir).unwrap(); + Arc::new(config) +} + +fn node(id: &str, kind: NodeKind, config: serde_json::Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: Vec::new(), + position: None, + } +} + +fn edge(from: &str, to: &str) -> Edge { + Edge { + from_node: from.to_string(), + from_port: "main".to_string(), + to_node: to.to_string(), + to_port: "main".to_string(), + } +} + +// ── build_capabilities smoke ──────────────────────────────────────────── + +#[test] +fn build_capabilities_constructs_every_slot_without_panicking() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + // Purely a construction smoke test — no capability is invoked here. + let _caps = build_capabilities(config, "test:build"); +} + +// ── HTTP adapter ───────────────────────────────────────────────────────── + +fn http_adapter(allowed_domains: Vec) -> OpenHumanHttp { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let security = Arc::new(SecurityPolicy::from_config( + &config.autonomy, + &config.workspace_dir, + &config.action_dir, + )); + OpenHumanHttp { + security, + http_config: crate::openhuman::config::HttpRequestConfig { + allowed_domains, + ..Default::default() + }, + } +} + +#[tokio::test] +async fn http_adapter_blocks_loopback_host_as_capability_error() { + let adapter = http_adapter(vec![]); // open allowlist mode + let err = adapter + .request( + json!({ "method": "GET", "url": "http://127.0.0.1:1/" }), + None, + ) + .await + .expect_err("loopback host must be blocked by the SSRF guard"); + let msg = err.to_string(); + assert!( + msg.to_lowercase().contains("private") || msg.to_lowercase().contains("local"), + "expected an SSRF-guard message, got: {msg}" + ); +} + +#[tokio::test] +async fn http_adapter_rejects_host_outside_strict_allowlist() { + let adapter = http_adapter(vec!["example.com".to_string()]); + let err = adapter + .request( + json!({ "method": "GET", "url": "https://not-allowed.test/" }), + None, + ) + .await + .expect_err("host outside the strict allowlist must be rejected"); + assert!( + err.to_string().contains("not-allowed.test") + || err.to_string().to_lowercase().contains("allowed"), + "expected an allowlist rejection message, got: {err}" + ); +} + +// ── StateStore adapter ─────────────────────────────────────────────────── + +#[tokio::test] +async fn flow_state_store_round_trips_and_is_namespace_scoped() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + let ns1 = FlowStateStore { + config: config.clone(), + namespace: "ns1".to_string(), + }; + let ns2 = FlowStateStore { + config: config.clone(), + namespace: "ns2".to_string(), + }; + + assert!(ns1.load("k").await.unwrap().is_none()); + + ns1.store("k", json!({ "v": 1 })).await.unwrap(); + assert_eq!(ns1.load("k").await.unwrap(), Some(json!({ "v": 1 }))); + + // A different namespace never sees ns1's value. + assert!(ns2.load("k").await.unwrap().is_none()); + + // Overwrite. + ns1.store("k", json!(2)).await.unwrap(); + assert_eq!(ns1.load("k").await.unwrap(), Some(json!(2))); +} + +// ── Engine smoke: real seam end to end ─────────────────────────────────── + +#[tokio::test] +async fn engine_run_drives_trigger_to_http_request_through_the_real_seam() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let caps = build_capabilities(config, "test:smoke"); + + // A deterministically-blocked loopback URL with `on_error: continue` so + // the run completes even though the (real, SSRF-guarded) HTTP adapter + // necessarily rejects it — see the module doc for why a real network + // round-trip isn't testable here. + let graph = WorkflowGraph { + nodes: vec![ + node("t", NodeKind::Trigger, serde_json::Value::Null), + node( + "http", + NodeKind::HttpRequest, + json!({ "method": "GET", "url": "http://127.0.0.1:1/", "on_error": "continue" }), + ), + ], + edges: vec![edge("t", "http")], + ..Default::default() + }; + let compiled = tinyflows::compiler::compile(&graph).expect("compile"); + + let outcome = tinyflows::engine::run(&compiled, json!({ "seed": 1 }), &caps) + .await + .expect("run should complete (on_error: continue)"); + + assert!(outcome.pending_approvals.is_empty()); + assert_eq!( + outcome.output["nodes"]["http"]["items"][0]["json"]["error"]["node"], + json!("http") + ); +} + +// ── Code adapter ────────────────────────────────────────────────────────── + +/// Requires `node` on `PATH`. Ignored by default (per the B1 test plan); +/// run explicitly with `cargo test -- --ignored` on a host with Node +/// installed. +#[tokio::test] +#[ignore = "requires a `node` binary on PATH"] +async fn code_adapter_javascript_passthrough_round_trips_json() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let runner = OpenHumanCode { config }; + + let input = json!([{ "json": { "n": 7 } }]); + let result = runner + .run(CodeLanguage::JavaScript, "return input;", input.clone()) + .await + .expect("javascript passthrough should succeed when node is present"); + assert_eq!(result, input); +}