A3S Flow

A3S Flow is an AI Native Workflow Engine and Rust SDK for work that must survive process restarts, delayed retries, timers, asynchronous messages, callbacks, and worker replacement. Every meaningful transition is appended to history. The engine projects state from that history and rejects replay drift instead of silently accepting a different decision. The same repository also maintains the reusable authoring package, React and Vue hooks, CLI, and coding-agent Skill that operate on Flow's versioned workflow document contract.

[!IMPORTANT] Flow owns workflow graph validation, append-only history, durable replay, and lifecycle state. The host owns node implementations, authorization, tenant policy, credentials, tool access, and logical idempotency for external side effects. A3S Cloud binds those product capabilities; it does not duplicate Flow's compiler or runtime.

Authoring components, hooks, CLI, and Skill

@a3s-lab/flow-ui is the reusable authoring package for Flow workflows. It keeps the node catalog, editor components, framework hooks, command-line tools, and agent instructions on the same manifest and graph contracts.

The Workflow Playground, React guide, Vue guide, custom node guide, CLI reference, and Skill guide document each surface. The complete node catalog and configuration reference live under Workflow nodes.

Quick start

Add Flow and an async runtime:

[dependencies]
a3s-flow = "=1.1.0"
async-trait = "0.1"
serde_json = "1"
tokio = { version = "1", features = ["macros", "rt"] }

Keep deterministic workflow decisions separate from side-effecting steps:

use a3s_flow::{
    FlowEngine, FlowError, FlowRuntime, RuntimeCommand, StepInvocation,
    WorkflowInvocation, WorkflowSpec,
};
use async_trait::async_trait;
use serde_json::json;
use std::sync::Arc;

struct GreetingRuntime;

#[async_trait]
impl FlowRuntime for GreetingRuntime {
    async fn run_workflow(
        &self,
        invocation: WorkflowInvocation,
    ) -> a3s_flow::Result<RuntimeCommand> {
        let ctx = invocation.context();

        if let Some(output) = ctx.step_output("greet") {
            return Ok(ctx.complete(output.clone()));
        }

        Ok(ctx.schedule_step(
            "greet",
            "greet_user",
            json!({ "name": ctx.input()["name"] }),
        ))
    }

