TinyFlows

A Rust-native, host-agnostic workflow automation engine, shipped as a library
crate.

tinyflows models an automation as a WorkflowGraph — a directed graph of typed
nodes — that is validated, compiled, and lowered per run onto the
tinyagents state-graph engine, then
driven to completion by engine::run. It is deliberately host-agnostic:
everything that touches the outside world — LLMs, integration tools, HTTP, code
execution, persistence — goes through capability traits the embedding
application implements, so the crate never hard-codes a vendor.

Rust 2024 · MSRV 1.85 · #![forbid(unsafe_code)] · GPL-3.0-or-later.

Features

Engine

• Typed workflow model (WorkflowGraph of Nodes and Edges) with JSON as the
  wire format, structural validation, and per-run compilation onto tinyagents.
• Item-based data flow: a connection carries an array of items
  ({ json, binary?, paired_item? }); nodes map their logic over input items.
• =-prefixed config expressions (e.g. "=item.name") resolved against the run
  scope.
• Linear execution, conditional routing on output ports, parallel fan-out
  (concurrent successors sharing a port), and a merge fan-in barrier (a node
  runs only once all its predecessors finish).

Nodes

• Full node catalog implemented and tested — control-flow (condition,
  switch, merge, split_out, transform) and capability-backed (agent,
  tool_call, http_request, code, output_parser, sub_workflow), plus the
  trigger entry node.

Reliability

• Per-node error handling: on_error policy (stop / continue / route),
  bounded retry, and an error output port for routing failures to a recovery
  sub-graph.
• Human-in-the-loop approval gating: a node with requires_approval pauses the
  run and is surfaced via RunOutcome::pending_approvals; engine::resume
  approves and continues. A host can also drive durable, cross-process resume by
  injecting a Checkpointer via engine::run_with_checkpointer /
  resume_with_checkpointer.
• Observability via tracing plus a RunObserver hook and Run /
  ExecutionStep records.

Extensibility

• Host-injected capability traits: LlmProvider, ToolInvoker, HttpClient,
  CodeRunner, and StateStore. Deterministic in-memory mocks ship behind the
  mock cargo feature (caps::mock::mock_capabilities()).
• Opaque connection_ref credential references — the host resolves them to real
  secrets; the crate never sees them.
• Versioned wire format: graph schema_version and per-node type_version, with
  a migrate framework for load-time upgrades.

How it works

model::WorkflowGraph  ->  validate  ->  compiler::compile  ->  engine::run
   (typed graph)        (structural)     (validated handle)     (lowers onto
                                                                 tinyagents,
                                                                 drives to done)

compile validates the graph and returns an opaque CompiledWorkflow; the graph
is lowered onto a fresh tinyagents state graph once per run, inside
engine::run, which captures that run's capabilities in each node handler. Run
state is a single JSON value shaped as
{ "run": { "trigger": … }, "nodes": { "": { "items": [ … ] } } }: a merge
reducer folds each node's item output under its own id, so independent nodes
never collide (which keeps parallel fan-out deterministic). Every outside-world
effect is reached through the Capabilities traits the host supplies for the
run.

Quickstart

Add the crate:

[dependencies]
tinyflows = "0.1"

Build a trigger -> transform graph, compile it, and run it against the mock
capabilities. The mock feature provides the in-memory capability impls used by
tests and examples:

use serde_json::{Value, json};
use tinyflows::caps::mock::mock_capabilities;
use tinyflows::compiler::compile;
use tinyflows::engine::run;
use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let graph = WorkflowGraph {
        nodes: vec![
            Node {
                id: "t".into(),
                kind: NodeKind::Trigger,
                type_version: 1,
                name: "start".into(),
                config: Value::Null,
                ports: vec![],
                position: None,
            },
            Node {
                id: "greet".into(),
                kind: NodeKind::Transform,
                type_version: 1,
                name: "greet".into(),
                config: json!({ "set": { "greeting": "=item.name" } }),
                ports: vec![],
                position: None,
            },
        ],
        edges: vec![Edge {
            from_node: "t".into(),
            from_port: "main".into(),
            to_node: "greet".into(),
            to_port: "main".into(),
        }],
        ..Default::default()
    };

    let compiled = compile(&graph).expect("compile");
    let outcome = run(&compiled, json!({ "name": "Ada" }), &mock_capabilities())
        .await
        .expect("run");
    println!("{}", serde_json::to_string_pretty(&outcome.output).unwrap());
}

