Installation

acts is a fast, lightweight, extensible workflow engine library that executes workflows defined in YAML format with a message-driven architecture.

Install acts Library

Install via cargo:

cargo add acts

External Storage

The persistent storage backends (sqlite/postgres/redis/nats/sled) live in the acts-store crate — enable the matching feature and import the backend from acts_store, then pass it to EngineBuilder::set_store (when unset, an in-memory MemoryStore is used):

# SQLite
cargo add acts-store --features sqlite

# PostgreSQL
cargo add acts-store --features postgres

# NATS
cargo add acts-store --features nats

# Redis
cargo add acts-store --features redis

# Sled
cargo add acts-store --features sled

use acts::Engine;
use acts_store::SqliteStore; // or PostgresStore / RedisStore / NatsStore / SledStore
use std::sync::Arc;

#[tokio::main]
async fn main() -> acts::Result<()> {
    let store = SqliteStore::open("data/acts.db").await?;
    let engine = Engine::builder()
        .set_store(Arc::new(store))
        .start()
        .await?;
    Ok(())
}

acts itself ships only MemoryStore and the KvStore trait custom stores implement; a custom backend implementing acts::KvStore is injected the same way via set_store.

Each database has one writer. The engine’s document locks are process-local — every engine in one process shares one lock table (two databases that happen to use the same keys are serialized together, which costs contention and nothing else) — and they do not cross processes, so two processes writing one database have no mutual exclusion: their concurrent updates of a row can leave the index entries disagreeing with the data row, a query matching a value the row no longer holds or missing the one it does. A single-instance deployment is unaffected; a deployment with several gives each its own database, or coordinates outside the engine (a backend conditional write, or a lock spanning the read) — batch makes one write atomic, not a read plus another process’s write.

Create Engine

#![allow(unused)]
fn main() {
use acts::{Engine, Principal};

let engine = Engine::builder().start().await.unwrap();
// The executor acts for a caller; an embedder driving the engine itself is
// the engine's own principal (see the access-control chapter).
let executor = engine.executor(&Principal::unrestricted());
}

Deploy and Start Workflow

#![allow(unused)]
fn main() {
use acts::{Engine, Principal, Vars, Workflow};

let engine = Engine::builder().start().await.unwrap();

// Load YAML model
let model = r#"
id: my_model
name: my model
steps:
  - name: step 1
    uses: acts.transform.set
    params:
      a: 10
  - name: step 2
    uses: acts.transform.code
    params: |
      return { data: a + 10 };
"#;
let workflow = Workflow::from_yml(model).unwrap();

// Deploy model
let executor = engine.executor(&Principal::unrestricted());
executor.model().deploy(&workflow).expect("fail to deploy workflow");

// Start workflow
let mut vars = Vars::new();
vars.set("a", 0);
vars.set("pid", "w1");
executor.proc().start(&workflow.id, vars).expect("fail to start workflow");
}

Related Projects

Access Control

Every operation of the engine is checked against the identity of the caller performing it. An [acl] section in the engine config is what names those callers. Without it, the engine answers to anyone and hands out only the catalogue: every request is attributed to the built-in anonymous subject, which may list and get models and packages and nothing else — no other read, no write, no control action, no admin action, no snapshot scope, no subscription. An unconfigured deployment is for looking at what is deployed, not for changing it or for reading what a run, a message or a trigger holds.

Two ways out of that default:

• add [acl] — the smallest useful section is one token, which grants that token everything (the requirepass equivalent) and switches every caller from anonymous to authenticated;
• write enabled = false inside [acl], the explicit opt-out: nothing is enforced and every caller is unrestricted. That is the pre-ACL behaviour, and it is a deliberate choice rather than the absence of a section. An embedder says the same thing with Engine::builder().disable_acl(), which is the setting a test or a local demo uses.

There is no implicit unrestricted policy anywhere: not a missing section, not a process nobody claimed, not a snapshot scope nobody named. Each of those is the read-nothing case described where it appears below.

Tokens and roles

A request carries a token; the token selects a role; the role’s allow / deny action patterns decide. deny always wins. A request with no token — or with a token matching no role — is refused unless default_role names a role to fall back to ([[acl.role]] name = "anonymous" is how a deployment keeps the read-only default while configuring everything else around it).

Tokens are compared by SHA-256 digest. Write sha256:<64 hex digits> to keep the clear text out of the config file, or write the token itself and let the server hash it at load:

[acl]
# role applied to an absent/unknown token; omit to refuse such requests
default_role = "guest"

# shorthand: one unrestricted token (the requirepass equivalent)
token = "sha256:9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08"

[[acl.role]]
name = "operator"
tokens = ["sha256:2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae"]
allow = ["model:ls", "model:get", "proc:ls", "proc:get", "task:*", "msg:ls",
         "msg:ack", "snap:get", "snap:ls", "acl:whoami"]
deny = ["model:rm", "pack:publish"]
snapshot = { secrets = ["$subject"], profile = ["$subject/*"] }
workdir = "/srv/acts"

[[acl.role]]
name = "guest"
tokens = ["sha256:..."]
allow = ["model:ls", "acl:whoami"]

