5.5. Activities

Activities

An Activity is a Stateflows behavior modeled as a UML Activity: a graph of nodes connected by control and object flows. Actions perform work, control nodes route or synchronize execution, and typed tokens carry data. Activities can also react to events while event-accepting nodes are active.

flowchart LR
    start((●)) --> read(Read order)
    read -->|Order| valid{Has order lines?}
    valid -->|nonempty| ship(Reserve stock)
    valid -->|else| reject(Reject order)
    ship --> done(((◎)))
    reject --> done

Loading

The diagram illustrates routing: rounded boxes represent actions, the diamond represents a decision, and the circles mark entry and completion. See the node pages for execution rules.

The implementation is in the Activities core, with examples of supported behavior in the Activity integration tests.

Browse Concepts for token handling, streaming, and recovery, or Building Blocks for nodes and flows.

Define and register an Activity

The lambda builder style defines nodes by names and callbacks. AddAction takes the node name, its asynchronous action, and an optional builder for outgoing flows. Object flows declare the token type they carry; control flows carry no application data.

using Stateflows;
using Stateflows.Activities;

builder.Services.AddStateflows(stateflows => stateflows
    .AddActivities(activities => activities
    .AddActivity("OrderFulfilment", activity => activity
        .AddInitial(node => node.AddControlFlow("LoadOrder"))
        .AddAction(
            "LoadOrder",
            async c => c.Output(await orders.LoadAsync(
                orderId, c.CancellationToken)),
            node => node.AddFlow<Order>("CheckStock"))
        .AddDecision<Order>("CheckStock", decision => decision
            .AddFlow("ReserveStock", flow => flow
                .AddGuard(async c => c.Token.Lines.Count > 0))
            .AddElseFlow("RejectOrder"))
        .AddAction(
            "ReserveStock",
            async c => await stock.ReserveAsync(
                c.GetTokensOfType<Order>().Single(), c.CancellationToken),
            node => node.AddControlFlow<FinalNode>())
        .AddAction("RejectOrder", async c =>
            await orders.RejectAsync(orderId, c.CancellationToken),
            node => node.AddControlFlow<FinalNode>())
        .AddFinal())));

For larger definitions, implement IActivity and define the graph in its static Build(IActivityBuilder) method, then register the type with AddActivity<TActivity>(). See Behaviors for behavior registration, identity, and event handling.

Typed runtime elements can inject application services, IActivityContext, and common behavior/execution contexts. Node activation supplies INodeContext; flow guard/transformation activation supplies IFlowContext instead. Graph exception-handler nodes additionally receive IExceptionContext. Input/output token interfaces are injectable into workers as well. These execution services are not available to the static graph-building method. See Dependency Injection for the availability table and examples.

How execution works

Declare supported boundary types with AddInputPin<TToken> and AddOutputPin<TToken>. Input pin flows use pin.AddFlow(target) without repeating the token type; producers target OutputPin<TToken>. Each type has a distinct pin at the root or inside a structured, iterative, or parallel scope. Pin-only inputs reject the entire unsupported token batch before initialization or streaming admission; a graph with no input pins rejects every nonempty token batch. Structural boundary object flows must match declared input/output pins. Effect Activities invoked with AddEffectActivity receive the transition event as an input token and must declare its matching AddInputPin<TEvent>, even when the pin has no outgoing flow. ICurrentNodeContext exposes scoped InputPins and OutputPins metadata. Root output pins publish raw notifications; nested output pins route tokens to their containing node's parent flows. See Input Pins and Output Pins.

Initialization executes Initial nodes and follows enabled flows. Nodes activate according to their incoming flows, node options, and event triggers. Action nodes receive routed tokens, execute, and may output new tokens. A Final Node terminates the current Activity execution. Activities that should remain reactive can instead leave event-accepting nodes active.

Ordinary event delivery is run-to-completion: the caller waits while the event's enabled data-flow work executes until there is nothing more to process. Waiting for a future event or missing token is not executing work, so an initialized Activity can remain reactive between calls.

Streaming on Orleans

