Writing workflows in YAML

YAML is the quickest way to describe a workflow when each task is a command or a script. This page is the practical reference for the schema.

Workflow

name: my_workflow        # required; letters, digits, spaces, _ and -
kind: horus_workflow     # required; selects the built-in workflow
tasks: []                # the list of tasks (below)
edges: []                # dependencies between tasks (below)
failure_policy: fail_fast  # optional; fail_fast (default) or continue

Failure policy

failure_policy controls what the workflow does when a task fails:

• fail_fast (default): the first task to fail cancels every other running task and fails the run immediately.
• continue: a failed task blocks only its own downstream tasks (they never run); every other branch of the DAG runs to completion. The run still ends as FAILED, and the failed tasks are reported when it finishes.

Either way a run with any failure ends FAILED. The policy only decides how much of the graph runs before that happens.

Task

tasks:
  - id: prep             # required; unique within the workflow
    name: Prepare data   # required; shown in the dashboard
    kind: horus_task     # required
    runtime: { ... }     # required; what to run
    executor: { ... }    # required; how to run it
    target: { ... }      # where to run it (defaults to local)
    inputs:  [ ... ]      # artifacts this task consumes
    outputs: [ ... ]      # artifacts this task produces
    resources: { ... }   # optional compute hints
    skip_if_complete: true   # skip when outputs already exist (default true)

Runtimes: what to run

runtime:
  kind: command
  command: "python train.py --data $dataset --out $model"

Executors: how to run

Targets: where to run

target: { kind: local, working_directory: "./horus-work" }

Artifacts: inputs and outputs

outputs:
  - { kind: file, id: model, path: "./out/model.pkl" }