A malformed policy is a startup error, never a silent allow or a silent refuse: a pattern that does not compile, a token assigned to two roles, a default_role naming no role, or an enabled section where no role declares any token.

Action names

allow/deny match the action names of the shared dispatch table, with * and ? globs — so act:* covers every act operation. They are the same names the CLI and the channel client use.

The last row is the embedder’s own surface — installing a user var module into the expression environment, publishing a package definition. No wire action maps to it, so a transport caller cannot reach it; an embedder that extends the engine it hosts passes its own principal (see The executor).

allow = ["*"] is unrestricted: every action passes, and every snapshot scope too. acl:whoami reports the caller’s own identity and effective patterns; it is implicitly allowed for an authenticated caller, so it works as a startup check without widening anything.

The anonymous subject an engine without [acl] resolves to gets exactly model:ls model:get pack:ls pack:get — the catalogue. That is the least a caller the engine cannot name may be trusted with, and the list is pinned by a test rather than left to interpretation. Everything else is out, including reads a named caller would take for granted: a process row names who ran what, a delivery names who was meant to receive it, a trigger names the model it will start — a caller the engine cannot identify is not the one those rows are about. msg:sub is out for the same reason and one more (a stream both carries live payloads and stores a delivery row per message for a channel it holds), and snap:get/snap:ls are out because a snapshot scope has an owner only when a policy names one.

The executor

The engine’s operations are grouped on one object, the executor, and every one of its methods is checked before it runs — model().deploy(), proc().start() and the rest. It is bound to a caller when it is created:

#![allow(unused)]
fn main() {
// a transport's request, after its token resolved
let executor = engine.executor(&principal);
executor.proc().start("my_model", vars).await?;

// a request that carried no token
let executor = engine.executor(&engine.anonymous());

// the engine's own work, and what a test or a local demo passes
let executor = engine.executor(&Principal::unrestricted());
}

The executor also decides what a run it starts carries: proc().start() and evt().start() seal the principal’s snapshot scopes and workdir root into the process, so a caller cannot widen its own reading by putting an authority in the request. Two callers of the same model therefore read what each of them owns.

An embedder is not outside this: it reaches the engine through an executor like everyone else, so its operations are checked against the principal it passed. Passing Principal::unrestricted() is a statement (“this is the engine’s own work”, or “this deployment opted out”) and it is written at the call site.

Snapshot scope ownership

A snapshot target is addressed by target × scope (snapshot-backed sealed data). The snapshot table of a role narrows which scopes of which targets that role owns; $subject is replaced by the role name, so ["$subject"] means “my own scope only”.

The rule is enforced twice:

• On the snap:* actions — a read or write of a scope outside the caller’s set is refused, and snap:ls returns only the scopes the subject owns, so one tenant cannot enumerate another’s.
• At seal time — a process carries its starter’s rules, and the scheduler re-checks them before freezing a snapshot value into a task. A workflow therefore cannot read another subject’s data by being started with someone else’s uid.

A run’s authority comes from one place: the principal whose executor started it. A subflow inherits its parent’s authority, so it can never read more than the run that opened it. A run the engine started with no caller at all — a schedule trigger, an embedder calling Runtime::start directly — carries no authority and reads no snapshot scope: an absent authority is not an unlimited one, and a task that needs owned data fails at its seal with the subject it lacks instead of being handed the data plane.

Message face

Messages and their deliveries are authorized by action grants, like every other operation — a process’s owner has no say over who may read or ack the messages it emitted.

• Subscribing is an action. Opening a stream (gRPC on_message, SSE /msg/sse) requires msg:sub in the role’s allow list; without it the transport answers PERMISSION_DENIED / 403 instead of a stream. The transport hands the client id it received to that action and registers the channel under the key the action answers with, so the checked path and the occupied key cannot drift apart.
• The key is namespaced by subject: {subject}/{transport id} — the subject as the prefix, the transport’s own client id (SSE keeps its acts-flow-client- segment) behind it. A second caller naming a client id another subject already uses therefore subscribes to a different channel instead of replacing that subject’s handler. msg:unsub composes the same key from the same id, so a caller only names channels in its own namespace — and a role name carrying / is refused at config load, which is what keeps the prefix unambiguous.
• Delivery follows the filters and the grants. A channel receives every message that matches its self-declared type/state/uses/options globs, whoever started the emitting process; what a caller may do with the messages it receives is decided by the actions it holds. A role that must not read message payloads simply has no msg:sub (and no msg:ls/msg:get).
• A subscription’s backlog is bounded. Every subscriber has one fixed-size queue ([grpc].queue_size, default 128; [web].queue_size, default 100): a delivery never waits for the client, and a full queue means the client stopped reading, so that subscription is disconnected — the engine does not leave one waiting task per message that did not fit. What the engine still owes the channel is not dropped with it: a delivery that was handed over but not acked, of a process that has not settled, is re-sent by the retry timer once the client subscribes again under the same id (which composes the same channel key). As after any disconnect, a channel only receives messages emitted while it is registered.
• msg:ack and msg:unsub are ordinary actions too: any role granted them may ack any delivery id and unsubscribe any channel in its own namespace. The delivery id is not addressed per client, so grant msg:ack the way you would grant a write — its holder can silence another caller’s unacked messages.

What is enforced where

