StemWorkflowApp class

Helper that bootstraps a workflow runtime on top of StemApp.

This wrapper wires together broker/backend infrastructure, registers flows, and exposes convenience helpers for scheduling and observing workflow runs without having to manage WorkflowRuntime directly.

Implemented types

• WorkflowCaller
• WorkflowEventEmitter
• StemTaskApp

Available extensions

• TaskEnqueuerBuilderExtension

Properties

app → StemApp

Underlying Stem app used for broker/worker coordination.

final

eventBus → EventBus

Event bus used to deliver workflow events.

final

hashCode → int

The hash code for this object.

no setter, inherited

isRuntimeStarted → bool

Whether the workflow runtime has been started.

no setter

isStarted → bool

Whether both the runtime and managed worker have been started.

no setter, override

isWorkerStarted → bool

Whether the managed worker has been started.

no setter

ownsStemApp → bool

Whether this wrapper owns the provided app and may shut it down.

final

runtime → WorkflowRuntime

Workflow runtime responsible for executing workflow runs.

final

runtimeType → Type

A representation of the runtime type of the object.

no setter, inherited

store → WorkflowStore

Store backing workflow run persistence.

final

Methods

close() → Future<void>

Alias for shutdown.

emitEvent<T>(WorkflowEventRef<T> event, T value) → Future<void>

Emits a typed event through a WorkflowEventRef.

override

emitJson<T extends Object>(String topic, T payloadJson, {String? typeName}) → Future<void>

Emits a DTO-backed external event without requiring a manual payload map.

emitValue<T>(String topic, T value, {PayloadCodec<T>? codec}) → Future<void>

Emits a typed event to resume runs waiting on topic.

override

emitVersionedJson<T extends Object>(String topic, T payloadJson, {required int version, String? typeName}) → Future<void>

Emits a DTO-backed external event and stores a schema version beside the JSON payload.

enqueue(String name, {Map<String, Object?> args = const {}, Map<String, String> headers = const {}, TaskOptions options = const TaskOptions(), DateTime? notBefore, Map<String, Object?> meta = const {}, TaskEnqueueOptions? enqueueOptions}) → Future<String>

Enqueue a task by name.

override

enqueueCall<TArgs, TResult>(TaskCall<TArgs, TResult> call, {TaskEnqueueOptions? enqueueOptions}) → Future<String>

Enqueue a typed task call.

override

enqueueJson<T extends Object>(String name, T argsJson, {Map<String, String> headers = const {}, TaskOptions options = const TaskOptions(), DateTime? notBefore, Map<String, Object?> meta = const {}, TaskEnqueueOptions? enqueueOptions, String? typeName}) → Future<String>

Available on TaskEnqueuer, provided by the TaskEnqueuerBuilderExtension extension

Enqueues a name-based task from a DTO that already exposes toJson().

enqueueValue<T>(String name, T value, {PayloadCodec<T>? codec, Map<String, String> headers = const {}, TaskOptions options = const TaskOptions(), DateTime? notBefore, Map<String, Object?> meta = const {}, TaskEnqueueOptions? enqueueOptions}) → Future<String>

Enqueue a dynamic-name task using a typed value plus optional codec.

override

enqueueVersionedJson<T extends Object>(String name, T argsJson, {required int version, Map<String, String> headers = const {}, TaskOptions options = const TaskOptions(), DateTime? notBefore, Map<String, Object?> meta = const {}, TaskEnqueueOptions? enqueueOptions, String? typeName}) → Future<String>

Available on TaskEnqueuer, provided by the TaskEnqueuerBuilderExtension extension

Enqueues a name-based task from a DTO and persists a schema version beside the JSON payload.

executeRun(String runId) → Future<void>

Executes the workflow run identified by runId.

getGroupStatus(String groupId) → Future<GroupStatus?>

Reads the latest group status by group id.

override

getRun(String runId) → Future<RunState?>

Returns the current RunState of a workflow run, or null if not found.

getTaskStatus(String taskId) → Future<TaskStatus?>

Reads the latest task status by task id.

override

listRunViews({String? workflow, WorkflowStatus? status, int limit = 50, int offset = 0}) → Future<List<WorkflowRunView>>

Returns normalized workflow run views filtered by workflow/status.

listWatchers(String topic) → Future<List<WorkflowWatcher>>

Lists event watchers registered for topic.

noSuchMethod(Invocation invocation) → dynamic

Invoked when a nonexistent method or property is accessed.

inherited

registerFlow(Flow<Object?> flow) → void

Registers flow into this app's workflow registry.

