Run a first workflow on 0.13.1

Pin the release and add an async runtime.

Cargo.toml

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

Implement the runtime

The workflow function reads history and returns decisions. The step function performs external actions.

src/main.rs

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" => Ok(json!({
                "message": format!(
                    "hello {}",
                    invocation.input["name"]
                        .as_str()
                        .unwrap_or("unknown"),
                ),
            })),
            name => Err(FlowError::Runtime(
                format!("unknown step: {name}"),
            )),
        }
    }
}

#[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={:?}", snapshot.status);
    println!("output={:?}", snapshot.output);
    Ok(())
}

Run the program.

cargo run

The first replay returns ScheduleStep. After step output commits, the second replay returns Complete. Execution position comes from event history rather than an in-process cursor.

Inspect history

for envelope in engine.history(&run_id).await? {
    println!(
        "sequence={} key={}",
        envelope.sequence,
        envelope.event.event_key(),
    );
}

Event sequence starts at 1 and remains contiguous. Concurrent appends use an expected sequence. A losing writer rereads history and cannot overwrite committed events.

External idempotency

Physical step delivery is at least once. If a process exits after an external action succeeds but before step output commits, a replacement worker invokes the step again. Real step input needs a stable business idempotency key.

greeting-ada:greet

The target service should return the original result for the same key and parameters.

Use the native TypeScript compiler

0.13.1 publishes an installable compiler.

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

Production configuration should use NativeTsDependencyMode::CompilerManifest so the dependency graph resolved by Bun participates in source identity. The Rust host still creates FlowEngine, stores, and workers.

Next steps

Read Execution model for fixed-delay retry, waits, and hooks.
Read Stores and migrations before adding SQLite or PostgreSQL.
Use API reference for exact entry points at this tag.

#Run a first workflow on 0.13.1

#Implement the runtime

#Inspect history

#External idempotency