Four kinds of rule, and each is checked at the layer that can express it:

Every operation goes through the action table, which is what makes the check universal: a transport resolves a token into a principal, and so does an embedder — the executor an operation runs on carries that principal, and the dispatch table and the executor match the same action names, defined once per operation. The two callers that do not have an identity are the two the table has an answer for anyway:

• a request with no token resolves to default_role, or to the read-only anonymous subject when the engine has no [acl] section — it is still a caller, and it is checked like any other;
• a run nothing claims (a schedule trigger) and a subflow (which inherits its parent’s authority) are the engine’s own starts, with no caller to check: what they may read is the authority they ended up carrying, and neither can gain one from the outside.

Directory control

The configured workdir — on [acl], or per role to override it — is a root: every process the policy starts gets its own directory <workdir>/<pid>, and the run’s filesystem access is confined to that one. It is created at start, and the process id becomes a path segment — so a pid that is not one safe component (empty, ., .., or containing a path separator or colon) is refused rather than placing the run outside the root it was given.

The root travels the same private route as the scope authority — it is part of it: sealed into the process env under a key the workflow’s $env proxy refuses, persisted with the process, and never a start option, so a caller cannot name the directory a run is confined to. The run’s own directory (<root>/<pid>) is what an act reads through Context::workdir() and what a script reads as $env.WORK_DIR — the same directory under two names, neither of them the configured root. $env.WORK_DIR is engine-owned, so a write to that name is dropped and a run cannot redefine where it runs. The directory lives exactly as long as the process’s durable rows: the sweeper removes it with them (a finished run whose delivery errored and awaits a manual retry keeps its row, and its directory with it), and a start that never became durable removes its own directory immediately — so nothing left in there outlives the run.

acts.app.shell uses it: the script runs with that directory as its working directory, HOME, TMPDIR/TEMP/TMP and PWD point inside it, and ACTS_WORKDIR names it for the script. A script that names an absolute path (/etc/passwd, C:\Windows) or a .. segment is refused before it runs.

That textual check is policy, not a sandbox: it is what makes the direct escape a loud failure instead of a silent success, but a shell can spell a path in ways no textual check follows (a=/etc; cat $a/passwd, a symlink inside the workdir) — the containment that actually holds is the child’s working directory. Treat a hostile workflow as needing an OS boundary (a container or namespace around the server); per-process directories keep such runs from colliding meanwhile.

The shell package has a second, script-level policy of its own — two glob lists over the whole script text:

[shell]
# when non-empty, only a script matching one of these may run
allow = ["ls", "ls *", "cat *.txt"]
# always refused, allow or not
deny = ["*rm -rf*", "*sudo *"]

* matches any run of characters, / and newlines included, and deny wins. Both lists empty means no restriction; a pattern that does not compile fails startup. Like the workdir check it is policy rather than a sandbox — a glob over script text cannot see what the script will do (a=rm; $a -rf / names no forbidden word), so it is for stating intent and refusing the obvious, and a hostile workflow still needs an OS boundary.

Without workdir, no directory control applies and a process may touch whatever the server’s own account can — the behaviour before this option existed.

Transport credentials

Clients

# CLI: the flag wins over the environment variable
acts-cli --token "$TOKEN"
ACTS_TOKEN="$TOKEN" acts-cli

The CLI resolves its identity with acl:whoami before entering the REPL, so a missing or stale token fails at startup instead of on the first command.

#![allow(unused)]
fn main() {
use acts_channel::ActsChannel;

let mut client = ActsChannel::connect_with_token("http://127.0.0.1:10080", Some(token)).await?;
}

connect is the tokenless form: it is the anonymous caller, so against a server without [acl] it can read and nothing else, and against a configured one it is refused unless a default_role admits it.

Channel Client

The client channel is used by application services to connect to the workflow service, subscribe to messages, and execute workflow tasks.

Installation

Install the client library via cargo:

cargo add acts-channel

Supported Languages

Basic Usage

#![allow(unused)]
fn main() {
use acts_channel::{Client, ChannelOptions};

let mut client = Client::new("http://localhost:8080", &ChannelOptions::default());

// Connect to the service
client.connect().await?;

// Subscribe to messages
client.subscribe("client1", "act*", None, None).await?;

// Deploy a model
client.deploy(&model_str).await?;

// Start a workflow
let mut vars = Vars::new();
vars.set("input", 100);
client.start("model_id", vars).await?;
}

Install acts-channel client

Install via cargo:

cargo add acts-channel

Connect

Application services create and connect to the service via ActsChannel.

#![allow(unused)]
fn main() {
use acts_channel::ActsChannel;

let mut client = ActsChannel::connect("http://localhost:8080");
}

Subscribe

Subscribe to workflow messages via the client channel.

Subscribe to Messages

#![allow(unused)]
fn main() {
use acts_channel::{ActsChannel, ActsOptions};

let mut client = ActsChannel::connect("http://127.0.0.1:10080").await?;

// ActsOptions fields support glob patterns, e.g. "act*" matches every message
// starting with "act"
let options = ActsOptions {
    state: Some("{created,completed}".to_string()),
    r#type: Some("act*".to_string()),
    ..ActsOptions::default()
};

let sub = client
    .subscribe(
        "client-1",
        move |message| {
            println!("{message:?}");
        },
        // faults that do not end the feed: decode failures, failed auto-acks
        move |err| eprintln!("subscription fault: {err}"),
        &options,
    )
    .await?;

// the end of the feed: Ok(()) on a clean close, Err(status) on a failure
if let Err(err) = sub.wait().await {
    eprintln!("subscription closed: {err}");
}
}