    async fn run_step(
        &self,
        invocation: StepInvocation,
    ) -> a3s_flow::Result<serde_json::Value> {
        match invocation.step_name.as_str() {
            "greet_user" => {
                let name = invocation.input["name"].as_str().unwrap_or("unknown");
                Ok(json!({ "message": format!("hello {name}") }))
            }
            step => Err(FlowError::Runtime(format!("unknown step: {step}"))),
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> a3s_flow::Result<()> {
    let engine = FlowEngine::in_memory(Arc::new(GreetingRuntime));
    let spec = WorkflowSpec::rust_embedded("demo.greeting", "0.1.0", "demo", "main");

    let run_id = engine
        .start_with_id("greeting-ada", spec, json!({ "name": "Ada" }))
        .await?;
    let snapshot = engine.snapshot(&run_id).await?;

    println!("status={:?} output={:?}", snapshot.status, snapshot.output);
    Ok(())
}

start_with_id() makes creation safe to retry when the run ID, workflow spec, and input match. Authority drift returns a conflict. For a complete typed example with two durable steps, run:

cargo run --example sequential_steps

Execution model

One replay cycle has four explicit phases:

1. Project the current WorkflowRunSnapshot from immutable history.
2. Ask FlowRuntime for one deterministic RuntimeCommand.
3. Validate the command and append its events at an expected sequence.
4. Replay, suspend on durable external state, or reach one terminal outcome.

That boundary produces concrete guarantees:

• A successful step becomes visible to workflow code only after StepCompleted is durable.
• Reusing an ID with different input, retry policy, deadline, signal name, callback token, or metadata fails as non-deterministic replay.
• Timers, delayed retries, signals, and hooks release compute while the run is suspended.
• Crash recovery reconstructs state from typed events rather than an in-memory stack.

The physical side-effect boundary is intentionally at least once. If a process dies after an external effect succeeds but before its output commits, the attempt is delivered again. Step implementations must use a stable idempotency key derived from workflow and step identity.

Capability map

The following contracts are implemented on the current main branch and backed by runnable examples or integration tests.

Bounded retries

Retry policy is part of the replayed command. Exponential policies derive full jitter from immutable run, step, and attempt identity, so a restart cannot change the selected durable deadline:

use a3s_flow::RetryPolicy;
use std::time::Duration;

let retry = RetryPolicy::exponential(
    8,
    Duration::from_secs(1),
    Duration::from_secs(30),
);

Ok(ctx.schedule_step_with_retry(
    "charge-card",
    "charge_card",
    input,
    retry,
))

Bounded child fan-out

start_child_workflows() validates the entire batch, persists every generated child run ID, then advances siblings concurrently. A batch contains at most MAX_CHILD_WORKFLOW_BATCH_SIZE children (currently 64), and parent outcomes are recorded in durable request order rather than completion order.

let children = items
    .into_iter()
    .enumerate()
    .map(|(index, item)| {
        ctx.child_workflow(
            format!("item-{index:04}"),
            child_spec.clone(),
            json!({ "item": item }),
        )
    })
    .collect();

Ok(ctx.start_child_workflows(children))

Split larger fan-outs into stable windows and emit the next window only after the current outcomes are durable. Single and batched children share the same RequestCancellation and Abandon policies.

Workflow DAG

WorkflowDsl is the versioned portable document contract. Its executable payload is a directed graph of nodes and edges. Flow validates structure and derives a deterministic plan; the host binds each node's data.type to an authorized executor.

Compile the same wire shape into a stable plan and semantic identity:

use a3s_flow::WorkflowDag;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let source = std::fs::read_to_string("workflow.json")?;
    let graph = WorkflowDag::from_json(&source)?;
    let plan = graph.execution_plan()?;

    println!("order={:?}", plan.top_level());
    println!("digest={}", graph.execution_digest()?);
    Ok(())
}

The compiler rejects duplicate IDs, missing endpoints, self-edges, cycles, invalid cross-scope edges, and malformed iteration or loop containers. Unknown fields round-trip, while layout, selection, and viewport do not affect the execution digest. An empty canvas remains importable as a draft but cannot produce an execution plan. See the runnable workflow_dsl_import example.

Production operations

Persistence

All stores preserve the same event envelope and replay contract.

SQLite and PostgreSQL use a3s-orm for typed access, checksummed migrations, transactional appends, active-hook routing, scheduled-wakeup indexes, and whole-history retention. Production PostgreSQL deployments run migration authority separately, then admit serving workers only after verifying the canonical migration ledger. See Upgrading to Flow 1.0.

Retention removes only complete eligible continuation/child components and leaves checksum tombstones. Flow never compacts part of an event stream; workflows use continue_as_new to bound replay history without rewriting the source of truth.

Dispatch and optional features

BootFlowTaskManager owns processor registration, job state, retry/timeout policy, stalled-job handling, logical deduplication, startup, and shutdown. FlowWorker plus the in-memory, local-file, or PostgreSQL compatibility queues remain available to embedded hosts.

Native TypeScript

NativeTsRuntime compiles TypeScript workflow and step source into a native artifact and invokes it through a versioned JSON protocol. Artifact identity binds source, compiler backend, working directory, protocol, OS, and architecture. In strict compiler-manifest mode, Flow verifies the complete dependency graph before and after atomic publication.

Install the compiler and provide Bun on PATH (or set A3S_FLOW_BUN):

cargo install a3s-flow --version 1.1.0 --locked \
  --bin a3s-flow-native-compiler

a3s-flow-native-compiler capabilities

TypeScript is an adapter, not a second SDK, event store, scheduler, or workflow lifecycle. Read the compiler and protocol contract.

Ownership boundary

This split keeps Flow reusable as the sole durable orchestration authority without turning the SDK into a hosted product control plane.

Examples and guides

Start with one executable path, then move to the concern you need.

Development

Run checks from this crate, not from the A3S monorepo root:

cargo +1.88.0 check --all-targets --all-features --locked
cargo fmt --all -- --check
cargo test --all-targets
cargo clippy --all-targets --all-features -- -D warnings
RUSTDOCFLAGS="-D warnings" cargo doc --all-features --no-deps

Rust 1.88 is the minimum supported Rust version. Repository recipes provide the deeper matrices:

just deep-test-non-pg
A3S_FLOW_POSTGRES_URL=postgres://user:pass@localhost:5432/a3s_flow \
  just postgres-test
A3S_FLOW_NATIVE_TS_COMPILER=/path/to/a3s-flow-native-compiler \
  just native-ts-bun-test

CI also checks public API compatibility, the feature matrix, a real PostgreSQL gate, package contents, and an end-to-end Bun workflow on Linux and Windows.

Release status

The crate currently declares version 1.1.0. This compatible minor release adds bounded concurrent child-workflow batches, bounded exponential retries, and hardened custom workflow-node authoring while preserving the Flow 1.x runtime, replay, and persistence contracts.

Reusable workflow-authoring components, React and Vue hooks, the CLI, and the Skill are maintained in this repository. Hosted tenancy, authorization, product-specific capability binding, deployment policy, and the multi-tenant control plane remain outside the Rust crate; A3S Cloud owns those product surfaces.

Maintenance is contract-led: preserve SemVer and replay compatibility, keep SQLite/PostgreSQL parity gates aligned, track the native compiler protocol and supported targets, and add adapters only for concrete deployments. The functional plan is the source of truth for capability evidence and release gates.

License

MIT © A3S Lab