Stem

Stem is an experimental Dart-first background job and workflow platform: enqueue work, run workers, and orchestrate durable workflows while the runtime contracts and adapter guarantees continue to harden.

For full docs, API references, and in-depth guides, visit https://kingwill101.github.io/stem.

For production-shaped task definitions, prefer stem_builder generated typed definitions. Manual TaskDefinition<TArgs, TResult> is the supported typed fallback. Low-level map handlers live under package:stem/advanced.dart for transport integrations and migrations.

Packages

Package	Description	pub.dev
`stem`	Core runtime: contracts, worker, scheduler, in-memory adapters, signals, Canvas, workflows
`stem_cli`	Command-line tooling (`stem` executable) and CLI utilities
`stem_memory`	Compatibility package for the explicit `package:stem/memory.dart` in-memory library
`stem_sqlite`	SQLite broker and result backend for local dev/testing
`stem_redis`	Redis Streams broker, result backend, and watchdog helpers
`stem_postgres`	Postgres broker, result backend, and scheduler stores
`stem_flutter`	Adapter-neutral Flutter helpers for mobile worker isolates and queue monitoring
`stem_flutter_sqlite`	Flutter SQLite runtime/storage helpers for mobile Stem apps
`stem_builder`	Build-time code generator for annotated tasks and workflows
`stem_adapter_tests`	Shared contract test suites for adapter implementations
`stem_dashboard`	Hotwire-based operations dashboard (experimental)	—

Features

Task pipeline - enqueue with delays, priorities, idempotency helpers, and retries.
Workers - isolate pools with soft/hard time limits, autoscaling, and remote control (stem worker ping|revoke|shutdown).
Scheduling - Beat-style scheduler with interval/cron/solar/clocked entries and drift tracking.
Workflows - Durable Flow runtime with pluggable stores (in-memory, Redis, Postgres, SQLite) and CLI introspection via stem wf.
Observability - Dartastic OpenTelemetry metrics/traces, heartbeats, CLI inspection (stem observe, stem dlq).
Security - Payload signing (HMAC or Ed25519), TLS automation scripts, revocation persistence.
Adapters - In-memory drivers included here; Redis Streams and Postgres adapters ship via the stem_redis and stem_postgres packages.
Quality tooling - Package-level format, analysis, test, adapter contract, chaos, benchmark, and standalone-resolution gates in CI.

Install

dart pub add stem
# Optional adapters
dart pub add stem_redis     # Redis broker/backend
dart pub add stem_postgres  # Postgres broker/backend
dart pub add stem_sqlite    # SQLite broker/backend
dart pub add -d stem_builder # for annotations/codegen (optional)
dart pub add -d stem_cli      # for CLI tooling

New application code can import package:stem/stable.dart. The historical package:stem/stem.dart barrel remains available for compatibility, while custom transports and instrumentation can use package:stem/advanced.dart.

For local development and tests, import the in-memory implementations explicitly with package:stem/memory.dart; the stem_memory package remains as a compatibility export for existing applications.

Examples

StemApp and StemWorkflowApp never start their managed worker implicitly. Call start() explicitly when the process is intended to consume work. StemWorkflowApp also exposes startRuntime() and startWorker() when you want those lifecycles split.

Minimal in-memory task + worker

import "dart:async";
import "package:stem/stable.dart";

class HelloArgs {
  const HelloArgs({required this.name});

  final String name;

  Map<String, dynamic> toJson() => {"name": name};

  factory HelloArgs.fromJson(Map<String, dynamic> json) =>
      HelloArgs(name: json["name"] as String);
}

final helloDefinition = TaskDefinition<HelloArgs, void>.json(
  name: "demo.hello",
  decodeArgsJson: HelloArgs.fromJson,
);

final helloTask = helloDefinition.handler(
  entrypoint: (context, args) async => print("Hello ${args.name}"),
);

Future<void> main() async {
  final client = await StemClient.inMemory(tasks: [helloTask]);
  final worker = await client.createWorker();
  unawaited(worker.start());

  await client.enqueueCall(
    helloDefinition.buildCall(const HelloArgs(name: "Stem")),
  );
  await Future<void>.delayed(const Duration(seconds: 1));

  await worker.shutdown();
  await client.close();
}

Reusable stack from URL (Redis)

import "package:stem/stable.dart";
import "package:stem_redis/stem_redis.dart";

Future<void> main() async {
  final client = await StemClient.fromUrl(
    "redis://localhost:6379",
    adapters: const [StemRedisAdapter()],
    overrides: const StemStoreOverrides(
      backend: "redis://localhost:6379/1",
    ),
    tasks: [HelloTask()],
  );

  final worker = await client.createWorker();
  unawaited(worker.start());

  await client.enqueueValue("demo.hello", const {"name": "Redis"});
  await Future<void>.delayed(const Duration(seconds: 1));

  await worker.shutdown();
  await client.close();
}

Typed task definition and waiting for result

class HelloArgs {
  const HelloArgs({required this.name});
  final String name;

