mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(flows): tinyflows Workflows (B1) — engine seam + flows domain (#4433)
This commit is contained in:
Cyrus Gray
Generated
+138
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -119,6 +119,8 @@ fn build_registered_controllers() -> Vec<RegisteredController> {
|
||||
.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<ControllerSchema> {
|
||||
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.",
|
||||
),
|
||||
|
||||
@@ -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),
|
||||
},
|
||||
];
|
||||
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<WorkflowGraph, String> {
|
||||
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<RpcOutcome<Flow>, 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<RpcOutcome<Flow>, 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<RpcOutcome<Vec<Flow>>, 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<String>,
|
||||
graph_json: Option<Value>,
|
||||
) -> Result<RpcOutcome<Flow>, 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<RpcOutcome<Value>, 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<RpcOutcome<Flow>, 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<RpcOutcome<Value>, 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;
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
@@ -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<ControllerSchema> {
|
||||
vec![
|
||||
schemas("create"),
|
||||
schemas("get"),
|
||||
schemas("list"),
|
||||
schemas("update"),
|
||||
schemas("delete"),
|
||||
schemas("set_enabled"),
|
||||
schemas("run"),
|
||||
]
|
||||
}
|
||||
|
||||
pub fn all_registered_controllers() -> Vec<RegisteredController> {
|
||||
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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let name = read_required::<String>(¶ms, "name")?;
|
||||
let graph = read_required::<Value>(¶ms, "graph")?;
|
||||
to_json(ops::flows_create(&config, name, graph).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_get(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let id = read_required::<String>(¶ms, "id")?;
|
||||
to_json(ops::flows_get(&config, id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_list(_params: Map<String, Value>) -> 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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let id = read_required::<String>(¶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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let id = read_required::<String>(¶ms, "id")?;
|
||||
to_json(ops::flows_delete(&config, id.trim()).await?)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_set_enabled(params: Map<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let id = read_required::<String>(¶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<String, Value>) -> ControllerFuture {
|
||||
Box::pin(async move {
|
||||
let config = config_rpc::load_config_with_timeout().await?;
|
||||
let id = read_required::<String>(¶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<T: DeserializeOwned>(params: &Map<String, Value>, key: &str) -> Result<T, String> {
|
||||
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<T: serde::Serialize>(outcome: RpcOutcome<T>) -> Result<Value, String> {
|
||||
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::<String>(¶ms, "id").unwrap_err();
|
||||
assert!(err.contains("missing required param 'id'"));
|
||||
}
|
||||
}
|
||||
@@ -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<T>`.
|
||||
//!
|
||||
//! 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<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
|
||||
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<Flow> {
|
||||
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<Option<Flow>> {
|
||||
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<Vec<Flow>> {
|
||||
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<Flow> {
|
||||
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<Flow> {
|
||||
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<Flow> {
|
||||
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<E: std::error::Error + Send + Sync + 'static>(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<Option<serde_json::Value>> {
|
||||
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;
|
||||
@@ -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))
|
||||
);
|
||||
}
|
||||
@@ -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<String>,
|
||||
/// Outcome of the most recent run: `"completed"` | `"pending_approval"` | `"failed"`.
|
||||
pub last_status: Option<String>,
|
||||
}
|
||||
|
||||
#[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());
|
||||
}
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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::<OrchestrationState>::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::<OrchestrationState>::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,
|
||||
|
||||
@@ -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::<OrchestrationState>::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::<OrchestrationState>::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");
|
||||
}
|
||||
|
||||
@@ -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<UsageInfo>) -> 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<Config>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LlmProvider for OpenHumanLlm {
|
||||
async fn complete(&self, request: Value, conn: Option<&str>) -> Result<Value> {
|
||||
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<ChatMessage> = 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<Config>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ToolInvoker for OpenHumanTools {
|
||||
async fn invoke(&self, slug: &str, args: Value, conn: Option<&str>) -> Result<Value> {
|
||||
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<SecurityPolicy>,
|
||||
pub http_config: HttpRequestConfig,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl HttpClient for OpenHumanHttp {
|
||||
async fn request(&self, request: Value, conn: Option<&str>) -> Result<Value> {
|
||||
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<Config>,
|
||||
}
|
||||
|
||||
const CODE_RUN_TIMEOUT_SECS: u64 = 60;
|
||||
|
||||
#[async_trait]
|
||||
impl CodeRunner for OpenHumanCode {
|
||||
async fn run(&self, language: CodeLanguage, source: &str, input: Value) -> Result<Value> {
|
||||
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::<Vec<_>>()
|
||||
.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<Config>,
|
||||
pub namespace: String,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl StateStore for FlowStateStore {
|
||||
async fn load(&self, key: &str) -> Result<Option<Value>> {
|
||||
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:<id>"`).
|
||||
pub fn build_capabilities(config: Arc<Config>, state_namespace: impl Into<String>) -> 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
|
||||
/// `<workspace_dir>/flows/checkpoints.db`.
|
||||
///
|
||||
/// Deliberately **not** a bespoke checkpointer: the crate ships its own
|
||||
/// SQLite-backed `Checkpointer<State>` 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<Arc<dyn tinyflows::engine::Checkpointer<serde_json::Value>>> {
|
||||
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::<serde_json::Value>::open(&db_path)
|
||||
.with_context(|| format!("Failed to open flows checkpointer: {}", db_path.display()))?,
|
||||
))
|
||||
}
|
||||
@@ -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};
|
||||
@@ -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(),
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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<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();
|
||||
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<String>) -> 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);
|
||||
}
|
||||
Reference in New Issue
Block a user