The id is a namespace component of the caller’s subject on the server ({subject}/{client_id}): two subscribers of different subjects can both subscribe as client-1 without colliding, and under [acl] a subscription carries the messages of the processes that subject started, not every tenant’s. See access control.

Message Types

Deploy

Deploy workflow models via the client channel.

Deploy Model

#![allow(unused)]
fn main() {
use acts_channel::ActsChannel;

let mut client = ActsChannel::connect("http://localhost:8080");

// Load model from file and deploy
let model = std::fs::read_to_string("workflow.yml").unwrap();
let resp = client
    .deploy(yml, Some("custom_model_id")).await?;
}

Build Model Dynamically

You can also build models dynamically via Rust code before deploying:

#![allow(unused)]
fn main() {
use acts::{Workflow, Vars};

let workflow = Workflow::new("my_model", "my workflow")
    .with_step(|step| {
        step.with_uses("acts.core.irq", Vars::new().with("key", "my_key"))
    });

let model_str = serde_yaml::to_string(&workflow).unwrap();
client.deploy(&model_str, Some("custom_model_id")).await?;
}

Start

Start a workflow via the client channel.

Start Workflow

#![allow(unused)]
fn main() {
use acts_channel::{ActsChannel, ChannelOptions};

let mut client = ActsChannel::connect("http://localhost:8080");

// Start workflow
let mut vars = Vars::new();
vars.set("a", 100);
client.start("model_id", vars).await?;
}

Start Parameters

You can pass variables when starting to override the workflow’s default vars:

#![allow(unused)]
fn main() {
let mut vars = Vars::new();
vars.set("input_value", 42);
vars.set("user_name", "admin");

client.start("my_workflow", vars).await?;
}

Execute

Execute actions on activities via client channel.

Complete Activity

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("result", "done");
client.complete(&pid, &tid, options).unwrap();
}

Trigger Error

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("ecode", "err_custom");
client.fail(&pid, &tid, options).unwrap();
}

Back to Specific Step

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("to", "step1");
client.back(&pid, &tid, options).unwrap();
}

Cancel Activity

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("to", "step1");
client.cancel(&pid, &tid, options).unwrap();
}

Skip Activity

#![allow(unused)]
fn main() {
client.skip(&pid, &tid, Vars::new()).unwrap();
}

Abort Activity

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("uid", "u1");
client.abort(&pid, &tid, EventAction::Abort, options).unwrap();
}

Remove Activity

#![allow(unused)]
fn main() {
client.remove(&pid, &tid, EventAction::Remove, Vars::new()).unwrap();
}

CLI Tool

Users can view data and manage workflows using the built-in CLI tool:

acts-cli -h <server_ip> -p <port>

Help

> help

Subscribe

Clients can subscribe to messages via the sub command:

sub <client_id> [type] [state] [tag] [key]
    subscribe server message
    type, state and tag are all support glob string

    client_id:  client id
    type: message types are in workflow, step, branch and act.
    state: message state in created, completed, error, cancelled, aborted, skipped and backed.
    tag: message tag which is defined in workflow model tag attribute.
    key: message key

    for examples:
    1. sub all messages:
    sub 1
    2. sub all act messages:
    sub 1 act
    3. sub created and complete messages
    sub 1 * {created,completed}
    4. sub all messages that the tag starts with abc
    sub 1 * * abc*
    5. sub all messages that the key starts with 123
    sub 1 * * * 123*

Deploy

Deploy a workflow model file:

deploy <path>
    deploy a workflow

    path: yml model local file path

Start

Start a workflow:

start <mid>
    start a workflow

    mid: workflow model id

Manager

Model List

List deployed models:

models [count]
    query the current deployed models

    count: expect to load the max model count

View Model

View model data:

model <mid> [fmt]
    query the model data
    mid: model id
    fmt: display format with text|json|tree

Process List

List all running processes:

procs [count]
    query the current running procs
    count: expect to load the max proc count

View Process

View process data:

proc <pid> [fmt]
    query the proc data
    fmt: display format with json|tree, the default is tree

Task List

List all task list of a process:

tasks <pid>
    query the proc tasks
    pid: the proc id

View Task

View task data:

task <pid> <tid>
    query the task data

Execute

Execution is for activities of type req. When the server generates a req activity, it is in an interrupted state, waiting for the client to execute.

Env

Each command execution requires some options parameters. This command generates the options parameters needed for subsequent execution actions.

env <op> [key] [value] [value-type]
    op: command with set, get, ls
            set: set key and value.
            get: get by key name
            ls: list all env values
            json: show in json format
    key: env key with string type
    value: env value
    value-type: value type with string, int, float and json, the default type is string

Push

Push a request (req) activity:

push <pid> <tid>
    push an action to a step

    pid: proc id
    tid: step task id

    extra options:
        id: act id, it is required
        name: act name
        inputs: input parameters
        outputs: expose vars to its parents
        rets: limits the request options when acting

