aion/activity

Typed activity values and policy configuration.

Types

Activity

opaque </>

A typed activity invocation value.

i is the statically-known input type and o is the statically-known output type. The input and output codecs are carried so workflow dispatch can encode the author value and decode the type-erased engine payload without reflection.

pub opaque type Activity(i, o)

Backoff

</>

Backoff strategy carried with an explicit retry policy.

The SDK only stores this configuration for the engine to interpret during dispatch. It does not apply retries or invent missing timing values.

pub type Backoff {
  Exponential(
    initial: duration.Duration,
    multiplier: Float,
    max: duration.Duration,
  )
  Linear(
    initial: duration.Duration,
    increment: duration.Duration,
    max: duration.Duration,
  )
  Fixed(delay: duration.Duration)
}

Constructors

Exponential(
  initial: duration.Duration,
  multiplier: Float,
  max: duration.Duration,
)

Exponential backoff from initial, scaled by multiplier, capped at max.

Linear(
  initial: duration.Duration,
  increment: duration.Duration,
  max: duration.Duration,
)

Linear backoff from initial, adding increment, capped at max.

```
Fixed(delay: duration.Duration)
```
Fixed backoff with the same delay between attempts.

Declaration

opaque </>

A type-erased activity declaration: the single per-activity artifact an author writes.

declare captures the typed input and output TypeRefs — so the declaration’s contract is checked by gleam build — and erases them to the names the generator needs. Erasure at the boundary is what lets a package’s activities, which have different input and output types, live together in one List(Declaration) (mirroring Aion’s type-erased event history). From this declaration aion generate derives the activity.new wrapper, the value-type codec pairs, the worker handler stub, the registration entry, the workflow.toml entry, and (for remote tiers) the wire-compat golden. No retry/timeout/heartbeat lives here: absence is intentional (ADR-001), so it is structurally impossible for codegen to emit a policy the author did not choose.

pub opaque type Declaration

RetryPolicy

</>

Explicit retry policy for an activity.

No default retry policy is baked into the SDK. An activity built with new and no retry decorator carries no policy and runs exactly once when the engine dispatches it.

pub type RetryPolicy {
  RetryPolicy(max_attempts: Int, backoff: Backoff)
}

Constructors

RetryPolicy(max_attempts: Int, backoff: Backoff)

Tier

</>

Where an activity’s side-effecting body executes.

The tier is authored data, not an invented default. aion generate reads it to decide which worker handler stub and registration entry to emit, and whether a wire-compat golden is generated (remote tiers only). Since the in-VM dispatch tier landed, the tier ALSO rides the runtime Activity value as an optional selection (see execution_tier): Some(InVm) routes the dispatch through the engine’s linked in-VM child-process path, while absence (None) — and every remote tier — keeps today’s remote worker wire. The tier is still never recorded in history: like the dispatcher choice, it is routing, not workflow-visible nondeterminism.

pub type Tier {
  InVm
  RemotePython
  RemoteRust
}

Constructors

```
InVm
```
Runs in-process inside the BEAM VM via a registered NIF.
```
RemotePython
```
Runs in a remote Python worker over the worker protocol.
```
RemoteRust
```
Runs in a remote Rust worker over the worker protocol.

TypeRef

opaque </>

A typed binding of a value type’s name to its codec.

declare takes one for the input and one for the output. The codec makes the binding type-check — type_ref("OrderInput", order_input_codec()) only compiles when the codec is a codec.Codec(OrderInput) — so a mismatched type and codec is a gleam build failure. The type_name is the handle the out-of-process aion generate extractor maps to schemas/<type>.json to drive codec and worker codegen; codecs carry no type information at runtime, so the name is supplied explicitly here.

pub opaque type TypeRef(a)

Values

declaration_input_type

</>

pub fn declaration_input_type(declaration: Declaration) -> String

Return the input value type name of a declared activity.

declaration_name

</>

pub fn declaration_name(declaration: Declaration) -> String

Return the engine-facing name of a declared activity.

declaration_output_type

</>

pub fn declaration_output_type(
  declaration: Declaration,
) -> String

Return the output value type name of a declared activity.

declaration_tier

</>

pub fn declaration_tier(declaration: Declaration) -> Tier

Return the tier a declared activity runs on.

declare

</>

pub fn declare(
  name: String,
  tier: Tier,
  input: TypeRef(i),
  output: TypeRef(o),
) -> Declaration

Declare an activity from its name, tier, and typed input/output references.

This is the only per-activity artifact an author hand-writes; the plumbing is generated from it. The side-effecting body (the runner) is written separately and referenced by the generated wrapper — codegen never synthesizes a body (the determinism boundary is unchanged).

execution_tier

</>

pub fn execution_tier(
  activity: Activity(i, o),
  tier: Tier,
) -> Activity(i, o)

Select the execution tier this activity dispatches on.

InVm routes the dispatch through the engine’s in-VM path: the activity’s runner executes ONCE, live, inside a linked child process of the workflow process — no remote worker, no task-queue routing — while its recorded history (ActivityScheduled/ActivityStarted/terminal) and replay semantics stay byte-identical to a remote dispatch. Remote tiers (RemotePython/RemoteRust) keep the remote worker wire; selecting one is equivalent to selecting nothing today.

Absence is intentional data, exactly like retry/timeout/heartbeat: an activity built with new and no execution_tier decorator carries no selection (None) and dispatches on the remote wire — today’s exact behavior. Later calls replace earlier values; the SDK does not merge.

heartbeat

</>

pub fn heartbeat(
  activity: Activity(i, o),
  heartbeat_interval: duration.Duration,
) -> Activity(i, o)