This is the hello_workflow example — run it with:

cargo run --example hello_workflow --features mock

Examples

The crate ships seven runnable examples under examples/. Each is
gated on the mock cargo feature, so run them with:

cargo run --example  --features mock

| Example | What it shows |
| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------- | ------ |
| hello_workflow | Build → compile → run a trigger → transform workflow against the mock capabilities. |
| conditional_branch | IF routing: a condition node takes exactly one of its true / false branches. |
| parallel_and_merge | Parallel fan-out (a node's same-port successors run concurrently) joined by a merge fan-in barrier. |
| capability_pipeline | A linear http_request → code → agent → tool_call pipeline through the host capability traits (mocked). |
| error_handling | Per-node retry plus on_error: "route" recovering a failing node via its error port. |
| hitl_approval | A requires_approval gate pauses the run (pending_approvals), then run_resumable(...).resume(...) continues from the checkpoint. |
| jq_expressions | The jaq-backed jq engine in a transform node (e.g. =.item.prices | add). |

Omitting --features mock is harmless: the demo body is
#[cfg(feature = "mock")]-gated, so a default build stays green and the example
just prints a hint to re-run with the feature enabled.

Run all of them in one go:

for ex in hello_workflow conditional_branch parallel_and_merge \
          capability_pipeline error_handling hitl_approval jq_expressions; do
  cargo run --example "$ex" --features mock
done

Node catalog

See the Node Catalog wiki page for config keys and
ports.

Status

The Phase-A engine is complete: model, validation, per-run compilation and
lowering onto tinyagents, the full node catalog, item-based data flow with
=-expressions, linear / conditional / parallel-fan-out / merge-barrier routing,
per-node error handling (on_error / retry / error port), human-in-the-loop
approval gating (pending_approvals + resume), tracing + RunObserver
observability, opaque connection_ref credentials, and schema_version /
type_version migration. The runtime runs end-to-end against the mock
capabilities, guarded by a reference-workflow e2e suite, and
cargo publish --dry-run is clean.

Also implemented:

• A full jq/jaq expression engine. Every =-prefixed string is a jq program
  compiled and executed by jaq with the
  evaluation scope as its input (src/expr.rs); a bare =.item.name dotted path
  is served by a fast structural walk, while anything richer (filters, pipes,
  add, …) routes to jaq.
• Retry backoff timing and per-node timeouts. A node's retry config takes
  backoff_ms plus backoff: "fixed" | "exponential" (capped at 60 s between
  attempts), and a run-level node_timeout_secs on the trigger applies a
  per-node timeout across the whole run.
• Sub-workflows by reference. A sub_workflow node runs a child either from
  an inline workflow graph or from a host-managed workflow_id, resolved
  through the injected WorkflowResolver capability. Nesting is depth-bounded
  and direct self-references are rejected.

Not yet:

• Automatic checkpointed super-step replay. Durable, cross-process resume is
  already supported by injecting a Checkpointer
  (engine::run_with_checkpointer / resume_with_checkpointer); only the
  super-step replay optimization that skips re-executing completed nodes on the
  in-process resume path remains.
• Visual and agent-first authoring (host-side).
• The OpenHuman host integration (Phase B, a separate repo).
• Publishing to crates.io.

Building & testing

Install Rust 1.85 or newer with rustup, then:

cargo build
cargo test                 # unit + compiler + engine tests (mocks auto-available)
cargo test --all-features  # also exercises the `mock` capability impls explicitly

The crate is #![forbid(unsafe_code)] and fully documented
(#![warn(missing_docs)]). The CI gate is:

cargo fmt --all -- --check
cargo clippy --all-targets --all-features -- -D warnings
cargo test --all-features

Documentation

The design and implementation guides live in the project
wiki — start with
Getting Started, then
Architecture and the
Node Catalog.

Contributing

Contributions are welcome. Start with CONTRIBUTING.md. In short:

1. Keep changes focused and easy to review.
2. Run the CI checks locally: cargo fmt --all -- --check,
   cargo clippy --all-targets --all-features -- -D warnings, and
   cargo test --all-features.
3. Include tests or documentation when behavior changes.
4. Follow the host-agnostic, no-unsafe, fully-documented conventions.

License

tinyflows is licensed under the GNU General Public License, version 3 or later.
See LICENSE for the full license text.