This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

Pipelines

Automate multi-step processes as declarative JSON DAGs — no worker code required. A pipeline definition says what to run (plugin functions, HTTP calls, timers, approvals, child workflows) and in what order; the engine executes it durably, retries failures, suspends on approvals, and renders every run on the standard workflow timeline.

Understanding Pipelines

Pipelines are built on top of the Workflows engine. Every pipeline run is an ordinary workflow run named pipeline.<name>, driven by the same event log, job queues, waits and signals machinery — so the run monitor, Gantt timeline, stats, circuit breaker, execution timeouts and crash recovery all apply to pipelines unchanged.

Use workflows when the logic lives in your own TypeScript or Python worker code. Use pipelines when the automation can be expressed as data: calling plugin functions, HTTP endpoints, waiting on timers or human approvals, and spawning other workflows or pipelines. Pipelines are edited in a visual builder (or raw JSON) and versioned server-side — there is nothing to deploy.

Definitions live in the pipelines system collection (inside workflows.db) and are version-bumped on every definition change. Each run pins a snapshot of the definition it started with, so editing a pipeline never mutates an in-flight run.

Creating Pipelines

From the UI

Dashboard → Pipelines → New pipeline. The builder opens with a starter definition: add steps from the canvas toolbar, connect nodes to declare dependsOn, and configure each step in the inspector that opens on click. A Canvas / JSON toggle edits the raw definition directly; both sides round-trip. Validate checks the definition server-side (per-step issues show on the badge and inside the inspector), Save stores a new version, and Run starts a manual run.

From the SDK

const pipeline = await client.pipelines.create({
  name: "note-intake",
  description: "Index and notify on new notes",
  triggers: [
    { type: "collection", database: "default", collection: "notes", actions: ["create"] },
  ],
  steps: [
    {
      id: "embed",
      type: "plugin",
      target: "indexer/reindex",
      params: { collection: "{{trigger.collection}}", id: "{{trigger.record.id}}" },
      retries: 2,
    },
    {
      id: "notify",
      type: "http",
      method: "POST",
      url: "https://hooks.example.com/notes",
      body: { id: "{{trigger.record.id}}" },
      dependsOn: ["embed"],
    },
  ],
});

const run = await client.pipelines.run("note-intake", { priority: true });
console.log(run.runId);

pipeline = await client.pipelines.create({
    "name": "note-intake",
    "description": "Index and notify on new notes",
    "steps": [
        {"id": "embed", "type": "plugin", "target": "indexer/reindex",
         "params": {"collection": "{{trigger.collection}}", "id": "{{trigger.record.id}}"}},
        {"id": "notify", "type": "http", "method": "POST",
         "url": "https://hooks.example.com/notes", "dependsOn": ["embed"]},
    ],
})

run = await client.pipelines.run("note-intake", {"priority": True})

Pipeline names match ^[a-zA-Z0-9_-]{1,64}$ and are unique per database. Patching the definition bumps version; toggling enabled does not.

The Definition

{
  "name": "note-intake",
  "description": "Index and notify on new notes",
  "enabled": true,
  "triggers": [
    { "type": "collection", "database": "default", "collection": "notes", "actions": ["create", "update"] },
    { "type": "cron", "schedule": "0 9 * * 1-5" }
  ],
  "steps": [
    {
      "id": "embed",
      "type": "plugin",
      "target": "indexer/reindex",
      "params": { "collection": "{{trigger.collection}}", "id": "{{trigger.record.id}}" },
      "timeout": 60,
      "retries": 2,
      "onError": "fail",
      "dependsOn": [],
      "condition": "{{trigger.action}} == 'create'"
    },
    { "id": "slowpath", "type": "delay", "duration": "30s", "dependsOn": ["embed"] },
    { "id": "approve", "type": "signal", "name": "approval", "dependsOn": ["embed"] },
    { "id": "vars", "type": "set", "values": { "n": "{{steps.embed.output.count}}" }, "dependsOn": ["embed"] },
    {
      "id": "notify",
      "type": "http",
      "method": "POST",
      "url": "https://hooks.example.com/notes",
      "headers": { "X-Api-Key": "..." },
      "body": { "count": "{{vars.n}}" },
      "dependsOn": ["approve", "vars"]
    },
    { "id": "child", "type": "workflow", "name": "heavy-job", "input": { "ref": "{{trigger.record.id}}" } },
    { "id": "sub", "type": "pipeline", "name": "audit-trail", "input": {} }
  ],
  "output": { "count": "{{steps.embed.output.count}}" },
  "options": { "concurrency": 4, "maxEvents": 10000, "executionTimeout": 3600000 }
}

