Highway DSL

Highway DSL is a Python-based domain-specific language for defining production-grade workflows with Temporal-style coordination patterns. It is part of the larger Highway project, an advanced workflow engine capable of running complex DAG-based workflows with durability guarantees.

Version 3.0.0 removes the highway_dsl.mcp_client_server module, which was a client for the retired DSL generator web service. The DSL language, its operators and its JSON and YAML formats are unchanged from 2.2.0, so workflows written for 2.2.0 load and serialise the same way in 3.0.0.

Version 2.2.0 - LTS Stable Release

This is a Long-Term Support (LTS) release consolidating all features from the 1.x series into a stable, production-ready API:

New Features & Fixes

1. JoinOperator for Explicit Coordination (Temporal-Style)

Replaces brittle TriggerRule-based joins with explicit coordination semantics:

from highway_dsl import WorkflowBuilder, JoinMode

builder = WorkflowBuilder("data_pipeline")
builder.task("start", "setup.init")

# Create parallel branches
builder.task("branch_a", "process.a", dependencies=["start"])
builder.task("branch_b", "process.b", dependencies=["start"])
builder.task("branch_c", "process.c", dependencies=["start"])

# Explicit join - wait for all branches to succeed
builder.join(
    task_id="sync_gate",
    join_tasks=["branch_a", "branch_b", "branch_c"],
    join_mode=JoinMode.ALL_SUCCESS,  # Fail fast if any branch fails
    description="Wait for all extractions"
)

builder.task("finalize", "process.merge", dependencies=["sync_gate"])

Join Modes:

• JoinMode.ALL_OF - Wait for all tasks to finish (any final state)
• JoinMode.ANY_OF - Complete when first task finishes (race condition)
• JoinMode.ALL_SUCCESS - Wait for all tasks to succeed (fail fast)
• JoinMode.ONE_SUCCESS - Complete when one task succeeds (fallback pattern)

2. Universal Result Storage

All operators can now store results in workflow context (not just TaskOperator):

# Now works for all operator types
builder.wait("sleep_5s", wait_for=Duration.seconds(5), result_key="sleep_result")
builder.emit_event("notify", event_name="done", result_key="emit_result")
builder.wait_for_event("wait", event_name="start", result_key="event_data")

2. Callback Validation

Prevents runtime errors from typos in callback references:

builder.task("main", "func").on_success("handler_task")
workflow = builder.build()  # Now validates "handler_task" exists!

3. Duration Helper Class

Simplified time duration creation without datetime imports:

from highway_dsl import Duration

builder.wait("pause", wait_for=Duration.minutes(30))
builder.retry(delay=Duration.seconds(5))
builder.timeout(timeout=Duration.hours(1))

4. Workflow Metadata

Set workflow description and version programmatically:

builder = WorkflowBuilder("my_workflow", version="2.0.0")
builder.set_description("Production ETL pipeline")
builder.set_version("2.1.0")  # Update version later

5. Reflexive Operator (Sherlock Pattern)

Implements the Reflexive Loop pattern (Generate -> Verify -> Self-Correct) in a single atomic step:

builder.reflexive(
    task_id="gen_code",
    generator="tools.llm.call",
    verifier="tools.python.run",
    max_turns=3
)

6. depends_on Alias

More intuitive alias for the dependencies parameter:

builder.task("extract", "etl.extract", result_key="data")
builder.task("transform", "etl.transform", depends_on=["extract"])  # More readable

7. Fork/Join Pattern with parallel_with_join

Simplified fork/join parallel execution with automatic join barrier:

builder.parallel_with_join(
    "data_processing",
    branches={
        "process_users": lambda b: b.task("users", "process.users"),
        "process_orders": lambda b: b.task("orders", "process.orders"),
    },
    timeout_seconds=120,
).task(
    "aggregate",  # Automatically depends on join task
    "tools.aggregate",
)

8. Per-Activity Circuit Breaker

Protect against cascading failures with per-activity circuit breakers:

from highway_dsl import CircuitBreakerPolicy

builder.task(
    "call_external_api",
    "api.fetch_data",
    circuit_breaker_policy=CircuitBreakerPolicy(
        failure_threshold=5,      # Open after 5 failures
        success_threshold=2,      # Close after 2 successes
        isolation_duration=timedelta(seconds=30),
    )
)

Breaking Changes

• Default workflow version changed from "1.1.0" to "1.3.0"
• WaitOperator now serializes durations as ISO 8601 format (PT<seconds>S) instead of duration:<seconds>

---

Version 1.1.0 - Feature Release

This major feature release adds Airflow-parity features to enable production-grade workflows:

New Features

1. Scheduling Metadata (Airflow Parity)

Define cron-based schedules directly in your workflow:

builder = (
    WorkflowBuilder("daily_pipeline")
    .set_schedule("0 2 * * *")  # Run daily at 2 AM
    .set_start_date(datetime(2025, 1, 1))
    .set_catchup(False)
    .add_tags("production", "daily")
    .set_max_active_runs(1)
)

This is the beginning of the README. Read the rest at the original. Read the full README ↗