Remove

Remove an activity:

remove <pid> <tid>
    remove an action

    pid: proc id
    tid: task id

Submit

Submit an activity:

submit <pid> <tid>
    submit an action

    pid: proc id
    tid: task id

Complete

Complete an activity:

complete <pid> <tid>
    complete the action

    pid: proc id
    tid: task id

Back

Back an activity:

back <pid> <tid>
    back to the history task

    pid: proc id
    tid: task id

    options:
        to: set a step id to point out which step to back

Cancel

Cancel an activity that is completed but whose next step has not yet been completed:

cancel <pid> <tid>
    cancel the act that is completed before

    pid: proc id
    tid: task id

Skip

Skip an activity and continue to the next step:

skip <pid> <tid>
    skip the action

    pid: proc id
    tid: task id

Abort

Abort an activity, terminating the entire workflow:

abort <pid> <tid>
    abort the workflow

    pid: proc id
    tid: task id

Error

Set an activity as error. If the activity has no error handling configured, the error propagates upward until the entire workflow ends.

error <pid> <tid>
    set an action as error

    pid: proc id
    tid: task id

    options:
        err_code:  error code, it is required
        err_message: error message

Model

The execution of the workflow engine depends on the workflow model. An acts workflow model is a standardized YAML file.

Model Structure

A complete workflow model consists of the following parts:

id: my_model
name: Model Name

# Default variables
vars:
  - name: value
    value: 0

# Input schema (JSON Schema)
inputs:
  type: object
  properties:
    value:
      type: number

# Output schema (JSON Schema)
outputs:
  type: object
  properties:
    data:
      type: object

# Start triggers
on:
  - id: event1
    kind: manual

# Execution options
options:
  exposes:
    - name: output_key

# Step list
steps:
  - id: step1
    uses: acts.core.irq

Core Concepts

Inputs

The workflow model can define input schema to constrain the variables passed when starting a workflow.

Input Schema

Uses JSON Schema format to define the input schema:

id: my_model
name: test
inputs:
  type: object
  properties:
    a:
      type: integer
      default: 10
    user_name:
      type: string
  required:
    - a

Passing Inputs When Starting

#![allow(unused)]
fn main() {
use acts::{Engine, Principal, Vars, Workflow};

let engine = Engine::builder().start().await.unwrap();
let executor = engine.executor(&Principal::unrestricted());

let mut vars = Vars::new();
vars.set("a", 100);
vars.set("user_name", "admin");
executor.proc().start("my_model", vars)?;
}

Dynamic Input Setting

You can set the inputs value using ModelBuilder:

#![allow(unused)]
fn main() {
use acts::model::Workflow;

let mut workflow = Workflow::new("my_model", "my workflow")
    .set_inputs(serde_json::json!({
        "type": "object",
        "properties": {
            "a": { "type": "integer" },
            "b": { "type": "string" }
        }
    }));
}

Step Inputs

Input data can also be received at the step level:

steps:
    - id: step1
      vars:
        - name: local_var
          value: '${{ inputs.a }}'

Outputs

The workflow model can define an output schema that constrains the data exposed when a workflow completes.

Exposing Outputs

Use exposes to filter which variables are exposed as outputs:

Workflow Level

id: my_model
name: test
exposes:
  - name: result
  - name: data

Step Level

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      exposes:
        - name: step_output

Triggers

Workflows declare triggers through the on field; a triggered workflow is started by the engine or by a caller. A trigger only declares the start surface of the workflow — it never runs inside a process.

Trigger Types (kind)

id: m1
name: test
on:
  - id: event_manual
    kind: manual
    name: start by manual
    # default start inputs used when the caller passes no payload
    params:
      value: 0

  - id: event_hook
    kind: hook

  - id: event_chat
    kind: chat

  # cron expression of 6 fields: sec min hour day month dow
  - id: event_schedule
    kind: schedule
    schedule: "0 * * * * *"
    params:
      value: 0

• manual/chat/hook fire through executor.evt().start("model-id:trigger-id", &payload).await; a null payload falls back to the declared params.
• manual triggers double as web URL triggers — an HTTP transport (e.g. acts-plugin-web’s POST /hooks/{model-id}:{trigger-id}) starts them with the request body as payload, so no separate webhook kind is needed.
• schedule triggers keep their run state (last_run/next_run) on the deployed trigger row and are polled by the engine timer. Re-deploying a model reconciles the trigger data — changed declarations are updated and removed triggers are cleaned up.
• kind may also be any registered event package id (custom triggers), fired through the package registry.

Catches

The error handling mechanism allows recovery when a step encounters an error.

Step-Level Catches

Steps define error handling via catches, which is a list of Step objects:

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      catches:
        # Match specific error code
        - uses: acts.core.msg
          if: $ecode() == 'err1'
          params:
            key: catch_err1

        # Match all unhandled errors
        - uses: acts.core.msg
          params:
            key: catch_others

Error Handling Flow

1. An activity in a step triggers an error (via EventAction::Error or acts.core.action)
2. The engine checks the catches list conditions in order
3. Executes the corresponding handler when the first matching catch is found
4. After handling, the step continues normal execution
5. If no catch matches, the error propagates upward