  Map<String, dynamic> toJson() => {"name": name};
  factory HelloArgs.fromJson(Map<String, dynamic> json) =>
      HelloArgs(name: json["name"] as String);
}

final helloDefinition = TaskDefinition<HelloArgs, String>.json(
  name: "demo.hello2",
  decodeArgsJson: HelloArgs.fromJson,
  metadata: const TaskMetadata(description: "typed hello task"),
);

final helloTask = helloDefinition.handler(
  entrypoint: (context, args) async => "Hello ${args.name}",
);

Future<void> main() async {
  final client = await StemClient.inMemory(tasks: [helloTask]);
  final worker = await client.createWorker();
  unawaited(worker.start());

  final result = await helloDefinition.enqueueAndWait(
    client,
    const HelloArgs(name: "Typed"),
  );
  print(result?.value);

  await worker.shutdown();
  await client.close();
}

Workflow quick-start (Flow)

import "package:stem/stable.dart";

final onboardingFlow = Flow<String>(
  name: "demo.onboarding",
  build: (flow) {
    flow.step("welcome", (ctx) async {
      return "Welcome ${ctx.requiredParam<String>("name")}";
    });
    flow.step("done", (ctx) async => "Done");
  },
);

Future<void> main() async {
  final appClient = await StemClient.inMemory();
  final app = await appClient.createWorkflowApp(
    flows: [onboardingFlow],
  );
  await app.start();

  final ref = onboardingFlow.refJson(HelloArgs.fromJson);
  final runId = await ref.start(app, params: const HelloArgs(name: "Stem"));
  final result = await ref.waitFor(app, runId);

  print(result?.value);
  await app.shutdown();
  await appClient.close();
}

Annotated workflow + task with `stem_builder`

import "package:stem/stable.dart";
import "package:stem_builder/stem_builder.dart";

part "definitions.stem.g.dart";

@WorkflowDefn(name: "builder.signup", kind: WorkflowKind.script)
class BuilderSignupWorkflow {
  Future<String> run(String email) async {
    final userId = await createUser(email);
    await finalizeSignup(userId: userId);
    return userId;
  }

  @WorkflowStep(name: "create-user")
  Future<String> createUser(String email) async {
    return "user-$email";
  }

  @WorkflowStep(name: "finalize")
  Future<void> finalizeSignup({required String userId}) async {}
}

@TaskDefn(name: "builder.send_welcome")
Future<void> sendWelcomeEmail(
  String email, {
  TaskExecutionContext? context,
}) async {
  // optional: use context for logger/meta/retry helpers
}

dart run build_runner build

# After generation, use module + generated defs

// example usage after codegen
final client = await StemClient.inMemory(module: stemModule);
final app = await client.createWorkflowApp();
await app.start();

final runId = await StemWorkflowDefinitions.builderSignup.startAndWait(
  app,
  "alice@example.com",
);
final result = await StemWorkflowDefinitions.builderSignup.waitFor(app, runId);
print(result?.value); // {user: alice@example.com}

Workflow with multiple worker queues

import "package:stem/stable.dart";
import "package:stem/advanced.dart";

final onboardingFlow = Flow<Map<String, String>>(
  name: "workflow.multi_workers",
  build: (flow) {
    flow.step("dispatch", (ctx) async {
      final notifyTaskId = await ctx.enqueue(
        "notify.send",
        args: {"email": "alex@example.com"},
        enqueueOptions: const TaskEnqueueOptions(queue: "notifications"),
      );
      final analyticsTaskId = await ctx.enqueue(
        "analytics.track",
        args: {"userId": "alex", "event": "account.created"},
        enqueueOptions: const TaskEnqueueOptions(queue: "analytics"),
      );
      return {"notifyTaskId": notifyTaskId, "trackTaskId": analyticsTaskId};
    });
  },
);

class NotifyTask extends TaskHandler<String> {
  @override
  String get name => "notify.send";

  @override
  TaskOptions get options => const TaskOptions(queue: "notifications");

  @override
  Future<String> call(TaskContext context, Map<String, Object?> args) async =>
      "notified:${args['email']}";
}

class AnalyticsTask extends TaskHandler<String> {
  @override
  String get name => "analytics.track";

  @override
  TaskOptions get options => const TaskOptions(queue: "analytics");

  @override
  Future<String> call(TaskContext context, Map<String, Object?> args) async =>
      "tracked:${args['event']}";
}

Future<void> main() async {
  final client = await StemClient.inMemory();
  final app = await client.createWorkflowApp(
    flows: [onboardingFlow],
    workerConfig: const StemWorkerConfig(queue: "workflow"),
  );
  await app.start();

  final notifications = await client.createWorker(
    workerConfig: StemWorkerConfig(
      queue: "notifications-worker",
      consumerName: "notifications-worker",
      subscription: RoutingSubscription.singleQueue("notifications"),
    ),
    tasks: [NotifyTask()],
  );
  final analytics = await client.createWorker(
    workerConfig: StemWorkerConfig(
      queue: "analytics-worker",
      consumerName: "analytics-worker",
      subscription: RoutingSubscription.singleQueue("analytics"),
    ),
    tasks: [AnalyticsTask()],
  );

  await notifications.start();
  await analytics.start();

  final result = await onboardingFlow.startAndWait(app);
  final taskIds = result?.value ?? const <String, String>{};
  print(await app.waitForTask<String>(taskIds['notifyTaskId']!));
  print(await app.waitForTask<String>(taskIds['trackTaskId']!));

  await notifications.shutdown();
  await analytics.shutdown();
  await app.close();
  await client.close();
}

