snakemake_bridge 0.2.0

Embed a real-time Snakemake workflow monitor in a Dart desktop app: WebSocket server, typed event models, state reducer and process launcher.

snakemake_bridge for Dart

Monitor Snakemake 9+ workflows in real time inside a Dart Desktop application.

The snakemake_bridge package opens a local WebSocket server, receives the events emitted by Snakemake and turns them into state that is easy to render: jobs, progress, DAG, logs and errors.

This package is the Dart half of the integration. The other half is the Python plugin snakemake-logger-plugin-dart, which must be installed in the same Python environment as Snakemake.

How it works

The Dart application is responsible for starting the server before the workflow. When launching Snakemake, it tells the plugin the port and an authentication token. The plugin streams events without blocking workflow execution.

If the connection drops, the plugin reconnects and resends the events that were missed. The server discards duplicates automatically.

Requirements

• Dart SDK ^3.5.0 (that is, >= 3.5.0 and < 4.0.0);
• Python >= 3.11;
• Snakemake >= 9.0;
• snakemake-logger-plugin-dart installed in the same Python environment as Snakemake.

The package uses dart:io and targets desktop or command-line applications: Linux, macOS and Windows. It declares no support for Android, iOS or Web — there is no snakemake process to launch on a phone, and Web has neither dart:io nor a local server to bind.

Installation

1. Install the Dart package

In your application's project:

dart pub add snakemake_bridge

Or add the dependency to pubspec.yaml by hand:

dependencies:
  snakemake_bridge: ^0.2.0

Then fetch the dependencies:

dart pub get

2. Install the Python plugin

Activate the Python environment Snakemake uses and run:

pip install snakemake-logger-plugin-dart

Confirm that Snakemake found the plugin:

snakemake --help | grep logger-dart

The command should list options such as --logger-dart-address. If it does not, check that the plugin and Snakemake were installed into the same Python environment.

Your first monitored workflow

The example below:

1. creates a secure token;
2. starts the local server on an available port;
3. folds the events into a WorkflowRunState;
4. launches Snakemake;
5. releases the resources at the end.

import 'dart:async';

import 'package:snakemake_bridge/snakemake_bridge.dart';

Future<void> main() async {
  final token = WorkflowServer.generateToken();
  final server = WorkflowServer(token: token);
  await server.start();

  print('Server listening on ${server.address}');

  final state = WorkflowRunState();
  final subscription = server.events.listen((event) {
    if (!state.apply(event)) return;

    print(
      'status=${state.status.name} '
      'jobs=${state.jobs.length} '
      'progress=${state.progress}',
    );
  });

  try {
    final runner = await SnakemakeRunner.start(
      workflowDir: '/path/to/workflow',
      serverPort: server.boundPort,
      token: token,
      cores: 4,
    );

    runner.stdout.listen((line) => print('[snakemake] $line'));
    runner.stderr.listen((line) => print('[snakemake:error] $line'));

    final exitCode = await runner.exitCode;
    print('Snakemake finished with exit code $exitCode');
  } finally {
    await subscription.cancel();
    await server.close();
  }
}

Replace /path/to/workflow with the directory containing the Snakefile.

An exit code of 0 means the Snakemake process finished successfully. Beyond that, state.status, state.progress, state.jobs, state.logLines and state.errors expose the state built from the events received.

Where the workflow writes

By default the workflow directory is also the working directory, so results/ and .snakemake/ are written inside the tree that holds the Snakefile. Pass workdir to separate them — one directory per run, with the workflow tree staying read-only and reusable:

final runner = await SnakemakeRunner.start(
  workflowDir: '/opt/myapp/workflow',   // where the Snakefile lives
  workdir: '/home/user/projects/p1/runs/2026-08-11',  // where it writes
  serverPort: server.boundPort,
  token: token,
);

Snakemake still discovers the Snakefile in workflowDir — all four accepted layouts (Snakefile, snakefile, workflow/Snakefile, workflow/snakefile) keep working — and only then changes into workdir.

A relative workdir resolves against workflowDir, not against the application's own working directory: the process starts in workflowDir so that discovery works. Passing 'runs/today' therefore writes into workflowDir/runs/today — inside the tree you were trying to keep clean. Use an absolute path unless you mean exactly that.