Error Code

Error codes are passed via ecode:

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("ecode", "err1");
rt.do_action2(&pid, &tid, EventAction::Error, options).unwrap();
}

Use $ecode() in the catch if condition to get the error code.

Package

Packages are reusable functional modules used via uses in steps and activities.

Built-in Packages

Core Packages

Transform Packages

Workflow start triggers are declared on the on field (manual/chat/hook/schedule); see Triggers.

Usage Example

steps:
    # IRQ — interrupt and wait for client response
    - id: step1
      uses: acts.core.irq
      params:
        key: act1

    # MSG — one-way notification
    - id: step2
      uses: acts.core.msg
      params:
        key: notification

    # Set variable
    - id: step3
      uses: acts.transform.set
      params:
        a: 10

    # Execute JavaScript
    - id: step4
      uses: acts.transform.code
      params: |
        return { result: a + 10 };

Custom Package

You can create custom packages. Refer to the package example for details.

Step

A Step is the basic execution unit of a workflow. Each step can use a built-in or custom package (uses) and pass parameters (params). Steps execute sequentially and can also include branches, error handling, and timeout handling.

name: test
steps:
    - id: step1
      name: step 1
      uses: acts.core.irq
      params:
        key: act1

    - id: step2
      name: step 2
      uses: acts.transform.set
      params:
        a: 10

Step Attributes

A while step is a bounded loop: the condition is re-evaluated before each iteration and the step re-executes while it holds; once it fails the step is skipped and the flow falls through to the next step declared after it:

steps:
    - id: add
      while: index < input
      uses: acts.transform.code
      params: |
          $set("value", value + index);
          $set("index", index + 1);

    - id: end

A step whose if condition fails is also skipped and falls through to the next declared step (a self/backward next is then not taken), so if and next keep their original meanings and while cannot be combined with next.

Multi-Act Steps

When a step needs to execute multiple activities, use the acts.core.block package and nest child activity lists in params:

steps:
    - id: step1
      uses: acts.core.block
      params:
        mode: sequence
        acts:
          - uses: acts.core.irq
            params:
              key: act1
          - uses: acts.core.msg
            params:
              key: msg1

acts.core.parallel can execute over a collection in parallel, and acts.core.sequence can execute sequentially.

Step Setup

Steps can define local variables and export options.

Local Variables

Use vars to define local variables for a step:

steps:
    - id: step1
      vars:
        - name: count
          value: 0
        - name: list
          value:
            - u1
            - u2
      uses: acts.core.irq
      params:
        key: act1

Exporting Outputs

Use options.exposes to export step-level output variables:

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      options:
        exposes:
          - name: result

The exported result variable will be available to subsequent steps.

Step Catches

When a step encounters an error, use catches to define error handling logic.

Basic Usage

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      catches:
        - uses: acts.core.msg
          if: $ecode() == 'err1'
          params:
            key: catch_err1

        - uses: acts.core.msg
          params:
            key: catch_others

Matching All Errors

If no if condition is specified, the catch matches all errors:

catches:
    - uses: acts.core.msg
      params:
        key: catch_all

Triggering Errors

Errors are triggered on the client side via EventAction::Error:

#![allow(unused)]
fn main() {
let mut options = Vars::new();
options.set("ecode", "err1");
rt.do_action2(&pid, &tid, EventAction::Error, options).unwrap();
}

Or via acts.core.action in timeouts:

timeouts:
    - uses: acts.core.action
      if: $cost_in('8s')
      params:
        action: error
        options:
          ecode: err_timeout

Catch Execution Order

1. The catcher checks conditions from top to bottom
2. The first catch with a matching condition is executed
3. After the catch completes, the step continues normal execution
4. If no catch matches, the error propagates upward

Step Timeout

When a step exceeds a specified time, use timeouts to define timeout handling logic.

Basic Usage

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      timeouts:
        # Trigger message in >=2 seconds and < 8 seconds
        - uses: acts.core.msg
          if: $cost_in('2s', '8s')
          params:
            key: step1_timeout_2s

        # Trigger error after 8 seconds
        - uses: acts.core.action
          if: $cost_in('8s')
          params:
            action: error
            options:
              ecode: err_timeout_8s

Time Duration

$cost_in() supports the following duration formats:

Timeout Check Interval

Set the tick interval for timeout checking via options:

options:
  tick_interval: 500

The default value is 1000 (milliseconds). The example sets it to 500ms for more frequent checks.

Step Triggers

Steps do not declare their own triggers. Workflow startup is managed through the workflow-level on field.

Workflow-Level Triggers

Workflow triggers are fired by the engine timer or by a caller. See Triggers for details.

name: test
on:
  - id: event1
    kind: manual
steps:
  - id: step1
    uses: acts.core.irq

Trigger Flow

1. A trigger fires (manual/chat/hook by a caller, schedule by the engine timer)
2. The workflow instance starts
3. All steps execute in sequence
4. Steps complete, workflow ends

Steps have no trigger declarations of their own; all startup control is via the workflow-level on configuration.

Step Condition

Use the if attribute to control whether the current step executes.