Triggers

Field	Values
type: "collection"	database (default "default"), collection, actions: subset of create, update, delete
type: "cron"	schedule: standard 5-field cron expression

• Collection triggers fire whenever a matching record is written — no changes to the collection's own trigger options are needed; the pipeline subscribes from its definition. Disabled pipelines ignore events.
• Cron triggers are registered as in-process named schedules when the dispatcher boots (and on every definition change); they do not create rows in the cron-jobs table. A disabled pipeline's schedules are retired.
• Manual runs (UI, POST /api/pipelines/:name/runs, SDK run) always work regardless of triggers.

The trigger context is available to every step as {{trigger.*}}:

// collection event
{ "type": "collection", "action": "create", "database": "default",
  "collection": "notes", "id": "rec_123", "record": { ... }, "timestamp": "..." }
// cron fire and manual runs
{ "type": "cron" }

Cron jobs can also start a pipeline ad-hoc with the pipeline:// command type — see Cron.

Steps

Every step shares these fields:

Field	Default	Description
id	—	Unique within the pipeline (^[a-zA-Z0-9_-]{1,64}$)
type	—	One of the types below
dependsOn	[]	Step ids that must complete first
condition	—	Expression evaluated when the step becomes ready; false skips the step
retries	0	Retry attempts on failure (linear backoff)
timeout	—	Per-step timeout in seconds
onError	"fail"	"fail" cancels dependents; "continue" records the error and moves on

plugin

Calls a plugin function over the warm in-process pipe — same path as the plugin:// trigger scheme, no network hop.

{ "id": "embed", "type": "plugin", "target": "indexer/reindex",
  "params": { "collection": "notes", "id": "{{trigger.record.id}}" },
  "timeout": 60, "retries": 2 }

target is <instance>/<function>; timeout is clamped to 1–120s (default 30). The function's return value becomes the step output.

http

A single synchronous HTTP attempt; every attempt is logged to the webhook logs (_webhook_logs). Retries use the step-level retries (the same step_retrying mechanism as every other step type).

{ "id": "notify", "type": "http", "method": "POST",
  "url": "https://hooks.example.com/notes",
  "headers": { "X-Api-Key": "..." },
  "body": { "count": "{{vars.n}}" } }

The step output is { "status": 200, "body": ... } — body is the parsed JSON response, or the raw string for non-JSON bodies. Non-2xx statuses fail the step (with a snippet of the response in the error).

set

Evaluates a map of templates and folds the results into the run context as {{vars.<key>}} — pure computation, no side effects.

{ "id": "vars", "type": "set",
  "values": { "n": "{{steps.embed.output.count}}", "label": "note" } }

delay

Durable timer. Emits a wait_created event with the resume time; the scheduler wakes the run exactly when due — even across server restarts (the wait row survives, the resume time does not move).

{ "id": "slowpath", "type": "delay", "duration": "30s" }

duration accepts 30s, 5m, 1h, or a bare number of seconds ("90").

signal

Durable approval / external gate. The step parks in signal_waiting until a matching signal arrives, then completes with the signal data as its output. Signals are matched by name, first-in-first-out across the whole run event log — a signal sent before the step starts waiting also satisfies it (mirroring worker-queue semantics).

{ "id": "approve", "type": "signal", "name": "approval" }

Resume paths (all of them requeue the run):

// by run id
await client.workflow.sendSignal(runId, "approval", { ok: true, by: "anton" });
// or via a workflow hook token (POST /api/workflows/hooks/:token)