Each artifact has an id (unique within the task's inputs or outputs) and a path. A task counts as complete when all of its outputs exist on disk.

An artifact may also carry labels, a free-form string: string mapping the runtime never reads:

outputs:
  - { kind: file, id: scored, path: batch_017.parquet, labels: { subject: batch_017, role: measurement } }

They exist for whatever reads the workflow afterwards, not for the run itself: a UI grouping nodes, or a lineage query answering "everything derived from subject X." A domain expresses its own structure through labels instead of the runtime growing a field per domain.

The directory containing each output is created before the task runs, so path: "./out/model.pkl" works even when out/ does not exist yet. No task needs a mkdir step of its own.

Resources (optional)

resources:
  cpus: 8
  gpus: 2
  memory_gb: 32
  vram_gb: 16
  walltime: "01:30:00"

These are advisory. Resource-aware targets use them; local ignores them, but the dashboard displays them.

Capacity: gating concurrency (optional)

Per-task resources say what a task wants. A workflow-level capacity block says how much a machine actually has, so a large fan-out never oversubscribes finite hardware (for example, only two GPUs means at most two GPU-requesting tasks run at once):

name: my_workflow
kind: horus_workflow
capacity:
  "local://my-host":   # keyed by the target's location id; targets that
                       # share a machine share one pool
    gpus: 2
    memory_gb: 64
tasks: [ ... ]

The key is the target's location id (for a local target that is local://<hostname>), so every target on the same machine draws from one pool. A task acquires its declared resources before it runs and releases them when it finishes; while a location is full, further tasks that need those dimensions wait. Only the dimensions you declare are gated; anything left out is treated as unconstrained. Capacity is opt-in: with no capacity (or a task with no resources, or a task on a location you did not list), tasks acquire immediately, exactly as before. max_concurrency still applies on top. A single task that asks for more than a location's total capacity fails fast rather than blocking forever.

Referencing artifacts in commands

Inside command, code, and args, use $-substitution (string.Template syntax) to reference paths:

So cat $raw > $clean reads the input artifact raw and writes the output artifact clean. Unknown $names are left untouched, and shell $(...) and {} pass through unchanged.

Edges: wiring the DAG

edges:
  - source: prep            # producer task id ...
    source_output: data     # ... its output artifact id ...
    target: train           # ... feeds this consumer task ...
    target_input: data      # ... as this input artifact id.

Edges are the only thing that orders tasks. The producer's output and the consumer's input may use different ids. Each consumer input can be fed by at most one edge, and the graph must be acyclic.

Ordering-only edges

An edge normally does two jobs: it orders the two tasks and it routes the producer's output into the consumer's input. Add transfer: false to keep the ordering but move no data:

edges:
  - { source: worker, source_output: part, target: assemble, target_input: kit, transfer: false }

The consumer still waits for the producer, but its input keeps whatever path it already has. Ordering-only edges are exempt from the "at most one edge per input" rule, so several of them can order-gate the same consumer input, alongside at most one regular (transferring) edge. This pattern is what lets the runtime wire expanding constructs into a live graph without disturbing their data paths.

Ordering tasks that have no artifacts

transfer: false still names an artifact on each end, so it cannot order a task that declares none — a cleanup step that just has to run last, say. Leave both artifact ids out for that:

edges:
  - source: prep       # prep must finish ...
    target: cleanup    # ... before cleanup starts, and neither task
                       #     needs to declare an artifact for the other.

Such an edge never carries data, so transfer does not apply to it (it is false either way), and any number of them can point at the same task.

Name both artifact ids or neither. Naming only one is an error (IncompleteEdgeError) rather than a silent downgrade to ordering-only, so a mistyped data edge cannot quietly stop transferring. The source must still be a real task: a root artifact (artifact-<rootId>) has nothing to order against.

Subworkflows

A task can embed a whole child workflow instead of running a single command. The child is inlined into the parent's DAG rather than run as an opaque sub-process, so its individual tasks show up in the dashboard and its artifacts are ordinary artifacts.

tasks:
  - id: sub
    sub:
      kind: horus_workflow
      name: child
      artifacts:
        - { kind: file, id: seed, path: seed_in.txt }
      tasks:
        - id: upper
          name: Uppercase
          kind: horus_task
          target: { kind: local }
          runtime: { kind: command, command: "tr a-z A-Z < $seed > $upped" }
          executor: { kind: shell }
          inputs:  [ { kind: file, id: seed,  path: seed_in.txt } ]
          outputs: [ { kind: file, id: upped, path: upped.txt } ]
        - id: report
          name: Report
          kind: horus_task
          target: { kind: local }
          runtime: { kind: command, command: "cat $upped > $report_out" }
          executor: { kind: shell }
          inputs:  [ { kind: file, id: upped,      path: upped.txt } ]
          outputs: [ { kind: file, id: report_out, path: report.txt } ]
      edges:
        - { source: artifact-seed, source_output: seed, target: upper, target_input: seed }
        - { source: upper, source_output: upped, target: report, target_input: upped }

The sub: key holds a complete child workflow document. It lowers to a native kind: subworkflow task (body: <the sub: document>) before the rest of the schema is parsed, so port_overrides, max_depth, and target are written as sibling keys of sub:, not inside it. to_yaml always dumps the native kind: subworkflow form, never the sub: sugar.

Ports: how data crosses the boundary

There is no binding table to fill in — the interface is derived from the child itself:

• In-ports are the child's own root artifacts. From the parent's side, feed one by name through the subworkflow task's inputs, exactly like any other task input (seed above).
• Out-ports are child outputs that no inner edge consumes, named by artifact id (or taskid.artifactid if two leaves happen to share an id). report_out above is the child's only out-port.
• port_overrides: { derived_name: parent_facing_name } renames a port for the parent-facing schema.

Each port is backed by a placeholder artifact so the subworkflow task looks like an ordinary task to the parent DAG. Edges that cross the boundary must be ordering-only (transfer: false); a transferring edge across the boundary raises SubworkflowError. When you use the sub: sugar, any boundary edge you write is downgraded to ordering-only automatically, and the real data edge is rewired directly to the inlined inner task at validation time — you never bind to the placeholder yourself.

Nesting

Inlined task and edge ids are prefixed <subId>/<innerId>, and this nests naturally for subworkflows within subworkflows (outer/inner/report). Inner ids may not themselves contain /, since that prefix is reserved.

max_depth (default 10) caps how deeply subworkflows may nest, so a body that embeds itself fails fast with SubworkflowError instead of expanding forever. The child is otherwise validated exactly like a top-level workflow: duplicate ids, unresolved edges, and cycles are all rejected the same way.

The Python guide covers the same feature through wf.subworkflow(...).

A complete example

A multi-stage pipeline: ingest → validate → { text features, image features } → train → report. It fans out into two parallel feature stages and merges back at training, so the dependency graph is diamond-shaped. The sleeps let you watch the live dashboard advance, and skip_if_complete: false makes every run actually execute.

name: pipeline
kind: horus_workflow

tasks:
  - id: ingest
    name: Ingest data
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    runtime:
      kind: command
      command: "sleep 1\necho 'rows: 1000' > $raw\n"
    executor:
      kind: shell
    outputs:
      - { kind: file, id: raw, path: "./horus-out/raw.txt" }

  - id: validate
    name: Validate schema
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    runtime:
      kind: command
      command: "sleep 1\ncat $raw > $valid\necho 'validated: true' >> $valid\n"
    executor:
      kind: shell
    inputs:
      - { kind: file, id: raw, path: "./horus-out/raw.txt" }
    outputs:
      - { kind: file, id: valid, path: "./horus-out/valid.txt" }

  - id: features_text
    name: Text features
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    runtime:
      kind: command
      command: "sleep 2\ncat $valid > $feat_text\necho 'text_features: 256' >> $feat_text\n"
    executor:
      kind: shell
    inputs:
      - { kind: file, id: valid, path: "./horus-out/valid.txt" }
    outputs:
      - { kind: file, id: feat_text, path: "./horus-out/feat_text.txt" }

  - id: features_image
    name: Image features
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    runtime:
      kind: command
      command: "sleep 2\ncat $valid > $feat_image\necho 'image_features: 512' >> $feat_image\n"
    executor:
      kind: shell
    inputs:
      - { kind: file, id: valid, path: "./horus-out/valid.txt" }
    outputs:
      - { kind: file, id: feat_image, path: "./horus-out/feat_image.txt" }

  - id: train
    name: Train model
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    resources:
      cpus: 8
      gpus: 2
      memory_gb: 32
      walltime: "01:00:00"
    runtime:
      kind: command
      command: "sleep 3\ncat $feat_text $feat_image > $model\necho 'model: trained' >> $model\n"
    executor:
      kind: shell
    inputs:
      - { kind: file, id: feat_text, path: "./horus-out/feat_text.txt" }
      - { kind: file, id: feat_image, path: "./horus-out/feat_image.txt" }
    outputs:
      - { kind: file, id: model, path: "./horus-out/model.txt" }

  - id: report
    name: Write report
    kind: horus_task
    skip_if_complete: false
    target: { kind: local, working_directory: "./horus-work" }
    runtime:
      kind: command
      command: "sleep 1\necho '=== REPORT ===' > $report\ncat $model >> $report\n"
    executor:
      kind: shell
    inputs:
      - { kind: file, id: model, path: "./horus-out/model.txt" }
    outputs:
      - { kind: file, id: report, path: "./horus-out/report.txt" }

edges:
  - { source: ingest,         source_output: raw,        target: validate,       target_input: raw }
  - { source: validate,       source_output: valid,      target: features_text,  target_input: valid }
  - { source: validate,       source_output: valid,      target: features_image, target_input: valid }
  - { source: features_text,  source_output: feat_text,  target: train,          target_input: feat_text }
  - { source: features_image, source_output: feat_image, target: train,          target_input: feat_image }
  - { source: train,          source_output: model,      target: report,         target_input: model }

Run it:

horus run pipeline.yaml

If a task needs real Python logic or has to prompt the user, see Writing workflows in Python. To run one task over a collection and collect the results, see Fan-out and fan-in.

The runtime, executor, target, and artifact kinds above are plugins. You can add your own when the built-ins are not enough; see Extending Horus.