An Activity can also process a live stream while accepting further token batches and regular steering events. Send input with StreamInputAsync to start or join this mode; no graph-wide opt-in is required:

var admission = await activity.StreamInputAsync(input => input.Add(workItem));
if (admission.Status != EventStatus.Accepted)
    throw new InvalidOperationException($"Admission failed: {admission.Status}.");

// Persist a firing-boundary snapshot without sending application input.
var checkpoint = await activity.CheckpointAsync();
if (checkpoint.Status != EventStatus.Consumed)
    throw new InvalidOperationException($"Checkpoint failed: {checkpoint.Status}.");

This is an Orleans-only Stateflows extension, not a change to the default execution mode or an additional UML node type. Both modes share one behavior instance and network. Regular transactions remain ordered and nonoverlapping, but a regular call can hand mixed-token descendants to streaming rather than waiting for them. Values are shared live state without rollback during the streaming busy period; values and marking are saved at natural quiescence or an explicit checkpoint. Accepted is volatile admission, not completion or durability.

Read Activity Streaming before relying on its concurrency, publication, cancellation, and failure guarantees. Activity Checkpoints explains barriers and restart behavior.

Control flow carries execution only. Object flow carries typed values. An object-flow guard can filter each token and a transformation can convert it while traversing that edge. See Flows, Control Flow, Data Flow, and Tokens.

Building blocks

Purpose	Node or concept
Entry, completion, and nested boundaries	Initial, Final, Input, Output
Work and routing	Action, Decision, Merge
Branching and synchronization	Fork, Join, Implicit fork and join
Nested execution	Structured, Iterative, Parallel Activity Nodes
Retained data and events	Data Store, Accept Event, Send Event, Publish Event, and Time Event action nodes

For node naming, typed node definitions, and available node types, see Nodes.

Activities also support built-in, scope-aware graph exception handling through node-level handlers. See Activity Exception Handling.

Interacting with Activities

Locate an Activity by its name and string instance ID with IActivityLocator, then send it an event through the behavior handle:

using System;
using System.Threading.Tasks;
using Stateflows;
using Stateflows.Activities;

public sealed record StartOrder;

public sealed class OrderService(IActivityLocator locator)
{
    public async Task StartAsync(string orderId)
    {
        if (!locator.TryLocateActivity(
                new ActivityId("OrderFulfilment", orderId),
                out var activity))
        {
            throw new InvalidOperationException(
                $"Activity for order '{orderId}' could not be located.");
        }

        await activity.SendAsync(new StartOrder());
    }
}

The located handle implements IActivityBehavior, which extends IBehavior. Its additions cover Activity status, token input, and raw output notifications:

Method	Result	Purpose
GetStatusAsync(headers)	Task<RequestResult<ActivityInfo>>	Query lifecycle status, expected events, and active nodes. Headers are optional.
WatchStatusAsync(Action<ActivityInfo>)	Task<IWatcher>	Watch published Activity status notifications.
SendInputAsync(tokensAction, headers)	Task<RequestResult<TokensOutput>>	Build an input batch containing one or more token types, with optional headers.
SendInputAsync<TToken>(params TToken[] tokens)	Task<RequestResult<TokensOutput>>	Submit a batch of one token type.
StreamInputAsync(tokensAction, headers)	Task<SendResult>	Orleans-only nonblocking token admission. Accepted does not mean completed or durable.
CheckpointAsync(headers)	Task<SendResult>	Orleans-only standalone checkpoint; Consumed means the snapshot committed. Does not implicitly initialize the Activity.
GetOutputAsync<TEvent>(since)	Task<IEnumerable<TEvent>>	Get latest retained output, optionally including available history since a timestamp.
GetOutputAsync(notificationNames, since)	Task<IEnumerable<EventHolder>>	Retrieve notification envelopes for specified event names.
WatchOutputAsync<TEvent>(Action<TEvent>)	Task<IWatcher>	Receive each raw output payload of the declared notification type.
WatchOutputAsync(notificationNames, Action<EventHolder>)	Task<IWatcher>	Watch envelopes for specified notification names.