5) CLI at a glance

# Start a worker or run built-in introspection commands
stem --help
stem worker start --help
stem wf --help

General worker management (multi-worker setup)

import "package:stem/stable.dart";
import "package:stem/advanced.dart";

class EmailTask extends TaskHandler<void> {
  @override
  String get name => "notify.send";

  @override
  TaskOptions get options => const TaskOptions(queue: "notify");

  @override
  Future<void> call(TaskContext context, Map<String, Object?> args) async {
    print("notify queue: ${args['to']}");
  }
}

class ReportTask extends TaskHandler<void> {
  @override
  String get name => "reports.aggregate";

  @override
  TaskOptions get options => const TaskOptions(queue: "reports");

  @override
  Future<void> call(TaskContext context, Map<String, Object?> args) async {
    print("reports queue: ${args['reportId']}");
  }
}

Future<void> main() async {
  final client = await StemClient.inMemory();

  final notifyWorker = await client.createWorker(
    workerConfig: StemWorkerConfig(
      queue: "notify-worker",
      consumerName: "notify-worker",
      subscription: RoutingSubscription.singleQueue("notify"),
    ),
    tasks: [EmailTask()],
  );

  final reportsWorker = await client.createWorker(
    workerConfig: StemWorkerConfig(
      queue: "reports-worker",
      consumerName: "reports-worker",
      subscription: RoutingSubscription.singleQueue("reports"),
    ),
    tasks: [ReportTask()],
  );

  await notifyWorker.start();
  await reportsWorker.start();

  await client.enqueue(
    "notify.send",
    args: {"to": "ops@example.com"},
  );
  await client.enqueue(
    "reports.aggregate",
    args: {"reportId": "r-2026-q1"},
  );

  await Future<void>.delayed(const Duration(milliseconds: 400));

  await notifyWorker.shutdown();
  await reportsWorker.shutdown();
  await client.close();
}

Full example that combines a workflow dispatching to dedicated workers: multiple_workers.dart

Want depth?

This README is intentionally example-focused. For implementation details, runtime semantics, adapter tuning, and operational playbooks, see the full docs at https://kingwill101.github.io/stem.

Documentation & Examples

Guided onboarding: Guided onboarding (install → infra → ops → production).
Examples (each has its own README):
workflows - end-to-end workflow samples (in-memory, sleep/event, SQLite, Redis). See versioned_rewind.dart for auto-versioned step rewinds.
multiple_workers.dart - workflow dispatching tasks to notifications and analytics workers.
cancellation_policy - demonstrates auto-cancelling long workflows using WorkflowCancellationPolicy.
rate_limit_delay - delayed enqueue, priority clamping, Redis rate limiter.
dlq_sandbox - dead-letter inspection and replay via CLI.
autoscaling_demo - autoscaling worker concurrency under queue backlog.
scheduler_observability - Beat drift metrics, schedule signals, and CLI checks.
microservice, monolith_service, mixed_cluster - production-style topologies.
progress_heartbeat - task progress + heartbeat reporting.
task_context_mixed - TaskContext + TaskInvocationContext enqueue patterns, plus Celery-style apply_async options.
task_usage_patterns.dart - in-memory TaskContext enqueue, TaskInvocationContext builder, and typed enqueue calls.
worker_control_lab - worker ping/stats/revoke/shutdown drills.
unique_tasks - enables TaskOptions.unique with a shared lock store.
signing_key_rotation - rotate HMAC signing keys with overlap.
ops_health_suite - CLI health checks + queue/worker snapshots.
security examples - payload signing + TLS profiles.
postgres_tls - Redis broker + Postgres backend secured via the shared STEM_TLS_* settings.
otel_metrics - OTLP collectors + Grafana dashboards.

Stem

Packages

Features

Install

Examples

Minimal in-memory task + worker

Reusable stack from URL (Redis)

Typed task definition and waiting for result

Workflow quick-start (Flow)

Annotated workflow + task with `stem_builder`

Workflow with multiple worker queues

5) CLI at a glance

General worker management (multi-worker setup)

Want depth?

Documentation & Examples

Libraries

stem package

Stem

Packages

Features

Install

Examples

Minimal in-memory task + worker

Reusable stack from URL (Redis)

Typed task definition and waiting for result

Workflow quick-start (Flow)

Annotated workflow + task with stem_builder

Workflow with multiple worker queues

5) CLI at a glance

General worker management (multi-worker setup)

Want depth?

Documentation & Examples

Libraries

stem package

Annotated workflow + task with `stem_builder`