registerFlows(Iterable<Flow<Object?>> flows) → void

Registers flows into this app's workflow registry.

registerModule(StemModule module) → void

Registers all tasks and workflows from module into this app.

registerModules(Iterable<StemModule> modules) → void

Registers all tasks and workflows from modules into this app.

registerScript(WorkflowScript<Object?> script) → void

Registers script into this app's workflow registry.

registerScripts(Iterable<WorkflowScript<Object?>> scripts) → void

Registers scripts into this app's workflow registry.

registerWorkflow(WorkflowDefinition<Object?> definition) → void

Registers definition into this app's workflow registry.

registerWorkflows(Iterable<WorkflowDefinition<Object?>> definitions) → void

Registers definitions into this app's workflow registry.

resumeDueRuns([DateTime? now]) → Future<List<String>>

Marks all runs due at now as resumed and returns their ids.

rewindToCheckpoint(String runId, String checkpointName) → Future<void>

Rewinds runId to checkpointName and marks it runnable again.

shutdown() → Future<void>

Stops the runtime, worker, and disposes associated resources.

override

start() → Future<void>

Starts the workflow runtime and the underlying Stem worker.

override

startRuntime() → Future<void>

Starts the workflow runtime without starting the managed worker.

startWorker() → Future<void>

Starts the managed worker used for workflow execution.

startWorkflow(String name, {Map<String, Object?> params = const {}, String? parentRunId, Duration? ttl, WorkflowCancellationPolicy? cancellationPolicy}) → Future<String>

Schedules a workflow run.

startWorkflowCall<TParams, TResult extends Object?>(WorkflowStartCall<TParams, TResult> call) → Future<String>

Schedules a workflow run from a prebuilt WorkflowStartCall.

override

startWorkflowJson<T extends Object>(String name, T paramsJson, {String? parentRunId, Duration? ttl, WorkflowCancellationPolicy? cancellationPolicy, String? typeName}) → Future<String>

Starts a workflow from a DTO that already exposes toJson().

startWorkflowRef<TParams, TResult extends Object?>(WorkflowRef<TParams, TResult> definition, TParams params, {String? parentRunId, Duration? ttl, WorkflowCancellationPolicy? cancellationPolicy}) → Future<String>

Schedules a workflow run from a typed WorkflowRef.

override

startWorkflowValue<T>(String name, T value, {PayloadCodec<T>? codec, String? parentRunId, Duration? ttl, WorkflowCancellationPolicy? cancellationPolicy}) → Future<String>

Starts a workflow from a typed value plus optional codec.

startWorkflowVersionedJson<T extends Object>(String name, T paramsJson, {required int version, String? parentRunId, Duration? ttl, WorkflowCancellationPolicy? cancellationPolicy, String? typeName}) → Future<String>

Starts a workflow from a DTO and stores a schema version beside the JSON payload.

toString() → String

A string representation of this object.

inherited

viewCheckpoints(String runId) → Future<List<WorkflowCheckpointView>>

Returns persisted checkpoint views for runId.

viewRun(String runId) → Future<WorkflowRunView?>

Returns the normalized run view for runId, or null if not found.

viewRunDetail(String runId) → Future<WorkflowRunDetailView?>

Returns the combined run + checkpoint detail view for runId.

waitForCompletion<T extends Object?>(String runId, {Duration pollInterval = const Duration(milliseconds: 100), Duration? timeout, T decode(Object? payload)?, T decodeJson(Map<String, dynamic> payload)?, T decodeVersionedJson(Map<String, dynamic> payload, int version)?}) → Future<WorkflowResult<T>?>

Polls the workflow store until the run reaches a terminal state.

waitForTask<TResult extends Object?>(String taskId, {Duration? timeout, TResult decode(Object? payload)?, TResult decodeJson(Map<String, dynamic> payload)?, TResult decodeVersionedJson(Map<String, dynamic> payload, int version)?}) → Future<TaskResult<TResult>?>

Waits for a task result by task id.

override

waitForWorkflowRef<TParams, TResult extends Object?>(String runId, WorkflowRef<TParams, TResult> definition, {Duration pollInterval = const Duration(milliseconds: 100), Duration? timeout}) → Future<WorkflowResult<TResult>?>

Waits for runId using the decoding rules from a WorkflowRef.

override

workflowManifest() → List<WorkflowManifestEntry>

Returns the manifest entries for workflows registered with this app.

Operators

operator ==(Object other) → bool

The equality operator.

inherited

Static Methods