Downstream steps template the payload: {{steps.approve.output.ok}}.

workflow / pipeline

Spawns a child run — fire-and-forget: the step completes immediately with { "runId": "..." } and the pipeline continues. The child carries parentRunId and shows under the parent in the run monitor.

{ "id": "child", "type": "workflow", "name": "heavy-job", "input": { "ref": "{{trigger.record.id}}" } }
{ "id": "sub",   "type": "pipeline", "name": "audit-trail", "input": {} }

workflow children are ordinary worker workflows (a job lands on the __wkf_workflow_<name> queue for your workers); pipeline children get their own definition snapshot and run through the pipeline dispatcher. Waiting for a child to finish is not supported in v1 — model it with a signal step the child's completion hook sends.

Templates and conditions

Any string value in params, headers, body, input, values, condition and output supports {{dot.path}} substitution over:

Path	Contents
trigger.*	The trigger context (see above)
input.*	The run input (manual runs, spawn input)
steps.<id>.output	A completed step's output
steps.<id>.status	completed / failed / skipped
steps.<id>.error	Error text of a failed-but-continued step
vars.*	Values folded by set steps
run.id	The current run id

Array indexing works: {{trigger.record.tags[0]}}. A template that is the entire string preserves the value's type (numbers, objects, arrays); templates embedded in longer strings stringify. A missing path is an error — guard with condition when a path may be absent.

Conditions are data-only expressions — comparisons == != > >= < <=, and or not, parentheses, and literals ('x', "x", numbers, true/false/null). Missing paths evaluate to null; there are no function calls, so the expression language is sandboxed by construction.

"condition": "{{trigger.action}} == 'create' and {{input.priority}} >= 2"

A false condition skips the step (step_skipped with reason condition); its dependents are skipped transitively, exactly like a failure.

Output

When the run settles, output (a map of templates) is evaluated and stored on the run. Without an output, the run output defaults to a map of the terminal steps' outputs (steps nothing depends on).

Options

Option	Default	Description
concurrency	4	Max steps of one run executing in parallel
maxEvents	1000000	Circuit breaker: run fails if events exceed this
executionTimeout	engine default (365 days)	Wall-clock limit for the whole run (ms)

Execution Semantics

• Readiness: a step runs when every dependency has completed. Independent branches execute in parallel (bounded by concurrency); joins simply list several dependsOn.
• Serialization: runs of one pipeline execute one at a time; independent pipelines run concurrently.
• Version pinning: the run snapshots the definition at trigger time into its executionContext. Later edits (or deletion) do not affect in-flight runs.
• Durability: the event log is the source of truth. Steps execute at-least-once across crashes — the engine restarts, replays events, re-executes only steps that had not completed. Plugin functions and HTTP handlers should therefore be idempotent, same as with worker workflows.
• Suspension: while any step waits on a delay or signal, the run parks (no busy job). A failed branch does not fail the run while another branch is still waiting; the run settles only when every step is terminal.
• Failure: with onError: "fail" (default), the failed step's transitive dependents are skipped (step_skipped events) while unrelated branches continue; the run ends failed once no work remains. With onError: "continue", the error is captured as steps.<id>.error and dependents still run.
• Replicas: replicas never dispatch pipelines or register triggers — only the master does.

Run events

Pipelines emit the standard worker event contract with correlationId = <stepId>, so the Gantt timeline renders them natively:

Event	Payload
step_started	{ attempt, name, type }
step_completed	{ output }
step_retrying	{ error, attempt }
step_failed	{ error, attempt, continued? }
step_skipped	{ reason }
wait_created / wait_completed	{ duration, resumeAt }
signal_waiting / signal_received	{ name }

Monitoring Runs

