flux-workflow

flux-workflow

An embeddable, durable DAG workflow engine for Go.

flux-workflow runs directed-acyclic-graph workflows and owns their runtime state itself. You embed it as a library, point it at a database, register your tools and workflows, and submit tasks — the engine handles execution, persistence, crash recovery, and human-in-the-loop suspension. It is designed for AI / tool orchestration pipelines but knows nothing about any particular domain.

rt, _ := runtime.NewLocal("state.db", runtime.WithLocalTool(myTool))
rt.RegisterWorkflow(ctx, def)
res, _ := rt.Run(ctx, def, map[string]any{"name": "flux"})
fmt.Println(res.Status) // success

Why

Closest relatives: DBOS Transact (durable execution as a library), LangGraph (graph + checkpointer + interrupts), Temporal/Cadence (signals ≈ resume, event history ≈ persistent events) — but those are either not Go, or require a server.

Install

go get github.com/tuxi/flux-workflow

Requires Go 1.25+.

Quickstart

A complete, self-contained example (SQLite + in-memory queue, no external services) lives in examples/quickstart:

go run ./examples/quickstart

[sync ] task=... status=success output={"final":{...,"extras":{"greeting":"Hello, flux!"}}}
[async] task=... status=success output={"final":{...,"extras":{"greeting":"Hello, async!"}}}

It defines a greet tool, wires a start → greet → end workflow, and runs it both synchronously (Run) and asynchronously via background workers (Start + Submit).

The Runtime API

runtime.Runtime is the single entry point. Everything you need is a method on it.

Construction

// Local mode — self-contained: SQLite (WAL), in-memory queue / job queue / lock,
// no external dependencies. Ideal for embedding, tests, and single-process apps.
rt, err := runtime.NewLocal("state.db", runtime.WithLocalTool(myTool))

// Server mode — you inject the infrastructure.
rt, err := runtime.New(
    runtime.WithDB(pgDB),          // *gorm.DB (Postgres or SQLite)
    runtime.WithQueue(redisQ),     // repository.TaskQueue
    runtime.WithJobQueue(jobQ),    // engine.AsyncJobQueue
    runtime.WithEventBus(bus),     // *eventbus.EventBus
    runtime.WithLock(dLock),       // lock.DistributedLock
    runtime.WithTool(myTool),
)

Lifecycle & execution

Worker concurrency is tunable:

rt.Start(ctx,
    runtime.WithTaskWorkers(4),
    runtime.WithAsyncWorkers(4),
    runtime.WithAwaitPollWorkers(1),
)

Building a service layer on top

The facade covers the common path. To build your own HTTP/service layer, reach the underlying handles and use the core packages directly:

db  := rt.DB()            // *gorm.DB — construct any query repo from repository/query
bus := rt.EventBus()      // publish custom events / attach listeners
eng := rt.Engine()        // power-user ops: Replay, Redo, Cancel, Fork, ...
reg := rt.NodeRegistry()  // register custom node types

The runtime facade is deliberately thin: advanced operations stay on the core packages (engine, repository, eventbus) rather than being mirrored here.

Human-in-the-loop

An await node suspends the task. When the external event arrives, resume it:

res, _ := rt.Run(ctx, def, input)   // res.Status == "suspended"
// ... webhook / approval arrives ...
rt.Resume(ctx, res.TaskID, "wait_approval", map[string]any{"approved": true})

Workflows & node types

A workflow is a definition.WorkflowDefinition (from github.com/tuxi/flux): Nodes, Edges, and an Output mapping. Built-in node types:

Node inputs are expressions (input.name, nodes.greet.output.greeting); the workflow Output maps node outputs into the final result. Register your own node types via rt.NodeRegistry().

Architecture

runtime/                one-stop facade (this is the public API)
engine/                 DAG execution, resume, retry, replay, fork, cancel
workflow/, workflow/nodes/  builder + node implementations
repository/, repository/query/  storage interfaces + GORM implementation
eventbus/               layered event routing (transient / persistent / audit)
worker/                 task, async, await-poll workers + recovery scanner
domain/                 core entities

The core is dependency-light: go list -deps ./runtime pulls no web framework, cloud SDK, or LLM client. Redis-backed drivers are optional subpackages so pure local embedding never links go-redis:

• engine/redisjobqueue — Redis Stream AsyncJobQueue
• repository/query/redisqueue — Redis TaskQueue
• pkg/lock/redislock — Redsync DistributedLock

Status

flux-workflow was extracted from a production AI product and is being hardened for general use. The engine core and the runtime facade are stable; some higher-level HTTP/presentation code is still being separated out. APIs may shift before a tagged release.

License

MIT