A3S Flow

Language / 语言: English · 中文

AI Native Workflow Engine for Agents, tools, approvals, and child workflows.
Author with React or Vue, automate with the CLI and Skill, and recover every run from append-only history.

中文文档 · English documentation · Workflow Playground · Quick start · Execution model · Capabilities · Workflow DAG · Operations · Examples · Status

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.

@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.

Add Flow and an async runtime:

[dependencies]
a3s-flow = "=2.0.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

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.
• First-class activities persist a created/started/result ledger before and after host side effects. Every attempt carries an idempotency key and fencing token; hosts can append fenced heartbeats with checkpoints through heartbeat_activity. If the provider response is lost, return FlowError::UnknownOutcome and reconcile the suspended attempt with resolve_unknown_activity before allowing a retry or completion. A per-attempt timeout_ms deadline follows the same rule: timeout becomes unknown rather than an automatic duplicate retry.
• 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.
• A terminal failure in schedule_steps aborts unsettled sibling futures and persists step_cancelled markers first. The marker records when an external side-effect outcome is unknown, so hosts can reconcile the stable attempt idempotency key before any compensating retry.
• 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.

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

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,
))

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.

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. Digest format v2 is shared with the UI and follows JavaScript number formatting and UTF-16 key ordering, rejects unsafe integers, and bounds nesting to 256 levels. Edge labels remain presentation-only. An empty canvas remains importable as a draft but cannot produce an execution plan. See the runnable workflow_dsl_import example.

All stores preserve the same event envelope and replay contract.

SQLite and PostgreSQL use a3s-orm for typed access, checksummed migrations, projection-validated transactional appends, active-hook routing, scheduled-wakeup indexes, durable projection checkpoints, and whole-history retention. FlowEngine::checkpoint persists only disposable materialized state; SQL append transactions advance that cache from the validated event tail, while reads verify the latest sequence and event ID and fall back to authoritative history replay when metadata is stale or corrupt. FlowEngine::history_page exposes an exclusive sequence cursor with a bounded page size for archive/export and visibility rebuilds without loading an entire history into memory. FlowEngine::history still returns the complete log, assembled from those pages instead of one unbounded read. Production PostgreSQL deployments run migration authority separately, then admit serving workers only after verifying the canonical migration ledger. See Upgrading to Flow 1.0.

For archive workers, FlowEngine::export_history_pages pins the initial history tip, validates contiguous sequence pages, and invokes a host-owned sink one page at a time. Flow keeps the event log authoritative; Cloud or another host chooses the archive format, retention policy, and destination.

Flow task queues expose lease fencing, stale-task requeue, dead-letter inspection, and an administrative redrive_dead_lettered operation. Built-in local and PostgreSQL queues make repeated redrive safe; custom queue adapters must explicitly implement the redrive contract.

Hosts that own a compatibility FlowWorker loop can call run_until_idle_bounded(limit) to drain at most a fairness/backpressure budget before yielding to other work. A zero limit is rejected before any lease is acquired; run_until_idle() remains available when an unbounded drain is intentional.

Workers advertise the versioned FlowWorkerCapabilities contract. Hosts should call worker.ensure_compatible(&required) before leasing work; protocol or task-capability mismatches fail closed. Cloud remains responsible for queue admission, tenant fairness, placement, and processor lifecycle.

The event bridge retains stable Step/Activity attempt correlation for logs, traces, and audit sinks while keeping attempt IDs and idempotency keys out of safe_metric_labels().

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.

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.

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 2.0.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.

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

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

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.

The crate currently declares version 2.0.0. This major release documents the closed native TypeScript NativeRuntimeKind::Activity protocol extension and related public enum discriminant shifts while preserving Flow's durable replay and persistence contracts for compatible 2.x hosts.

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.

MIT © A3S Lab