Query and watch active nodes

ActivityInfo extends BehaviorInfo with a typed ActivityId and ActiveNodes, an IReadOnlyTree<string> describing the active node configuration. The inherited BehaviorStatus and ExpectedEvents describe lifecycle status and expected event names.

The following examples assume an IActivityBehavior activity has been located:

using Stateflows.Common;

var result = await activity.GetStatusAsync();
if (result.Status != EventStatus.Consumed || result.Response is null)
{
    throw new InvalidOperationException(
        $"Could not query the Activity: {result.Status}.");
}

foreach (var node in result.Response.ActiveNodes.GetAllNodes())
{
    Console.WriteLine(node.Value);
}

await using var statusWatcher = await activity.WatchStatusAsync(
    (ActivityInfo info) => Console.WriteLine(info.BehaviorStatus));

Keep the handle typed as IActivityBehavior to obtain ActivityInfo; IBehavior.GetStatusAsync requests general BehaviorInfo. The explicitly typed watch callback selects the Activity-specific overload. Status notifications are snapshots published when the engine reports a state change, not a trace of each node invocation.

Send input tokens

Token input enters through the Activity's declared input pins. The graph must route the supplied token types to consumers; sending tokens does not bypass flows or inject them into arbitrary action nodes. See Input Pin and Tokens.

For a graph that accepts decimal tokens:

var result = await activity.SendInputAsync<decimal>(10m, 25m);
if (result.Status != EventStatus.Consumed)
{
    throw new InvalidOperationException(
        $"The Activity did not consume input: {result.Status}.");
}

For a mixed-type batch, use the builder overload:

var result = await activity.SendInputAsync(input => input
    .Add("order-42")
    .AddRange(10m, 25m));

ITokensInput.Add<TToken> adds one token; AddRange<TToken> adds several. The builder overload also accepts an optional header dictionary. The generic params overload has no headers parameter.

Input may initialize an Activity or activate its input pins in an already-running graph. Root output pins publish notifications independently; their payloads are not response tokens returned by SendInputAsync. Do not treat a successful send as eventual completion of an interactive Activity. Use output watches or retained-output queries for results.

While an Orleans stream is active, regular token requests (SendInputAsync/RequestAsync with token input) are rejected. Use StreamInputAsync for admission, or SendAsync(new TokensInput().Add(...)) for a regular transactional send; consume results through watches or output queries. A regular send can return after its work is handed off to streamed descendants, so its return is not an eventual-output guarantee.

Query output notifications

Root output pins publish each arriving token as a raw notification. GetOutputAsync uses notification retrieval, not an Activity-specific token archive.

var latest = await activity.GetOutputAsync<decimal>();
foreach (var total in latest)
{
    Console.WriteLine(total);
}

var history = await activity.GetOutputAsync<decimal>(since);
var envelopes = await activity.GetOutputAsync(
    new[] { Event<decimal>.Name }, since);

Without a timestamp, only the latest unexpired retained notification is returned. With a timestamp, matching unexpired history is included alongside that retained value. Default output pins (Retain = false, TTL 0) keep no history, so queries return no past output. Configure publication on Output Pins when results must remain requestable. Requests do not clear notifications or wait for future output or Activity completion.

Watch new output

await using var outputWatcher = await activity.WatchOutputAsync<decimal>(
    total => Console.WriteLine(total));

The named watch overload receives EventHolder envelopes. Typed callbacks receive one payload at a time, not a batch.

Publication uses each pin's declared event type, without synthesizing subtype, base-type, or interface notifications. Register the watcher before the input or event that may produce output. See notification retention for history, expiry, and subscription retry semantics.

Status and output watches return IWatcher. They are volatile client watches, not durable behavior-to-behavior subscriptions; keep the handle and watcher alive, then dispose the watcher when finished.

Common Behavior operations

The handle also supports SendAsync, RequestAsync, general notification watches, ResetAsync, and FinalizeAsync. Event-accepting nodes determine which application events can drive the running graph. See Messaging Patterns for shared delivery and subscription semantics, and Observability for internal execution hooks.