VSKI v0.14.0

Pipelines: automations as data

Workflows have been VSKI’s answer to “run this reliably” — but they live in worker code you write, deploy and version yourself. v0.14.0 adds pipelines: multi-step automations defined as JSON and edited on a visual canvas, executed durably by the engine itself. No worker, no deployment — the definition is the automation.

A pipeline is a DAG of steps — call a plugin function, make an HTTP request, compute values, wait on a timer, park for a human approval, spawn a child workflow or pipeline — with per-step dependencies, conditions, retries, timeouts and failure policies. Pipelines start from collection events (record created/updated/deleted), cron schedules, or manually via API/UI.

They are not a parallel universe: every pipeline run is an ordinary workflow run named pipeline.<name>, driven by the same event log, job queues, waits and signals machinery. The run monitor, Gantt timeline, stats, circuit breaker, execution timeouts and crash recovery all apply unchanged — pipeline steps emit the standard step_* events with correlationId = step id, so the existing timeline renders them for free.

The definition

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

Step types: plugin (warm in-process invoke, instance/function, timeout 1-120s), http (synchronous, every attempt in _webhook_logs, output {status, body}), set (template evaluation into {{vars.*}}), delay (durable wait — survives restarts, fires exactly at the resume time), signal (approval gate — parks until a matching signal arrives, FIFO by name, resumable via sendSignal or workflow hook tokens), and workflow/pipeline spawn steps (fire-and-forget children linked via parentRunId).

Common step fields: dependsOn (independent branches run in parallel, bounded by options.concurrency), condition (data-only expression language — comparisons, and/or/not, literals; sandboxed by construction), retries (linear backoff), timeout, onError: "fail" | "continue" (failed steps skip their transitive dependents while unrelated branches continue; continue captures the error as steps.<id>.error).

Templates: {{trigger.*}}, {{input.*}}, {{steps.<id>.output|.status|.error}}, {{vars.*}}, {{run.id}} - apply to any string in params/headers/body/input, with array indexing; a whole-string template preserves the value’s type.

Execution semantics worth knowing

• Version pinning: each run snapshots the definition into its executionContext. Editing or deleting a pipeline never mutates an in-flight run.
• At-least-once across crashes: the event log is the source of truth; after a restart the dispatcher replays events and re-executes only steps that hadn’t completed (boot recovery re-pends orphaned jobs and requeues active runs).
• Per-run serialization: runs of one pipeline execute one at a time; independent pipelines run concurrently. Steps of one run run in parallel up to options.concurrency (default 4).
• True suspension: while a run waits on a delay or signal it holds no job; a failed branch doesn’t fail the run while another branch still waits.
• Replicas stay cold: triggers are only registered and pipelines only dispatched on the master.

Triggers

• Collection events: the pipeline subscribes from its definition, no collection-options changes, no auth-header webhook wiring. Disabled pipelines ignore events; the registry rebuilds on every definition change.
• Cron: {"type": "cron", "schedule": "0 9 * * 1-5"} inside the definition registers an in-process named schedule (retired on disable/update/delete, no cron-job rows). Ad-hoc: any cron job with url: "pipeline://<name>" fires a pipeline.
• Manual: POST /api/pipelines/:name/runs or the UI Run dialog.

The builder

The dashboard grows a Pipelines section: a list (trigger chips, version, last-run status) and a full-page visual builder on React Flow - drag nodes to declare dependsOn, per-type step inspectors parsed from live plugin catalogs (function lists come from manifests), a JSON⇄Canvas toggle that round-trips the definition, server-side Validate with per-step issues, and a Run dialog. A Runs button polls the last 10 runs with a dot indicator (pulsing green while anything is active), linking into the existing run views; the workflows page accepts ?workflowName=pipeline.<name> to pre-filter its list.

API & SDKs

GET/POST /api/pipelines, GET/PATCH/DELETE /api/pipelines/:name (definition replace bumps version; rename rejected), POST /api/pipelines/validate (per-step issue maps: cycles, refs, cron, durations, template/condition syntax, option ranges), POST /api/pipelines/:name/runs. Auth: the workflows role or admin, like the rest of the workflow API. pipelines is also queryable as a system collection.

• JS/TS (@vski/sdk 2.3.0): client.pipelines — list, get, getDefinition, create, update, remove, validate, run; exports Pipeline, PipelineDefinition, PipelineStep, PipelineTrigger, ValidationIssue types.
• Python (vski-sdk 0.7.0): client.pipelines — list, get, get_definition, create, update, delete, validate, run.

Correctness fixes

• plugin:// cron jobs never worked — documented since the plugins release, but the cron executor routed every command through the SQL branch, so plugin:// jobs failed as SQL syntax errors. Cron plugin:// commands now dispatch through the webhook retry/log pipeline over the warm pipe, and replicas correctly skip all dispatch-type commands (http/plugin/pipeline).
• ExecuteStandaloneWebhook dropped the job’s database whenever a body was set, breaking plugin:// instance resolution for POST cron jobs.
• The OpenAPI spec documented POST/GET/PATCH/DELETE /workflows definition CRUD endpoints that never existed (routes were never implemented). The dead spec block is removed.
• Workflows dashboard: pre-existing 'sdk' possibly null type errors fixed; the run list prefills its filter from ?workflowName= deep links.

Tests

test/e2e/55_pipelines_test.ts (9 steps): validation + CRUD, DAG with parallel branches/join/condition skip/output templating, durable delay, signal approval with templated payload, failure semantics (transitive skips, unrelated branches complete, onError: continue), collection + cron triggers (definition schedule and pipeline:// command), spawn steps with parentRunId, and replay-skip resume through the real dispatcher. The engine gains its first Go unit-test suite (internal/pipelines): template resolver, condition grammar, validator (cycles, per-type fields), and the executor against a fake runtime (retries, suspension, resume-skip, vars fold order, terminal idempotency), plus trigger-registry tests.

Compatibility

• No breaking API or storage changes. One new table (pipelines in workflows.db, registered as a system collection) and new, additive endpoints/SDK namespaces. Existing workflows, triggers and cron jobs are untouched.
• Pipeline dispatch is master-only; upgrading a replica is a no-op for pipelines.
• The standalone Docker build now ships the SDK source beside the web app (build/client), since the web app compiles against local SDK source.
• JS SDK 2.3.0 publishing to vski.sh/npm is part of the release run; external consumers need it for client.pipelines.

What’s next

Deliberately out of v1: foreach fan-out over arrays, wait-for-completion on spawned children (model it with a signal step today), a public token endpoint for external triggers, realtime status overlay on the canvas, and a native llm step type. The Pipelines wiki page documents all of it.