A3S Flow

A3S Flow is an event-sourced workflow engine and Rust SDK for work that must survive process restarts, delayed retries, timers, callbacks, and worker replacement. It persists every meaningful transition, projects run state from append-only history, and rejects non-deterministic replay instead of silently accepting drift.

[!IMPORTANT] Flow is the durability and replay layer. The host owns product graphs, authorization, tenant policy, tool access, and the logical idempotency of external side effects. A3S Cloud compiles product-level Workflow semantics onto Flow; it does not replace or duplicate this engine.

Why Flow

The public SDK is Rust-first. An optional NativeTsRuntime and installable a3s-flow-native-compiler compile TypeScript workflow source into a native artifact while Rust still owns history, replay, storage, workers, scheduling, and observability.

Execution model

One replay cycle has four explicit phases:

1. Flow projects the current run from immutable history.
2. FlowRuntime receives that projection and returns one RuntimeCommand.
3. Flow validates the command and appends the resulting events with an expected sequence.
4. The run replays, suspends on a wait or hook, or reaches one terminal state.

This boundary produces three important guarantees:

• Reusing a step, wait, or hook ID with different input, retry policy, deadline, token, or metadata fails as non-deterministic replay.
• A successful step is visible to workflow code only after StepCompleted is durable.
• The physical side-effect boundary remains at-least-once. If a process dies after an effect succeeds but before its output commits, the same attempt is redelivered. Step implementations must use a stable idempotency key.

Quick start

Add the engine and an async runtime:

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

Implement workflow decisions and side-effecting steps separately:

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(json!({ "message": output["message"] })));
        }

        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]
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. The same run ID, workflow specification, and input return the existing run; authority drift returns a conflict.

Core primitives

Runtime commands are deliberately small: Complete, Fail, Cancel, Timeout, RecordProgress, LinkChildOperation, ScheduleStep, ScheduleSteps, WaitUntil, and CreateHook.

Durable patterns

Stable steps and retries

Use stable IDs and make external effects logically idempotent with the run and step identity. Retry policy is part of the replayed command:

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

Ok(ctx.schedule_step_with_retry(
    "charge-card",
    "charge_card",
    json!({ "invoiceId": ctx.input()["invoiceId"] }),
    RetryPolicy::fixed(3, Duration::from_secs(30)),
))

Immediate retries stay inside the drive loop. Delayed retries persist a deadline and suspend. continue_workflow_on_failure() lets replay choose an explicit fallback or compensation after exhaustion.

Fan-out and fan-in

schedule_steps() durably creates a stable batch before executing siblings concurrently. Each settled outcome commits independently, so a slow sibling does not hold completed work in memory:

Ok(ctx.schedule_steps(vec![
    ctx.step("load-user", "load_user", json!({ "id": ctx.input()["userId"] })),
    ctx.step("load-orders", "load_orders", json!({ "id": ctx.input()["userId"] })),
]))

Timers and callbacks

Waits suspend without holding a worker:

Ok(ctx.wait_until("approval-timeout", deadline))

Hooks suspend until an external callback is received or disposed:

let metadata = a3s_flow::HookMetadata::human_approval("invoice:2026-0001")
    .with_callback_route(a3s_flow::HookCallbackRoute::post(
        "/callbacks/flow/hooks/{token}",
    ));

Ok(ctx.create_hook_with_metadata("approval", approval_token, metadata)?)

Durable consumers should retry with stable run/hook identity. Public-token helpers intentionally route only active hooks and redact bearer values from diagnostics.

Cleanup-aware cancellation

request_cancellation() records intent and moves the run to Cancelling. Workflow replay observes the request, schedules host-owned cleanup as ordinary idempotent steps, then returns ctx.cancel() for the single terminal outcome.

use a3s_flow::CancellationRequest;

engine
    .request_cancellation(
        &run_id,
        CancellationRequest::new(Some("user requested cancellation".into())),
    )
    .await?;

force_cancel() and the compatibility cancel() API intentionally skip that cleanup path.

Persistence and dispatch

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 audit-safe whole-history retention. Retention deletes only complete eligible linked components and leaves checksum tombstones; partial event-stream compaction is intentionally unsupported.

For background work, prefer BootFlowTaskManager with an A3S Boot queue. It 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 remains available to embedded hosts.

Runtime build fencing

Pin new runs with WorkflowSpec::with_runtime_build(...). A configured engine admits its current build and only the older builds the host explicitly marks compatible. RuntimeBuildTaskRouter sends due work to exact build queues; missing routes fail before a scheduler tick partially enqueues work.

Keep an old route alive until its pinned histories terminate. Use accept_unpinned() only as a bounded migration for legacy histories.

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 executable, compiler backend, working directory, protocol, OS, and architecture. In compiler-manifest mode, cold compilation verifies the complete dependency graph before and after atomic publication.

Rust remains the SDK and authority. TypeScript does not create another event store, worker, scheduler, or workflow lifecycle.

Install the compiler from crates.io and provide Bun on PATH (or set A3S_FLOW_BUN to its executable):

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

a3s-flow-native-compiler capabilities

The bundled compiler reports and verifies its Bun content fingerprint, derives the source graph from Bun's metafile, includes applicable package, lock, Bun, and TypeScript configuration files, and supervises Bun so cancellation does not leave it orphaned. NativeTsDependencyMode::CompilerManifest enables this strict graph identity. The default EntrypointOnly mode preserves compatibility with existing third-party compilers; hosts using it must continue to bump WorkflowSpec.version when imported or compiler-owned inputs change.

• Compiler and protocol contract
• Authoring definitions
• Workflow example
• Rust host example

Examples and guides

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

The deeper references keep operational detail out of this homepage:

Ownership boundary

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

Development

From this crate:

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

Repository recipes provide the supported 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 checks the public API against the latest released crate, runs the PostgreSQL gate against a real database without silently skipping store, wakeup, hook-token, retention, or worker-queue coverage, and executes the bundled Bun compiler plus a complete TypeScript workflow on Linux and Windows.

Roadmap

• Preserve semver compatibility for the implemented runtime, store, worker, and scheduler contracts.
• Keep SQLite and PostgreSQL parity gates aligned on replay, hooks, wakeups, retention, and reconnect behavior.
• Maintain the Native TypeScript compiler, dependency-manifest protocol, and artifact identity as Bun and supported targets evolve.
• Add queue or hosted observability adapters only for a concrete deployment requirement; they are extension points, not missing engine primitives.

See the functional plan for capability-level status and non-goals.

License

MIT © A3S Lab