steps:
    - id: step1
      uses: acts.transform.set
      params:
        a: 10

    - id: step2
      if: '${{ a }} > 0'
      uses: acts.core.irq
      params:
        key: act1

    - id: step3
      if: '${{ a }} <= 0'
      uses: acts.core.msg
      params:
        key: skipped

Expression Syntax

Step conditions use ${{ }} for variable interpolation:

# Numeric comparison
if: '${{ count }} >= 10'

# String comparison
if: '${{ status }} == "active"'

# Boolean check
if: '${{ flag }} == true'

# Logical AND
if: '${{ a }} > 0 && $get("b") == "yes"'

# Logical OR
if: '${{ a }} > 0 || $get("b") == "yes"'

When the if condition is not met, the step is skipped and execution continues with the next step.

Step Acts

A step can contain multiple activities. Activities can be combined in different ways depending on the package used.

Single Act

The simplest case: a step uses a single act:

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1

Block (Nested Acts)

Use acts.core.block to nest multiple acts within a step. Acts execute in sequence mode by default:

steps:
    - id: step1
      uses: acts.core.block
      params:
        mode: sequence
        acts:
          - uses: acts.core.irq
            params:
              key: act1
          - uses: acts.transform.set
            params:
              a: 10
          - uses: acts.core.msg
            params:
              key: done

Parallel (Loop over List)

Use acts.core.parallel to execute acts in parallel over a list:

steps:
    - id: step1
      vars:
        - name: users
          value:
            - u1
            - u2
      uses: acts.core.parallel
      params:
        in: '${{ users }}'
        acts:
          - uses: acts.core.irq
            params:
              key: act1

The engine automatically injects index and value into each child act’s variable context.

Sequence (Chain over List)

Use acts.core.sequence to execute acts sequentially over a list (each act waits for the previous one to complete):

steps:
    - id: step1
      vars:
        - name: users
          value:
            - u1
            - u2
      uses: acts.core.sequence
      params:
        in: '${{ users }}'
        acts:
          - uses: acts.core.irq
            params:
              key: act1

Comparison

Branch

Branches allow conditional branching at a step. Set the branches attribute to define multiple branches, each with its own if condition and list of child steps.

name: test
steps:
    - id: step1
      uses: acts.transform.set
      params:
        a: 5
    - id: step2
      branches:
        - id: b1
          name: branch 1
          if: '${{ a }} > 0'
          steps:
            - id: step3
              uses: acts.transform.set
              params:
                result: positive
        - id: b2
          name: branch 2
          steps:
            - id: step4
              uses: acts.transform.set
              params:
                result: zero_or_negative

Branch Attributes

Branch Dependencies

Use needs to declare dependencies between branches:

branches:
    - id: b1
      needs: [b2]
      steps:
        - id: step3
    - id: b2
      steps:
        - id: step4

If branch b1 depends on b2, the engine sets b1 to Pending state until b2 is completed.

Expressions in Conditions

Branch conditions use ${{ }} expression syntax:

# Variable comparison
if: '${{ a }} > 0'

# Multi-condition
if: '${{ a }} > 0 && $get("status") == "active"'

Act

An Act is the actual action execution body, using uses to specify a functional package.

Act Attributes

Usage

Single Act (step-level)

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1

Multiple Acts (nested in block)

steps:
    - id: step1
      uses: acts.core.block
      params:
        mode: sequence
        acts:
          - uses: acts.core.irq
            params:
              key: act1
          - uses: acts.core.msg
            params:
              key: msg1

Built-in Packages

Set

Use acts.transform.set to set variable values.

steps:
    - id: step1
      uses: acts.transform.set
      params:
        a: 10
        b: hello
        c:
          x: 1

Variable Override

If the setting variables already exist in the current context, they will be overridden.

Set at Act Level

Set can also be used within a block:

steps:
    - id: step1
      uses: acts.core.block
      params:
        mode: sequence
        acts:
          - uses: acts.transform.set
            params:
              count: 0
          - uses: acts.core.irq
            params:
              key: act1

IRQ

acts.core.irq is an interrupt request activity that pauses workflow execution and waits for a client response.

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1

Parameters

Client Handling

On the client side, subscribe to req type messages, then complete the activity:

#![allow(unused)]
fn main() {
use acts::event::EventAction;

fn on_message(msg: &Message) {
    if msg.r#type == "req" && msg.key == "act1" {
        // Process the business logic
        let mut outputs = Vars::new();
        outputs.set("result", "processed");

        // Complete the activity
        rt.do_action2(&msg.pid, &msg.tid, EventAction::Next, outputs).unwrap();
    }
}
}

Other Actions

In addition to Next (complete), IRQ activities can also be handled with other actions:

MSG

acts.core.msg is a one-way message activity that sends a notification to the client without pausing workflow execution.

steps:
    - id: step1
      uses: acts.core.msg
      params:
        key: notification

Parameters

Difference from IRQ

Client Reception

On the client side, subscribe to msg type messages to receive notifications:

#![allow(unused)]
fn main() {
fn on_message(msg: &Message) {
    if msg.r#type == "msg" {
        match msg.key.as_str() {
            "notification" => {
                println!("Received notification: {:?}", msg);
                // No need to call do_action to respond
            }
            _ => {}
        }
    }
}
}

Block

