The Reflow Runtime

reflow_rt is the public Rust API for Reflow.

Reflow is a modular workflow execution engine that uses the actor model for concurrent, message-passing computation. It enables graph-authored DAGs for data processing, real-time media, visual tooling, distributed workflows, and optional ML/CV taskpacks.

Use this crate when you are building applications, tools, services, examples, or taskpacks on top of Reflow. It is the stable runtime surface for authoring graph workflows, registering actors, running networks, streaming media, and opting into larger component families such as GPU rendering, native camera capture, API-service actors, and ML/CV pipelines.

The lower-level crates still exist because Reflow is modular internally. Application code should start here.

[dependencies]
reflow_rt = "0.1"
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

Runtime Model

Reflow runs actor graphs.

The same topology can be authored by code, generated by tools, or loaded from graph exports. Reflow does not require hardcoded pipelines in framework code; graph structure remains the source of truth.

Quick Start — Build And Run A Network

Register a template, add a node, wire it up, start.

use reflow_rt::prelude::*;
use serde_json::json;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let mut net = Network::new(NetworkConfig::default());

    // Register the component templates this graph uses.
    for tpl in ["tpl_interval_trigger", "tpl_passthrough"] {
        net.register_actor_arc(tpl, get_actor_for_template(tpl).unwrap())?;
    }

    // Create nodes with per-node config.
    let cfg = |v: serde_json::Value| v.as_object()
        .map(|m| m.clone().into_iter().collect::<HashMap<_, _>>());

    net.add_node("tick", "tpl_interval_trigger",
        cfg(json!({ "interval": 100, "maxExecutions": 10 })))?;
    net.add_node("echo", "tpl_passthrough", None)?;

    // Wire tick.trigger -> echo.in.
    net.add_connection(Connector {
        from: ConnectionPoint { actor: "tick".into(), port: "trigger".into(), ..Default::default() },
        to:   ConnectionPoint { actor: "echo".into(), port: "in".into(),      ..Default::default() },
    });

    // Kick the first tick with an initial packet.
    net.add_initial(InitialPacket {
        to: ConnectionPoint::new("tick", "start", Some(Message::Flow)),
    });

    net.start()?;
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
    net.shutdown();
    Ok(())
}

Connector, ConnectionPoint, and InitialPacket come from reflow_rt::network::connector. The prelude re-exports the most common authoring types.

Observing Runtime Events

Subscribe before calling net.start() so no events are missed.

use reflow_rt::network::network::NetworkEvent;

let events = net.get_event_receiver();
tokio::spawn(async move {
    while let Ok(evt) = events.recv_async().await {
        match evt {
            NetworkEvent::ActorStarted { actor_id, .. } => println!("start {actor_id}"),
            NetworkEvent::ActorFailed  { actor_id, error, .. } => eprintln!("fail {actor_id}: {error}"),
            _ => {}
        }
    }
});

Loading A Graph Export

Editors and tools persist flows as GraphExport JSON. Load once, hand to the network.

use reflow_rt::graph::{Graph, types::GraphExport};
use reflow_rt::network::network::{Network, NetworkConfig};

let json = std::fs::read_to_string("my_flow.json")?;
let export: GraphExport = serde_json::from_str(&json)?;
let graph = Graph::load(export, None);

let net = Network::with_graph(NetworkConfig::default(), &graph);
net.lock().start()?;

Templates referenced by nodes must already be registered, either individually or through the bundled component catalog.

Writing A Custom Actor

Use the #[actor(...)] macro from the prelude. Declare in-ports, out-ports, and state; the macro generates the actor struct and wiring.

use reflow_rt::prelude::*;
use anyhow::Error;
use std::collections::HashMap;

#[actor(
    DoubleActor,
    inports::<50>(number),
    outports::<50>(doubled),
    state(MemoryState)
)]
pub async fn double_actor(ctx: ActorContext) -> Result<HashMap<String, Message>, Error> {
    let inputs = ctx.get_payload();
    let n = inputs.get("number").and_then(|m| m.as_f64()).unwrap_or(0.0);
    Ok(HashMap::from([("doubled".into(), Message::Float(n * 2.0))]))
}

// Register with a stable template id.
net.register_actor_arc("tpl_double", std::sync::Arc::new(DoubleActor::default()))?;
net.add_node("x2", "tpl_double", None)?;

ActorContext exposes the current payload, config, shared state, and StreamHandle access. Returning a HashMap means "emit these packets on these ports"; returning an empty map means "consumed, no output this tick".

Subgraphs

Reuse authored flows as first-class actors. A SubgraphActor embeds a GraphExport and participates in the parent network like any other actor.

use reflow_rt::prelude::*;

