Programmatic Dynamic DAG

What this workflow does

This workflow demonstrates the runtime DAG-mutation API of horus-runtime. A stage does more than read and write artifacts. It generates the rest of the graph while it runs, from data that it sees only at run time.

The plan stage reads a small JSON dataset. It discovers how many distinct groups the dataset holds. From inside its own function, it adds one processing stage per group and one fan-in stage. None of those stages exist when the workflow starts.

There is no science and no YAML here. This is the pure-Python builder path. The pattern to read is HorusContext.get_context().workflow, then add_task and expand.

The problem it solves

A declarative map: block fans out over a static collection. Some fan-outs are not static. The shape of the graph is a decision that a stage makes.

Two common cases are "one stage per file format actually present in this upload" and "one stage per cluster returned by a discovery stage". You cannot write those stage lists in advance. You only know them after the first stage runs.

Without a mutation API, you split the pipeline in two. You run the first part, read the result, generate a second workflow file, and run that. You now own a generator, two runs, and a manual join between them.

Horus keeps it in one run. The running BaseWorkflow instance is reachable from any stage function. workflow.add_task(task) adds one stage at a time. That is the granular pattern. No edge is needed, because the new stage is created during the plan run, so it can only run after plan finishes. workflow.expand(tasks= [...], edges=[...]) adds the combine stage together with all of its fan-in edges in one call.

The inputs of combine are not known until run time. Its function signature uses **kwargs. The python_function runtime passes every declared input and output through when a function declares **kwargs.

Pipeline

plan                       examples/dataset.json (6 records, 3 groups) ──► (no static output)
   │  reads the dataset, discovers groups {alpha, beta, gamma}, then AT RUNTIME:
   │    - adds one process_<group> task per group  (workflow.add_task, one at a time)
   │    - adds a combine task + all its fan-in edges in one call (workflow.expand)
   ▼
process_alpha  process_beta  process_gamma     (generated tasks, run after plan)
   │  each summarizes its group's records       ──► results/<group>.json
   └──────────────┬──────────────┘
                   ▼
combine                     results/alpha.json + beta.json + gamma.json ──► results/combined.json
   │  sums count/sum across every generated group task

Inputs and outputs

Input

• examples/dataset.json: 6 records. Each record is {"group": ..., "n": ...}. The records span 3 groups: alpha has 2, beta has 3, gamma has 1.

Outputs land in results/:

• results/<group>.json: {"group", "count", "sum", "values"} for that group.
• results/combined.json: {"total_count": 6, "total_sum": 30, "per_group": {...}} for the bundled dataset. The sums are 8, 12, and 10.

Run the workflow

Install uv if you do not have it:

curl -LsSf https://astral.sh/uv/install.sh | sh

Then install the horus-runtime and run the workflow:

cd workflows/engine-showcases/w02-programmatic-dynamic-dag
uv sync
uv run python run.py

The entrypoint is run.py. There is no workflow.yaml, because the graph is built and mutated in Python.

To change the graph shape, add or remove records in examples/dataset.json. The plan stage adapts the generated stage count to the groups that are present. You change no code.

References

• Horus repositories