Use acts.core.block to combine multiple activities into a block. Supports sequence and parallel execution modes.

name: test
steps:
    - id: step1
      uses: acts.core.block
      params:
        # Execution mode: sequence or parallel
        mode: sequence
        acts:
          - uses: acts.transform.set
            params:
              count: 0
          - uses: acts.core.irq
            params:
              key: act1
          - uses: acts.core.msg
            params:
              key: done

Mode Comparison

Variable Export

Child activities within a block can export variables to the parent node via options.exposes:

steps:
    - id: step1
      uses: acts.core.block
      params:
        mode: sequence
        acts:
          - uses: acts.core.irq
            params:
              key: act1
            options:
              exposes:
                - name: result

Parallel

Use acts.core.parallel to execute a collection in parallel — all child activities start simultaneously without depending on each other.

name: test
steps:
    - id: step1
      vars:
        - name: items
          value:
            - u1
            - u2
            - u3
      uses: acts.core.parallel
      params:
        in: '${{ items }}'
        acts:
          # Generates 3 IRQ activities, all executing in parallel
          - uses: acts.core.irq
            params:
              key: act1

Comparison

Variable Injection

The engine automatically injects index and value into each child activity’s variable context, accessible via ${{ index }} and ${{ value }}.

Dynamic Collection with Code

Combine with acts.transform.code to dynamically generate collections:

steps:
    - id: step1
      uses: acts.transform.code
      params: |
        let list = ["u1", "u2", "u3"];
        $set("items", list);
    - id: step2
      uses: acts.core.parallel
      params:
        in: '${{ items }}'
        acts:
          - uses: acts.core.irq
            params:
              key: act2

Sequence

Use acts.core.sequence to execute sequential chains over a collection, where each subsequent execution depends on the completion of the previous one.

name: test
steps:
    - id: step1
      vars:
        - name: items
          value:
            - u1
            - u2
      uses: acts.core.sequence
      params:
        in: '${{ items }}'
        acts:
          # Generates 2 IRQ activities, executed one by one in sequence
          - uses: acts.core.irq
            params:
              key: act1

Comparison

The engine automatically injects index and value into each child activity’s variable context.

Subflow

Use acts.core.subflow to invoke another workflow model (sub-workflow).

name: test
steps:
    - id: step1
      uses: acts.core.subflow
      params:
        # The target sub-workflow model ID
        to: sub_workflow_id
        # Input data passed to the sub-workflow
        a: '${{ value }}'

Sub-Workflow Definition

A sub-workflow is an independent workflow model:

id: sub_workflow_id
name: sub_flow
inputs:
  type: object
  properties:
    a:
      type: integer
outputs:
  type: object
  properties:
    result:
      type: string
steps:
    - id: sub_step1
      uses: acts.core.irq
      params:
        key: sub_act

    - id: sub_step2
      uses: acts.core.msg
      params:
        key: sub_done

Data Passing

Sub-workflow inputs are passed via params, and sub-workflow outputs are exported back to the parent workflow via options.exposes:

steps:
    - id: step1
      uses: acts.core.subflow
      params:
        to: sub_workflow_id
        input_value: '${{ parent_var }}'
      options:
        exposes:
          - name: result

Action

Use acts.core.action to execute engine commands, such as triggering errors, completing steps, etc.

steps:
    - id: step1
      uses: acts.core.irq
      params:
        key: act1
      timeouts:
        # Trigger error on timeout
        - uses: acts.core.action
          if: $cost_in('8s')
          params:
            action: error
            options:
              ecode: err_timeout

Supported Commands

Client Commands

The client can also use do_action2 to perform the following operations to affect activity state:

#![allow(unused)]
fn main() {
// Complete activity
rt.do_action2(&pid, &tid, EventAction::Next, Vars::new()).unwrap();

// Trigger error
let mut options = Vars::new();
options.set("ecode", "err1");
rt.do_action2(&pid, &tid, EventAction::Error, options).unwrap();

// Back to specified step
let mut options = Vars::new();
options.set("to", "step1");
rt.do_action2(&pid, &tid, EventAction::Back, options).unwrap();
}

Code

Use acts.transform.code to execute JavaScript code (QuickJS engine) for variable computation, data transformation, and conditional logic within a workflow.

steps:
    - id: step1
      uses: acts.transform.code
      params: |
        let x = $get("a");
        let y = $get("b");
        $set("sum", x + y);
        $set("message", "Result: " + (x + y));

Built-in Functions

Use Cases

Variable computation:

- uses: acts.transform.code
  params: |
    let count = $get("count") || 0;
    $set("count", count + 1);

Array operations:

- uses: acts.transform.code
  params: |
    let a = ["u1", "u2"];
    let b = ["u2", "u3"];
    $set("merged", a.concat(b));

Conditional checks and errors:

- uses: acts.transform.code
  params: |
    if ($get("status") != "ok") {
      $set("ecode", "invalid_status");
    }

Examples

The following examples demonstrate various usage scenarios of the acts workflow engine.

Basic Examples

Interaction Examples

Error & Timeout

Advanced Features

Plugin Examples

Running Examples

# Run approval process example
cargo run --example approve

# Run error handling example
cargo run --example catches

# Run timeout handling example
cargo run --example timeout