Controlling the process environment

environment and includeParentEnvironment are handed to Process.start, with the run token merged in on top. The defaults inherit the environment the application was launched with, which is what most callers want.

Pass includeParentEnvironment: false when the workflow must not see it. An application that gives each tool its own conda environment needs this: a PYTHONPATH, a LD_LIBRARY_PATH or an active conda env in the user's shell is inherited by Snakemake, then by every job it launches, and the wrong interpreter or shared library wins.

final runner = await SnakemakeRunner.start(
  workflowDir: '/opt/myapp/workflow',
  serverPort: server.boundPort,
  token: token,
  environment: {
    'PATH': '$envPrefix/bin:/usr/bin:/bin',
    'CONDA_PREFIX': envPrefix,
    'HOME': home,
  },
  includeParentEnvironment: false,
);

Two things worth knowing:

• executable is resolved against the PATH of the map you passed, not the parent's. 'snakemake' with $envPrefix/bin first therefore picks that environment's Snakemake — an absolute path is not required;
• the token is merged last, so a stale SNAKEMAKE_LOGGER_DART_TOKEN in environment cannot shadow the run's own. On Windows, where environment names are matched case-insensitively, that holds for an exact-case key only.

Updating a graphical interface

WorkflowRunState does not depend on any particular state management solution. It can be wrapped in a ChangeNotifier, Riverpod, Bloc or anything else.

apply() returns false for the events that never change the state — PongEvent, ResourcesInfoEvent and UnknownEvent — and true for the rest. It is a conservative filter rather than exact change detection: a HelloEvent from a reconnection to the same run, for instance, returns true without having changed anything.

server.events.listen((event) {
  if (state.apply(event)) {
    notifyUi();
  }
});

In practice, notifyUi() should be replaced by whatever mechanism the application uses, such as notifyListeners(), updating a Notifier or emitting a new state from a Bloc.

The properties a monitoring screen usually needs are:

Property	Contents
state.status	Overall state: idle, running, finished or failed.
state.progress	Progress between 0.0 and 1.0, when the total is known.
state.jobs	Jobs indexed by their numeric id.
state.rulegraph	Rule DAG sent by Snakemake.
state.logLines	Most recent log lines.
state.errors	Structured errors received during the run.

Cancelling the workflow

Keep the SnakemakeRunner instance around while the workflow runs, and call:

final exitCode = await runner.cancel();

The runner sends SIGTERM, and forces termination with SIGKILL if the process is still alive after 15 seconds. The interval can be changed:

await runner.cancel(killAfter: const Duration(seconds: 30));

A cancelled run does not reach a terminal state.status. WorkflowRunState leaves running when a bye arrives (finished) or an error does (failed), and a killed Snakemake sends neither — after a cancel() that escalated, state.status is still running while runner.exitCode is -9. Treat the future returned by cancel(), not the run state, as the end of the run:

final exitCode = await runner.cancel();
// state.status is still RunStatus.running here.
showCancelled();

What cancelling does not do

It does not stop jobs that are already running. Both signals reach the Snakemake process only, and neither makes Snakemake 9 signal its own jobs:

• on SIGTERM the scheduler takes its graceful path — it stops launching new jobs and logs "Will exit after finishing currently running jobs", then waits. A job with an hour left to run keeps that hour, and cancel() waits with it;
• once killAfter elapses, the SIGKILL kills Snakemake and every job it had launched is orphaned. Each one survives, runs to completion, and goes on writing into the working directory long after the returned future completed and the application considers the run dead.

SIGINT is not a way around this: the local executor's cancel() shuts down its thread pool, and every thread in that pool is blocked waiting on its job's process.

For short workflows this rarely shows. For one that shells out to a tool running for tens of minutes, it means a cancelled run keeps burning CPU and writing files.

Stopping the jobs as well means signalling the whole process group, which this package does not do. Snakemake starts its jobs in its own group, so a SIGTERM on the group does reach them — an application that has to guarantee nothing survives a cancellation can launch Snakemake under setsid itself and signal -pid:

executable cannot be setsid directly, because the Snakemake arguments are built right after it. Point it at a small wrapper instead:

#!/bin/sh
# snakemake-setsid.sh — new process group, same pid.
exec setsid snakemake "$@"

final runner = await SnakemakeRunner.start(
  workflowDir: workflowDir,
  serverPort: server.boundPort,
  token: token,
  executable: '/opt/myapp/snakemake-setsid.sh',
);

// Instead of runner.cancel(): reaches Snakemake and every job it started.
Process.killPid(-runner.process.pid, ProcessSignal.sigterm);

exec and the absence of --fork are what keep the pid the runner holds usable as the group id: the child of a Dart process is not a process group leader, so setsid replaces it in place rather than forking.

setsid is POSIX; it does not exist on Windows, and macOS does not ship the setsid(1) command. A first-class hook for this is planned — see issue #1.

Manual mode, in two terminals

You can also start just the example server and run Snakemake by hand. This is useful to validate the installation and watch the events as they arrive.

In the first terminal, inside this repository's dart/ directory:

SNAKEMAKE_LOGGER_DART_TOKEN=dev \
  dart run example/snakemake_bridge_example.dart

The program prints an address similar to:

LISTENING port=43127 address=ws://127.0.0.1:43127

In the second terminal, activate Snakemake's Python environment, change into the workflow directory and use the port printed by the first terminal:

export SNAKEMAKE_LOGGER_DART_TOKEN=dev

snakemake --cores 2 \
  --logger dart \
  --logger-dart-address ws://127.0.0.1:43127

The first terminal should then show events such as:

EVENT HelloEvent seq=0
EVENT WorkflowStartedEvent seq=1
EVENT ProgressEvent seq=8 1/3
EVENT ByeEvent seq=9
DONE status=finished jobs=3 progress=1.0

The dev token is only appropriate for this local test. In a real application, always use WorkflowServer.generateToken() and generate a fresh token per run.

Main components

File	Role
lib/src/workflow_server.dart	Embedded WebSocket server (shelf), bound to 127.0.0.1, validates the token, deduplicates by seq, requests replay on every reconnection.
lib/src/workflow_events.dart	Typed event models (sealed classes, no codegen; unknown types become UnknownEvent).
lib/src/workflow_state.dart	Reducer: event stream → queryable run state (jobs, progress, DAG, logs, errors).
lib/src/snakemake_runner.dart	Launches and owns the snakemake process (cancellation = SIGTERM on the scheduler; running jobs are not signalled).
example/snakemake_bridge_example.dart	Minimal headless "app" for testing the bridge from the terminal.

WorkflowServer

Embedded WebSocket server that:

• listens on 127.0.0.1 only;
• validates the plugin's token;
• converts the JSON messages into typed Dart events;
• discards duplicates by sequence number;
• requests a resend of events after a reconnection.

WorkflowRunState

Folds the event stream into a queryable state. This pattern is also known as a reducer: each new event is applied on top of the current state.

SnakemakeRunner

Starts the snakemake process, wires up the connection to the server and exposes:

• standard output on runner.stdout;
• error output on runner.stderr;
• the process result on runner.exitCode;
• cancellation of the scheduler through runner.cancel(), which does not stop jobs that are already running.

Typed events

Incoming events are represented by classes such as WorkflowStartedEvent, JobInfoEvent, JobStartedEvent, JobFinishedEvent, ProgressEvent, LogLineEvent and WorkflowErrorEvent.

Types the installed version of the package does not know yet are delivered as UnknownEvent, so the application keeps working when the protocol gains new events.

If nothing shows up in the application

Check, in this order:

1. that the address passed to --logger-dart-address carries the right port;
2. that the application and the plugin use exactly the same token;
3. that snakemake --help | grep logger-dart finds the plugin;
4. that the plugin was installed in the same Python environment as the snakemake executable;
5. that the Dart server was started before the workflow.

For safety, the server only accepts local connections on 127.0.0.1. It is not designed to receive events directly from another machine.

Development and tests

Inside the dart/ directory:

dart pub get
dart analyze
dart test

Next steps

• See the full example.
• Read the architecture to understand authentication, reconnection and failure modes.
• Use the protocol specification to implement another compatible client or server.
• See the repository installation guide to run the end-to-end validation from source.