Attach an explicit heartbeat interval to an activity.

heartbeat_interval

</>

pub fn heartbeat_interval(
  activity: Activity(i, o),
) -> option.Option(duration.Duration)

Return the explicitly attached heartbeat interval, if one exists.

input

</>

pub fn input(activity: Activity(i, o)) -> i

Return the typed input captured by the activity value.

input_codec

</>

pub fn input_codec(activity: Activity(i, o)) -> codec.Codec(i)

Return the typed input codec captured by the activity value.

label

</>

pub fn label(
  activity: Activity(i, o),
  key: String,
  value: String,
) -> Activity(i, o)

Attach a display label to an activity.

Labels are human-meaningful key/value hints (for example #("brief", "IP-001") or #("repo", "ablative-io/yggdrasil")) that ride with the dispatch to the worker and surface in its logs and the dashboard. They are display metadata only: the engine never interprets them and they never affect routing, replay, or the recorded history. Repeated calls accumulate in call order; nothing is deduplicated or overwritten.

labels

</>

pub fn labels(
  activity: Activity(i, o),
) -> List(#(String, String))

Return the display labels attached to the activity, in call order.

name

</>

pub fn name(activity: Activity(i, o)) -> String

Return the activity name used by the engine dispatch boundary.

new

</>

pub fn new(
  name: String,
  input: i,
  input_codec: codec.Codec(i),
  output_codec: codec.Codec(o),
  run: fn(i) -> Result(o, error.ActivityError),
) -> Activity(i, o)

Build a typed activity value with no retry, timeout, or heartbeat config.

Absence of config is intentional data: there are no hidden defaults. In particular, an activity with no retry decorator carries no policy and an activity with no execution_tier decorator dispatches on the remote wire.

node

</>

pub fn node(
  activity: Activity(i, o),
  name: String,
) -> Activity(i, o)

Pin this activity’s dispatch to a specific node (per-activity affinity).

A node is a concrete worker host inside the activity’s (namespace, task_queue) pool. Pinning narrows the dispatch from “any worker in the pool” to that one host — for an activity that must run where its state or hardware lives, for example.

Affinity is OPTIONAL and has NO workflow-level default: an activity built with new and no node decorator carries no pin (None), so the engine dispatches it to any worker in the pool. There is no precedence to resolve — the activity either pins a node or it does not. Later calls replace earlier values; the SDK does not merge.

output_codec

</>

pub fn output_codec(activity: Activity(i, o)) -> codec.Codec(o)

Return the typed output codec captured by the activity value.

retry

</>

pub fn retry(
  activity: Activity(i, o),
  policy: RetryPolicy,
) -> Activity(i, o)

Attach an explicit retry policy to an activity.

Later calls replace earlier retry policy values; the SDK does not merge or synthesize policy fields.

retry_policy

</>

pub fn retry_policy(
  activity: Activity(i, o),
) -> option.Option(RetryPolicy)

Return the explicitly attached retry policy, if one exists.

runner

</>

pub fn runner(
  activity: Activity(i, o),
) -> fn(i) -> Result(o, error.ActivityError)

Return the typed runner captured by the activity value.

selected_node

</>

pub fn selected_node(
  activity: Activity(i, o),
) -> option.Option(String)

Return the explicitly pinned per-activity node affinity, if one exists.

Absence (None) means no pin was set, so the engine dispatches the activity to any worker in its (namespace, task_queue) pool. Unlike the task queue, node affinity has no workflow-level default: None is final.

selected_task_queue

</>

pub fn selected_task_queue(
  activity: Activity(i, o),
) -> option.Option(String)

Return the explicitly selected per-activity task queue, if one exists.

Absence (None) means no override was set, so the engine resolves the dispatch to the workflow-level default, else the named "default" queue.

selected_tier

</>

pub fn selected_tier(
  activity: Activity(i, o),
) -> option.Option(Tier)

Return the explicitly selected execution tier, if one exists.

Absence (None) means no execution_tier decorator was applied, so the dispatch takes the remote worker wire — today’s exact behavior.

task_queue

</>

pub fn task_queue(
  activity: Activity(i, o),
  name: String,
) -> Activity(i, o)

Select the task queue this activity is dispatched on (per-activity override).

The task queue is the routing pool inside the workflow’s namespace that a worker subscribes to; selecting it lets one workflow mix activities across pools (for example a "norn" step and a "gpu" step). This is the highest-precedence selection: it overrides any workflow-level default.

Absence is intentional data, exactly like retry/timeout/heartbeat: an activity built with new and no task_queue decorator carries no selection, so the engine resolves it to the workflow-level default when one is set, else the named "default" task queue. Later calls replace earlier values; the SDK does not merge.

tier_to_string

</>

pub fn tier_to_string(tier: Tier) -> String

Render a tier as the canonical string the generator’s wire form uses.

timeout

</>

pub fn timeout(
  activity: Activity(i, o),
  timeout_duration: duration.Duration,
) -> Activity(i, o)

Attach an explicit timeout duration to an activity.

timeout_duration

</>

pub fn timeout_duration(
  activity: Activity(i, o),
) -> option.Option(duration.Duration)

Return the explicitly attached timeout duration, if one exists.

type_ref

</>

pub fn type_ref(
  type_name: String,
  value_codec: codec.Codec(a),
) -> TypeRef(a)

Bind a value type’s name to its codec for use in an activity declaration.

type_ref_codec

</>

pub fn type_ref_codec(reference: TypeRef(a)) -> codec.Codec(a)

Return the codec captured by a type reference.