create({StemModule? module, Iterable<StemModule> modules = const [], Iterable<WorkflowDefinition<Object?>> workflows = const [], Iterable<Flow<Object?>> flows = const [], Iterable<WorkflowScript<Object?>> scripts = const [], Iterable<TaskHandler<Object?>> tasks = const [], StemApp? stemApp, StemBrokerFactory? broker, StemBackendFactory? backend, WorkflowStoreFactory? storeFactory, WorkflowEventBusFactory? eventBusFactory, StemWorkerConfig workerConfig = const StemWorkerConfig(queue: 'workflow'), String? continuationQueue, String? executionQueue, Duration pollInterval = const Duration(milliseconds: 500), Duration leaseExtension = const Duration(seconds: 30), WorkflowRegistry? workflowRegistry, WorkflowIntrospectionSink? introspectionSink, TaskPayloadEncoderRegistry? encoderRegistry, TaskPayloadEncoder resultEncoder = const JsonTaskPayloadEncoder(), TaskPayloadEncoder argsEncoder = const JsonTaskPayloadEncoder(), Iterable<TaskPayloadEncoder> additionalEncoders = const [], bool ownsStemApp = false}) → Future<StemWorkflowApp>

Creates a workflow app with custom backends and factories.

fromClient({required StemClient client, StemModule? module, Iterable<StemModule> modules = const [], Iterable<WorkflowDefinition<Object?>> workflows = const [], Iterable<Flow<Object?>> flows = const [], Iterable<WorkflowScript<Object?>> scripts = const [], Iterable<TaskHandler<Object?>> tasks = const [], WorkflowStoreFactory? storeFactory, WorkflowEventBusFactory? eventBusFactory, StemWorkerConfig workerConfig = const StemWorkerConfig(queue: 'workflow'), String? continuationQueue, String? executionQueue, Duration pollInterval = const Duration(milliseconds: 500), Duration leaseExtension = const Duration(seconds: 30), WorkflowIntrospectionSink? introspectionSink}) → Future<StemWorkflowApp>

Creates a workflow app backed by a shared StemClient.

fromUrl(String url, {StemModule? module, Iterable<StemModule> modules = const [], Iterable<WorkflowDefinition<Object?>> workflows = const [], Iterable<Flow<Object?>> flows = const [], Iterable<WorkflowScript<Object?>> scripts = const [], Iterable<TaskHandler<Object?>> tasks = const [], Iterable<StemStoreAdapter> adapters = const [], StemStoreOverrides overrides = const StemStoreOverrides(), StemWorkerConfig workerConfig = const StemWorkerConfig(queue: 'workflow'), String? continuationQueue, String? executionQueue, bool uniqueTasks = false, Duration uniqueTaskDefaultTtl = const Duration(minutes: 5), String uniqueTaskNamespace = 'stem:unique', bool requireRevokeStore = false, RevokeStore? revokeStore, UniqueTaskCoordinator? uniqueTaskCoordinator, Duration pollInterval = const Duration(milliseconds: 500), Duration leaseExtension = const Duration(seconds: 30), WorkflowRegistry? workflowRegistry, WorkflowIntrospectionSink? introspectionSink, WorkflowEventBusFactory? eventBusFactory, TaskPayloadEncoderRegistry? encoderRegistry, TaskPayloadEncoder resultEncoder = const JsonTaskPayloadEncoder(), TaskPayloadEncoder argsEncoder = const JsonTaskPayloadEncoder(), Iterable<TaskPayloadEncoder> additionalEncoders = const []}) → Future<StemWorkflowApp>

Creates a workflow app from a single backend URL plus adapter wiring.

inMemory({StemModule? module, Iterable<StemModule> modules = const [], Iterable<WorkflowDefinition<Object?>> workflows = const [], Iterable<Flow<Object?>> flows = const [], Iterable<WorkflowScript<Object?>> scripts = const [], Iterable<TaskHandler<Object?>> tasks = const [], StemWorkerConfig workerConfig = const StemWorkerConfig(queue: 'workflow'), String? continuationQueue, String? executionQueue, Duration pollInterval = const Duration(milliseconds: 500), Duration leaseExtension = const Duration(seconds: 30), WorkflowRegistry? workflowRegistry, WorkflowIntrospectionSink? introspectionSink, TaskPayloadEncoderRegistry? encoderRegistry, TaskPayloadEncoder resultEncoder = const JsonTaskPayloadEncoder(), TaskPayloadEncoder argsEncoder = const JsonTaskPayloadEncoder(), Iterable<TaskPayloadEncoder> additionalEncoders = const []}) → Future<StemWorkflowApp>

Creates an in-memory workflow app (in-memory broker, backend, and store).