🌊 hexaflow

Lightweight, embeddable Python workflow engine with stages, steps, splits, joins, and checkpointed resumption.

🌊 Foundational DAG engine powering Hexastack, Hexaqueue, and Hexaqual · 🛡️ Governed by Hexaqual

🎯 Vision & Architectural Intent

hexaflow is a zero-daemon, localhost-first workflow engine for Python developers who need resilient, multi-stage, multi-step execution graphs without the operational overhead of running heavy external schedulers (Airflow, Temporal, Prefect).

Key Features

• Hierarchical Execution: Workflows structured cleanly into Stages and Steps.
• DAG Splits & Joins: Concurrently fan-out (split) and synchronize at barrier dependencies (join).
• Pluggable Graph Analysis & Multi-Format Renderers: Structural reachability (ancestors, descendants, blast_radius), critical path latency bottleneck calculation, and export to Mermaid, Graphviz DOT, ASCII, JSON, SVG, and PNG.
• Dry-Run Simulation Mode: Simulate workflows safely with synthetic mocks (dry_run=...), side-effect fail-closed guards (side_effects=True, allow_unsafe), and chaos fault injection without mutating databases or external APIs.
• Workflow Memento & Time-Travel Rewind: Capture immutable state mementos (to_memento(), restore_from_memento()), rewind execution to before any historical step (rewind_to()), and replay downstream dependencies seamlessly.
• Fault Partitioning: Transient retries with exponential backoff vs. permanent failure suspension (SUSPENDED).
• Step Middleware Pipeline: Onion-layered execution interceptors (StepMiddlewarePort, InOutStepMiddleware) supporting structured logging (LoggingStepMiddleware), high-precision latency metrics (TimingStepMiddleware), and whole-system simulation mock fallbacks (MockFallbackMiddleware).
• Checkpointed Resumption: Automatically saves step inputs and outputs into an embedded SQLite store. Resume failed workflows from the point of failure without re-running completed steps.
• Dual DSL: Decorator-based syntax for Pythonic simplicity (@wf.stage, @wf.step) alongside declarative classes for programmatic pipelines.

📦 Installation

pip install hexaflow

Or using uv:

uv add hexaflow

⚡ Quickstart

from hexaflow import RetryPolicy, StageExecutionMode, Workflow

wf = Workflow(name="order_pipeline", version="1.0.0")


# Stage 1: Validation
@wf.stage("validation")
@wf.step("validate_cart", retries=RetryPolicy(max_attempts=3))
def validate_cart(ctx) -> dict:
    return {"order_id": "ord_101", "total_usd": 150.00}


# Stage 2: Parallel Processing (Split / Fan-Out)
@wf.stage("processing", execution_mode=StageExecutionMode.CONCURRENT_ALL)
@wf.step("authorize_payment", depends_on=["validate_cart"])
def authorize_payment(ctx) -> dict:
    cart = ctx.inputs["validate_cart"]
    return {"status": "PAID", "amount": cart["total_usd"]}


@wf.stage("processing")
@wf.step("reserve_inventory", depends_on=["validate_cart"])
def reserve_inventory(ctx) -> dict:
    return {"warehouse": "US-EAST-1", "reserved": True}


# Stage 3: Fulfillment (Join Barrier)
@wf.stage("fulfillment")
@wf.step("create_shipping_label", depends_on=["authorize_payment", "reserve_inventory"])
def create_shipping_label(ctx) -> dict:
    return {"tracking_id": "TRK-9812739"}


if __name__ == "__main__":
    state = wf.run()
    print(f"Workflow {state.run_id} completed with status: {state.status.value}")

🖥️ Command Line Interface (CLI)

hexaflow includes a fast, zero-daemon CLI (hexaflow or hf):

# Run a workflow directly
hf run examples/order_pipeline.py:wf

# Check run status & step timing
hf status <run_id>

# Inspect step inputs, outputs, or error tracebacks
hf inspect <run_id> create_shipping_label

# Visualize workflow DAG (Mermaid, DOT, ASCII, JSON)
hf graph render examples/order_pipeline.py:wf --format mermaid

# Render a GitHub Actions workflow DAG directly
hf graph render .github/workflows/ci.yml --format mermaid

# Simulate dynamic workflows in-memory and render materialized DAG
hf graph render pipelines/ci.py:make_pipeline --dry-run --params '{"affected": ["core"]}'

# Display topological metrics, critical path, and estimated latency
hf graph info examples/order_pipeline.py:wf

# Resume a suspended workflow from its latest checkpoint
hf resume <run_id> examples/order_pipeline.py:wf

# Clear checkpoints and restart a workflow from scratch
hf restart <run_id> examples/order_pipeline.py:wf

# Abort a workflow and execute compensating rollback actions
hf abort <run_id> examples/order_pipeline.py:wf

📖 Full Command Catalog: See USAGE.md for the complete CLI subcommand options, arguments, and formatting flags.

🏛️ Ecosystem Alignment

hexaflow is part of the Hexa architectural ecosystem:

• hexaflow (this repository): Lightweight, zero-daemon, localhost-first DAG workflow engine.
• hexastack: Monorepo framework providing CQRS, Event Sourcing, FastAPI, and out-of-the-box DevTools.
• hexaqueue: Flagship distributed batch & HPC cluster scheduler.

📄 License

Apache 2.0. See LICENSE for details.