let inner: GraphExport = serde_json::from_str(include_str!("material.json"))?;
let subgraph = SubgraphActor::new(inner, /* exposed_in */ vec!["uv"], /* exposed_out */ vec!["color"]);

net.register_actor_arc("tpl_material_subgraph", std::sync::Arc::new(subgraph))?;
net.add_node("material", "tpl_material_subgraph", None)?;

Expose only the ports the parent DAG needs to touch; internal wiring stays private to the subgraph.

Streams And Media

Large or continuous payloads travel as Message::StreamHandle. Consumers take a typed receiver and iterate until the stream ends.

use reflow_rt::actor_runtime::stream::{StreamFrame, StreamHandle};

if let Some(Message::StreamHandle(handle)) = ctx.get_payload().get("input").cloned() {
    let rx = handle.take_receiver().ok_or_else(|| anyhow::anyhow!("no receiver"))?;
    while let Ok(frame) = rx.recv_async().await {
        match frame {
            StreamFrame::Bytes(bytes) => { /* push to decoder */ }
            StreamFrame::End          => break,
            StreamFrame::Error(e)     => return Err(anyhow::anyhow!(e)),
            _ => {}
        }
    }
}

This is how tpl_render_frame_collector → tpl_video_encoder → tpl_file_save moves 4K RGBA frames without copying through the message bus.

First Import

use reflow_rt::prelude::*;

fn main() {
    let mut graph = Graph::new("example", false, None);
    graph.add_node("tap", "tpl_passthrough", None);

    let network = Network::with_graph(NetworkConfig::default(), &graph);
    let _ = network;
}

The prelude is intentionally small: graph, network, actor, message, stream, subgraph, and template types that most runtime users need first.

Public Modules

reflow_rt exposes the runtime as named modules. The exact crate names are re-exported for clarity, and short aliases are provided for application ergonomics.

Feature-gated runtime modules include api, dsp, media_types, media_codec, asset_registry, litert, cv_ops, ml_ops, and taskpacks.

Feature Policy

The default runtime is deliberately lean. It gives you the graph runtime and core utility components without pulling in GPU, AV, native camera, API-service generation, browser automation, or native ML runtime dependencies.

# The default: runtime + core utility component catalog.
reflow_rt = "0.1"

# Minimal import surface, if you only want the public modules.
reflow_rt = { version = "0.1", default-features = false }

# Opt into exactly the stacks your app needs.
reflow_rt = { version = "0.1", features = ["gpu", "camera-native", "media", "ml"] }

Real LiteRT execution is always opt-in. The ML stack remains deterministic and mockable unless external-litert is enabled and model manifests request the LiteRT backend.

Editor And Template Runtime

Reflow components carry template metadata for visual editors. The runtime surface exposes:

use reflow_rt::components::{get_actor_for_template, get_template_mapping, template_catalog};
use reflow_rt::network::template::TemplateRegistry;

This keeps authored flows portable: editors can discover templates, users wire DAGs, and the runtime resolves those templates into actors without framework-specific hardcoded pipelines.

Media And ML Runtime

With media and ml, Reflow graphs can author computer vision pipelines while staying graph-driven:

frame -> preprocess -> inference -> decode -> roi/taskpack -> output

A minimal inference graph built from templates:

use reflow_rt::prelude::*;
use serde_json::json;

for tpl in ["tpl_image_frame_source", "tpl_cv_preprocess",
            "tpl_ml_inference", "tpl_detection_decode"] {
    net.register_actor_arc(tpl, get_actor_for_template(tpl).unwrap())?;
}

net.add_node("src",     "tpl_image_frame_source", cfg(json!({ "path": "frame.jpg" })))?;
net.add_node("pre",     "tpl_cv_preprocess",      cfg(json!({ "resize": [640, 640], "normalize": "0_1" })))?;
net.add_node("infer",   "tpl_ml_inference",       cfg(json!({ "manifest": "detector.json" })))?;
net.add_node("decode",  "tpl_detection_decode",   cfg(json!({ "threshold": 0.4 })))?;

Packets remain ordinary Reflow messages. Synchronization uses actor inports. Model behavior comes from manifests and node configuration. Taskpacks are reusable graph exports, not privileged runtime code.

Documentation

• Workspace overview: https://github.com/offbit-ai/reflow#readme
• Reflow docs index: https://github.com/offbit-ai/reflow/tree/main/docs
• Standard component catalog: https://github.com/offbit-ai/reflow/blob/main/docs/components/standard-library.md
• Media / ML stack: https://github.com/offbit-ai/reflow/blob/main/docs/components/ml-stack.md
• Examples: https://github.com/offbit-ai/reflow/tree/main/examples

License

Reflow is dual-licensed under either MIT or Apache-2.0, at your option.