Workflow Definitions

Where definitions live

<product>/
  workflow/
    workflows/   # workflow definitions
    config/      # workflow config
    plugins/     # script plugins
    extend/      # entity extensions

Definitions are delivered through meta and reloaded on the tenant’s next refresh. A definition not yet loaded can still be started by path — the service fetches it from meta on demand.

File formats

Three shapes, auto-detected:

# 1. single flow
flow:
  name: order-approval
  tasks: [...]

# 2. multiple named flows in one file — select with `workflowname`
flows:
  - name: child-deploy
    tasks: [...]
  - name: parent-pipeline
    tasks: [...]

# 3. flat / legacy — top-level name + tasks
name: order-approval
tasks: [...]

Workflows can also be written in JavaScript, Lua, or Starlark.

Flow-level keys

Key	Type	Description
name	string	Human name — the lookup key for workflowname
id	string	Definition id (auto-generated if absent)
description, version	string	Metadata; version defaults to 1.0.0
tasks	list	Required. The node definitions
starttask	string	First node (defaults to the first listed)
vars	map	Flow variables merged into instance context
loglevel	string	debug / info / warn / error
workingdirectory	string	Working directory for shell tasks
continueonerror	bool	Keep going after a task fails
affinity	object	scope, payload_fields, restart
access	object	Authorization: rule, language, on_deny, replace
routinepool.size / .queue / .spawn	int	Defaults 100 / 100 / 50

Task-level keys

Common: name (required, unique), type, shell, print, vars, input, output, check (skip the node when the expression is false), selectors, group, final, before, after, events, parallel, access.

Routing: next: {go, when?, set?} and error: {go, when?, set?}.

Node type is selected by which key is present: assign (alias pass), choice, fail, succeed, foreach, map, wait, subworkflow. Absent all of them, the node is a Task.

Cross-cutting:

retry:
  max_attempts: 3
  backoff_base: 500ms
  backoff_max: 5s
  backoff_multiplier: 2.0     # default
  jitter_fraction: 0.1        # default
  retryable_errors: []
  non_retryable_errors: []
timeout:
  total: 10m
  execute: 5m

Node reference

# Assign — set context inline
- name: init
  assign:
    set:
      region: us-east
      services: [{name: api, port: 8080}]
  next: {go: choose}

# Choice — first matching rule wins
- name: choose
  choice:
    rules:
      - {condition: "strategy === 'canary'", next: canary}
      - {condition: "strategy === 'rolling'", next: rolling}
    default: invalid-strategy

# Fail / Succeed — terminate
- name: invalid-strategy
  fail:
    error: ConfigError
    cause: "Unknown strategy. Use rolling, blue-green, or canary."

# Map — parallel iteration
- name: build-all
  map:
    items_path: services
    task_name: build-one
    max_concurrency: 4
  next: {go: deploy}

# ForEach — sequential iteration
- name: deploy
  foreach:
    items_path: targets
    task_name: deploy-one
  next: {go: verify}

# SubWorkflow — call another definition
- name: run-deploy
  subworkflow:
    definition_id: child-deploy
    input_mapping:  {app_name: service_name, environment: target_env}
    output_mapping: {deploy_status: result_status}

Inside a foreach/map child task, the current element is {{_item}} and its position is {{_index}}.

Iterators

Instead of items_path, a loop can pull from a generator:

- name: process-files
  foreach:
    iterator:
      type: files          # numbers | array | time-window | lines | files | dirs | git-repos | shell-output
      path: ./data
      pattern: "*.csv"
    task_name: process-one

Loop position is checkpointed, so a restart resumes mid-iteration.

Context variables

Variable	Set by
_item, _index	ForEach / Map child tasks
_results, _result	Collected loop results / last task result
_error, _cause	Error routes
_signal	Wait node — the signal payload (also flattened to top level)
_wait_timeout, _wait_node	Wait node timeout
_foreach_node, _map_node	Enclosing loop node name

Task types

Available in this block: db, extract, encrypt, decrypt, git, hash, script, print, s3, shell, secret, setenv, setvars, ssh, scp, rsync, httpclient, http, supershell, suspend, template, throw, id, vault, yaml, yamlx, jq, json, file, liquid, glob, azureblob, archive, rule, python, mojo, scraper, wait, lock, docker, podman, k8s, kubectl, dns, letsencrypt.

Agent selection

- name: train
  type: shell
  selectors:
    tag: gpu           # tag | model | gpu | freegpu | cpu | memory | id | group

Worked example

A deployment pipeline exercising Assign → Choice → Map → ForEach → Succeed:

name: deployment-pipeline
vars:
  strategy: blue-green
  version: 2.1.0

tasks:
  - name: init-config
    assign:
      set:
        services:
          - {name: api,    port: 8080, replicas: 3}
          - {name: worker, port: 9090, replicas: 2}
        targets:
          - {host: prod-01.example.com, zone: us-east}
          - {host: prod-02.example.com, zone: us-west}
        deploy_version: "{{version}}"
    next: {go: choose-strategy}

  - name: choose-strategy
    choice:
      rules:
        - {condition: "strategy === 'rolling'",    next: build-services}
        - {condition: "strategy === 'blue-green'", next: build-services}
      default: invalid-strategy

  - name: invalid-strategy
    fail:
      error: ConfigError
      cause: "Unknown deployment strategy."

  - name: build-services
    map:
      items_path: services
      task_name: build-one-service
      max_concurrency: 4
    next: {go: deploy-to-targets}

  - name: build-one-service
    shell: echo "Building {{_item.name}}:{{deploy_version}}..."

  - name: deploy-to-targets
    foreach:
      items_path: targets
      task_name: deploy-to-server
    next: {go: deployment-complete}

  - name: deploy-to-server
    shell: echo "Deploying to {{_item.host}} ({{_item.zone}})..."

  - name: deployment-complete
    succeed: true

Authoring locally

Iterate on a definition with the CLI before committing it:

kis flow -f workflow.yaml                # run
kis flow -f workflow.yaml -v env=prod    # with vars
kis flow -f workflow.yaml -w 4           # 4 workers (needed for Map parallelism)
kis flow -f workflow.yaml -d             # validate only
kis flow -f multi.yaml -n build          # pick one flow from a flows: file

Actual Map parallelism is min(items, max_concurrency, workers) — with one worker, a Map runs serially.