Open Workflows → pipeline. (or click any run from the Pipelines list or the builder's Runs modal). The run page shows status, input, output, the event log and the Gantt timeline; the Runs modal polls the last 10 runs with a green dot while anything is active. All the usual workflow tooling applies — stats, children listing (parentRunId), signals, hooks.

Deep link: /workflows?workflowName=pipeline.<name> pre-filters the runs list.

API Endpoints

Requires the workflows role (or admin) — same as the rest of the workflow API.

Method	Endpoint	Description
GET	/api/pipelines	List pipelines
POST	/api/pipelines	Create (body = definition)
GET	/api/pipelines/:name	Get one (row + parsed definition)
PATCH	/api/pipelines/:name	Replace definition (bumps version) and/or enabled
DELETE	/api/pipelines/:name	Delete
POST	/api/pipelines/validate	Validate a definition (per-step issue map)
POST	/api/pipelines/:name/runs	Start a run (body = run input)

Validation rejects unknown step types, duplicate ids, bad dependsOn references, cycles, malformed cron schedules, bad durations, template and condition syntax errors, and out-of-range options — each with the offending step id.

SDKs

TypeScript (@vski/sdk ≥ 2.3.0) — client.pipelines:

Method	Notes
list() / get(name) / getDefinition(name)	getDefinition returns the parsed definition object
create(definition, options?)	options.enabled / options.dbName
update(name, { definition?, enabled? })	Definition replace bumps version; rename is rejected
remove(name)
validate(definition)	{ valid, issues }
run(name, input?, options?)	options.runId / options.idempotencyKey; returns the created WorkflowRun

Python (vski-sdk ≥ 0.7.0) — client.pipelines: list, get, get_definition, create, update, delete, validate, run.

Practical Examples

Index and notify on new records

{
  "name": "note-intake",
  "triggers": [
    { "type": "collection", "database": "default", "collection": "notes", "actions": ["create", "update"] }
  ],
  "steps": [
    { "id": "embed", "type": "plugin", "target": "indexer/reindex",
      "params": { "collection": "notes", "id": "{{trigger.record.id}}" }, "retries": 2 },
    { "id": "notify", "type": "http", "method": "POST",
      "url": "https://hooks.example.com/notes",
      "body": { "id": "{{trigger.record.id}}", "text": "{{trigger.record.text}}" },
      "onError": "continue", "dependsOn": ["embed"] }
  ],
  "output": { "indexed": "{{steps.embed.output.count}}" }
}

Notification failures (onError: "continue") never fail the indexing run.

Human approval gate

{
  "name": "weekly-report",
  "triggers": [{ "type": "cron", "schedule": "0 9 * * 1-5" }],
  "steps": [
    { "id": "build", "type": "plugin", "target": "reports/weekly", "params": {} },
    { "id": "approve", "type": "signal", "name": "publish-approval", "dependsOn": ["build"] },
    { "id": "publish", "type": "http", "method": "POST",
      "url": "https://cms.example.com/publish",
      "body": { "approvedBy": "{{steps.approve.output.by}}" },
      "dependsOn": ["approve"] }
  ]
}

The run parks at approve until someone sends the publish-approval signal (workflow hook URL, sendSignal, or any script) — then publishes with the approver's name templated into the request.

Fan-out to a worker workflow

{
  "name": "bulk-backfill",
  "steps": [
    { "id": "page1", "type": "workflow", "name": "heavy-job", "input": { "page": 1 } },
    { "id": "page2", "type": "workflow", "name": "heavy-job", "input": { "page": 2 } },
    { "id": "done", "type": "set", "values": { "queued": true }, "dependsOn": ["page1", "page2"] }
  ]
}

Both children are queued immediately (no dependsOn between them); done joins. Children appear linked via parentRunId.

Best Practices

• Keep steps small and idempotent — retries and crash recovery re-execute a step that may have already had side effects.
• Prefer onError: "continue" for notification-style side effects so a flaky webhook never fails the important branch.
• Use condition on trigger-driven steps to separate create from update handling instead of maintaining two pipelines.
• Set retries (with linear backoff) on any HTTP step that leaves your network.
• Model long waits as delay/signal steps — they park the run durably instead of occupying a worker or a timeout budget.
• Validate definitions in CI with POST /api/pipelines/validate; the builder's Validate button calls the same endpoint.