Pipelines

Pipelines define what happens when you run sparkwing run <name> (or sparkwing pipeline run <name>). See the SDK guide for API usage and Authoring pipelines for the rules enforced by sparkwing pipeline lint.

Host requirements. Pipelines that call sparkwing.Bash shell out to bash on the runner host. macOS and Linux have this by default. On Windows, install Git for Windows and run pipelines from the Git Bash terminal it ships -- sparkwing.Exec (no-shell, arg-vector form) works without it. The bundled sparkwing-runner service installer supports Linux and macOS. On Windows, supervise sparkwing-runner.exe agent yourself or use the Linux installer inside WSL when systemd user services are enabled.

Pipeline registry

The pipelines: block in .sparkwing/sparkwing.yaml is the registry of every pipeline in the repo (pipelines plus commands); the block holds both kinds. Each entry is a list item with a name:.

# .sparkwing/sparkwing.yaml
pipelines:
  - name: build-deploy
    entrypoint: BuildDeploy
    description: Build and deploy the app
    on:
      push:
        branches: [main]

  - name: release
    entrypoint: Release
    description: Cut a release
    # no on: -> command, manual-only

Each entry has (these are the only valid keys; an unknown field is a hard parse error):

• name - the pipeline name (sparkwing run build-deploy); must equal the Register("name", ...) string and match ^[A-Za-z0-9][A-Za-z0-9._-]*$
• entrypoint - the Go pipeline struct type implementing it (required); equals the struct name
• description - one-line summary surfaced by sparkwing pipeline list
• on - declarative trigger block: push (branches/paths), pull_request (actions/branches), schedule (one entry, or a list of them, each with name, cron, a required where, tz, overlap, catch_up, args), webhook, pre_commit, pre_push, post_commit. Absent means "manual only" (a command).
• guards - gate dispatch on profile, args, and git branch (reject / require token lists)
• args - per-arg default values, keyed by CLI flag name
• profile - the project profile this pipeline uses (from the profiles: map)
• requires - runner-label requirements for every job (e.g. [local] pins execution to this machine)
• hidden - omit from pipeline list (still invocable by exact name)

For the complete schema -- every top-level key, pipeline field, and trigger field with types -- see the generated config-reference.md.

Triggers

Trigger types live under on::

# .sparkwing/sparkwing.yaml
pipelines:
  # Fires on a git push the controller receives via webhook
  - name: build
    entrypoint: Build
    on:
      push:
        branches: [main]                 # declarative: records intent, not gated on
        paths: ["*.go", "go.mod"]        # declarative: records intent, not gated on

  # Run on every pull request (checks out the PR head)
  - name: pr-gate
    entrypoint: PRGate
    on:
      pull_request:
        branches: [main]                 # declarative: records intent, not gated on

  # Declarative path: the controller exposes POST /webhooks/github/{pipeline};
  # this path is recorded, not routed
  - name: review
    entrypoint: Review
    on:
      webhook:
        path: /review

  # Cron in UTC. `where` is required: `local` fires from a host once
  # armed there; until then invoke it with `sparkwing run nightly`
  - name: nightly
    entrypoint: Nightly
    on:
      schedule:
        cron: "0 2 * * *"
        where: local

branches / paths / actions record intent: the controller does not read your sparkwing.yaml, so it dispatches whichever pipeline the webhook URL names. To require a checked-out branch before any step runs, add guards: {require: [git:branch=main]} with its literal name. For pull requests this compares the head branch, not the base branch. git:branch=default requires default-branch metadata that controller webhook and local trigger claims do not supply. Branch guards do not express path restrictions, custom pull-request actions, or pull-request base-branch matching. See hooks for the enforcement boundary and examples.

Webhook delivery is handled by the controller - see POST /webhooks/github/{pipeline} in api. Git hooks are not installed automatically -- declaring an on: pre_commit / pre_push / post_commit trigger does nothing until you run sparkwing pipeline hooks install, which writes the hook files into .git/hooks/; see hooks for context.

The two-layer model

Sparkwing has two DAG layers, and almost every pipeline-authoring choice is a layer choice. Internalize this before reading the recipes below.

• Plan / Job is the outer DAG - units of dispatch. Each Job runs in its own process: a separate pod in cluster mode, a separate invocation of the pipeline binary in local mode. Nodes carry the dispatch envelope - Retry, Timeout, OnFailure, Memoize, Requires, BeforeRun / AfterRun, Approval gating - because each Job is the unit the scheduler can retry, time out, or route to a labeled runner.
• Work / WorkStep is the inner DAG - units of work within one Job's runner. Steps share the Job's runner, filesystem, environment, and ctx. They have Needs for ordering and SkipIf for predicates; they do not carry Job-only modifiers (Retry, Timeout, ...). Promote a step to a Job via JobSpawn if it needs one.

Each pipeline implements Plan(ctx, plan *sw.Plan, in T, rc sw.RunContext) error which registers nodes on the outer DAG. A job's Workable.Work method registers its inner steps. The orchestrator materializes the reachable graph, including spawn targets, before dispatch so inspection can show it.

Cost grid

The verb tells you the cost. The Plan-layer Job* adders are cheap; the Work-layer JobSpawn* adders flag the layer jump and the suspended-runner cost. Reach for JobSpawn when you genuinely need Job-only modifiers (Retry, Requires, distinct runner) on a unit decided mid-execution; otherwise stay inside Work.

Trivial single-step jobs

For pipelines that are one closure with no DAG, pass the function directly to sw.Job -- no struct, no wrapper:

type Lint struct{ sparkwing.Base }

func (p *Lint) Plan(_ context.Context, plan *sparkwing.Plan, _ sparkwing.NoInputs, rc sparkwing.RunContext) error {
    sw.Job(plan, rc.Pipeline, p.run)
    return nil
}

func (p *Lint) run(ctx context.Context) error {
    if err := sparkwing.Bash(ctx, "gofmt -l .").MustBeEmpty("formatting drift"); err != nil {
        return err
    }
    _, err := sparkwing.Bash(ctx, "go vet ./...").Run()
    return err
}

// In .sparkwing/main.go:
//     sparkwing.Register[sparkwing.NoInputs]("lint", func() sparkwing.Pipeline[sparkwing.NoInputs] { return &Lint{} })

sw.Job's third argument is any: a func(ctx context.Context) error is wrapped into an internal Workable, while a struct implementing Work(w *Work) (*WorkStep, error) registers as a multi-step Job. Reflection picks the right form at register time.

For typed-output Jobs (downstream nodes read the value via Ref[T] / RefTo[T]), define a struct that embeds sparkwing.Produces[T] and return the typed step from Work:

type Build struct {
    sparkwing.Base
    sparkwing.Produces[BuildOut]
}

func (j *Build) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    return sw.Step(w, "run", j.run), nil
}

func (j *Build) run(ctx context.Context) (BuildOut, error) {
    return BuildOut{Tag: "app:sha-abc"}, nil
}

build := sw.Job(plan, "build", &Build{})
buildRef := sparkwing.RefTo[BuildOut](build)
sw.Job(plan, "deploy", &Deploy{Build: buildRef}).Needs(build)

Multi-step jobs

For jobs whose body is more than one logical phase, implement Workable yourself. The struct's Work(w *Work) (*WorkStep, error) method registers steps onto the passed-in *Work and returns the result step (or nil for an untyped Job). Each sw.Step is a unit of work; Needs declares ordering.

type Build struct{ sparkwing.Base }

func (j *Build) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    fetch    := sw.Step(w, "fetch",    j.fetch)
    validate := sw.Step(w, "validate", j.validate)
    sw.Step(w, "compile", j.compile).Needs(fetch, validate)
    return nil, nil  // untyped Job; no result step
}

func (j *Build) fetch(ctx context.Context) error    { return j.gitFetch(ctx) }
func (j *Build) validate(ctx context.Context) error { return j.checkGoMod(ctx) }
func (j *Build) compile(ctx context.Context) error  { return j.goBuild(ctx) }

The DAG is built entirely from .Needs() chains. For sequential steps, chain Needs directly; there is no separate Sequence combinator:

func (j *Deploy) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    a := sw.Step(w, "render-manifests", j.render)
    b := sw.Step(w, "argo-sync",        j.sync).Needs(a)
    sw.Step(w, "verify",                j.verify).Needs(b)
    return nil, nil
}

For named clustering of related steps -- the dashboard's Work view folds members under one collapsible header -- use sw.GroupSteps:

func (j *Deploy) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    fetch := sw.Step(w, "fetch", j.fetch)

    safety := sw.GroupSteps(w, "safety",
        sw.Step(w, "lint",    j.lint).Needs(fetch),
        sw.Step(w, "secscan", j.secscan).Needs(fetch),
        sw.Step(w, "vet",     j.vet).Needs(fetch),
    )

    return sw.Step(w, "deploy", j.deploy).Needs(safety), nil
}

*StepGroup is both a Needs target (downstream steps that Needs(group) depend on every member) and a UI cluster. Initial modifiers mirror what *WorkStep has today (Needs, SkipIf); each applies to every member.

Typed step output

For the common case -- a Job with a single typed step whose return value IS the Job's output -- declare the step with a typed signature and return it from Work:

func (j *Build) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    return sw.Step(w, "compile", j.compile), nil
}

sw.Step's third argument is any: pass either a func(ctx context.Context) error (untyped) or a func(ctx context.Context) (T, error) (typed). Reflection at register time stores the step's output type.

For Works with multiple typed steps where downstream steps inside the same Work read intermediate values, use sw.StepGet[T](ctx, step) inside the consuming step's body:

func (j *Deploy) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    tags := sw.Step(w, "compute-tags", func(ctx context.Context) (Tags, error) {
        return loadTags(ctx)
    })
    return sw.Step(w, "publish", func(ctx context.Context) error {
        return publish(ctx, sw.StepGet[Tags](ctx, tags))
    }).Needs(tags), nil
}

StepGet reads a step's typed output within its job.

Inner step skip

step.SkipIf(predicate) skips a single step without aborting the Work. Multiple SkipIf calls accumulate with OR semantics.

sw.Step(w, "publish", j.publish).
    Needs(buildOut).
    SkipIf(func(ctx context.Context) bool { return os.Getenv("DRY_RUN") == "1" })

Plan-layer fan-out

Two type-safe verbs cover the Plan-layer fan-out cases. Both return a *JobGroup whose name becomes a collapsible cluster in the dashboard and a single Needs(group) target downstream.

Static: JobFanOut (slice in hand at Plan-time)

sw.JobFanOut[T] registers one Job per element of a slice already known when Plan() runs:

images := sw.JobFanOut(plan, "image-builds", Images, func(img imageSpec) (string, any) {
    return "build-" + img.Name, &BuildImage{Image: img}
}).Needs(webBuild, discover).Retry(2)

sw.Job(plan, "artifact", &Artifact{}).Needs(images)

The chained .Needs(...) / .Retry(...) apply to every member; see Group modifiers below.

Runtime: JobFanOutDynamic (slice produced by an upstream Job)

sw.JobFanOutDynamic[T] materializes one Plan-level Job per element of an upstream typed Job's output slice. Each fan-out child is a fresh Job with its own dispatch envelope:

type ListShards struct {
    sparkwing.Base
    sparkwing.Produces[[]string]
}

func (j *ListShards) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    return sw.Step(w, "run", j.run), nil
}

func (j *ListShards) run(ctx context.Context) ([]string, error) {
    return loadShards(ctx)
}

shards := sw.Job(plan, "list-shards", &ListShards{})

sw.JobFanOutDynamic(plan, "shard-work", shards, func(shard string) (string, any) {
    return "process-" + shard, &ProcessShard{Shard: shard}
})

JobFanOutDynamic creates children from its source job's output after the source completes and releases its runner.

Group modifiers

Every chainable *JobNode modifier has a *JobGroup twin: the call delegates to each member and returns the same *JobGroup for chaining. The generated sdk-reference.md lists the current set. Two carve-outs: OnFailure is intentionally per-Job, since group-level recovery has unclear semantics; and Memoize on a group is what the group-cache-shared lint rule rejects -- a constant key across N members makes them share a single cache entry and replay each other's results. Key each member instead, by ranging over group.Members(); see authoring-pipelines.md.

Layer escape: JobSpawn

When a unit of work decided mid-Work needs a Job-only modifier (Retry, Requires, distinct runner, separate cache key), promote it via sw.JobSpawn. The spawning runner suspends until the spawned Job completes:

func (j *ScanJob) Work(w *sparkwing.Work) (*sparkwing.WorkStep, error) {
    analyze := sw.Step(w, "analyze", j.analyze)
    scan := sw.JobSpawn(w, "compliance", &ComplianceJob{}).Needs(analyze)
    sw.Step(w, "publish", func(ctx context.Context) error {
        return publish(ctx, scan)
    }).Needs(scan)
    return nil, nil
}

The spawned Job id is namespaced as parent/spawnID (e.g. scan/compliance) so logs and the run history don't collide.

A spawned Job runs inside its parent node's own process -- a local node process, or a pod. It runs under the admission lease its parent already holds, so it is not charged against host capacity a second time, and concurrent children are capped the way the dispatcher caps nodes. It still records its own node row under parent/spawnID, its own logs, metrics, and output, and a spawn_dispatched event on the parent. A child whose WhenRunner labels the executing runner does not advertise is skipped, exactly as a planned node would be.

sw.JobSpawnEach(w, items, fn) is the cardinality-many variant. The generator runs once Needs are satisfied; each returned (id, Job) pair becomes a fresh Plan node. The spawning runner stays suspended across the entire fan-out:

sw.JobSpawnEach(w, targets, func(target string) (string, any) {
    return "deploy-" + target, &Deploy{Target: target}
}).Needs(buildStep)

Reach for spawn primitives sparingly. Each call holds a runner slot during the child's lifetime; a deeply nested spawn chain pins one slot per layer. The JobSpawn* prefix flags the layer jump (and the suspended-runner cost) at the call site.

Modifier scope discipline

A Step that needs Retry / Timeout / Requires is the canonical signal to promote it to a Job via sw.JobSpawn.

Scheduling modifiers

.Inline()

Marks a Job to run on the dispatcher's own host instead of being handed to the configured Runner, so no pod / warm-runner spin-up cost is paid.

sw.Job(plan, "setup", &Setup{}).Inline()
sw.Job(plan, "summarize", &Summarize{}).Needs(deploys).Inline()

A local inline job runs in its own process. With cluster dispatch, it executes inside the dispatcher and shares its worker pool. Long inline jobs occupy capacity the dispatcher needs for other nodes.

Approval gates expose no Inline modifier. Runner-selection labels are ignored for inline nodes.

Dynamic nodes

A node whose downstream work is runtime-variable is dynamic: JobFanOutDynamic source nodes are auto-marked dynamic at plan finalization. plan.IsDynamicNode(id) reports it and the plan preview shows [dynamic], so reviewers know to inspect the run for the actual child nodes rather than expecting the full shape at plan time.

GroupJobs(plan, "name", ...)

Groups existing nodes under one dashboard header and returns a handle that Needs can use to depend on every member.

sw.GroupJobs(plan, "safety",
    sw.Job(plan, "schema-check", &SchemaCheckJob{}),
    sw.Job(plan, "security-scan", &SecurityScanJob{}),
)

Eager Plan-time materialization

Every Job's Work() runs during the Pipeline's Plan(), not at runner dispatch. The orchestrator walks the entire reachable nested DAG - including transitive JobSpawn targets - before any node runs. What stays runtime-dynamic is bounded:

• Which Nodes execute (Plan-time branching on in, Job SkipIf).
• Which Steps execute (intra-Work SkipIf).
• Whether each JobSpawn fires and with what arguments.
• JobFanOutDynamic cardinality (count and keys come from the source's runtime output; the per-item shape is known).

Because the structure is reachable from source, sparkwing pipeline explain --name X and the dashboard render the full Plan -> Job -> Work -> Step tree before anything runs. The dashboard's per-Job card exposes a collapsible Work section showing inner steps and spawn declarations as placeholders (filled in once spawned children appear).

Cache

.Memoize(key, TTL(...)) turns a Job into a content-addressed cache entry. The orchestrator computes the key after upstream deps complete, looks it up across runs, and short-circuits the job on a hit, replaying the cached output without running. Misses execute normally and record (key -> output) on success. Identical content computing at the same time dedupes automatically.

sw.Job(plan, "build", &Build{}).Memoize(
    func(ctx context.Context) (sparkwing.CacheKey, error) {
        return sparkwing.Key("build", "v1"), nil
    },
    sparkwing.TTL(24*time.Hour),
)

sparkwing.Key(parts...) hashes arbitrary parts into a stable string -- use it rather than hand-concatenating. Return sparkwing.NoCache, nil to bypass memoization for one invocation. Return errors when inputs cannot be resolved. Errors, panics, empty keys, and resolution deadlines fail before dispatch. See Caching.

Caching is content only. To bound how many nodes run at once -- a mutex, a semaphore, a deploy gate -- use .Concurrency(group); see sdk.md and scheduling.

Do not cache nodes whose effect is the side effect itself (deploys, notifications, gitops commits). Caching replays the return value, not the external world - a "cached" deploy did not actually deploy anything. Cache pure builds, test runs against content-addressed sources, and artifact packaging; gate external side effects with .Needs on the cached Job.

Approval gates

Pause a run and wait for a human decision by registering a gate via sw.JobApproval. The orchestrator routes approval nodes through the approval-waiter flow, flipping the Job to approval_pending, writing an approvals row, and blocking until the dashboard, CLI, or the configured timeout resolves it.

approve := sw.JobApproval(plan, "approve-prod", sw.ApprovalConfig{
    Message:  fmt.Sprintf("Promote %s to prod?", git.SHA),
    Timeout:  2 * time.Hour,
    OnExpiry: sw.ApprovalFail,
}).Needs(integStg)
sw.Job(plan, "deploy-prod", &Deploy{Env: "prod"}).Needs(approve)

sw.JobApproval returns *ApprovalGate, a narrower handle than *JobNode -- only the modifiers that make sense for a human gate are methods on it (Needs, NeedsOptional, OnFailure, BeforeRun, AfterRun, SkipIf, Optional, ContinueOnError). Modifiers that don't apply to gates -- Retry, Timeout, Memoize, Requires, Inline -- are physically absent, so misuse is a compile error rather than a runtime panic / silent no-op.

ApprovalConfig fields:

• Message - operator-facing prompt shown in the dashboard banner and CLI list output. Compose with fmt.Sprintf if you need to weave in run-time values.
• Timeout - maximum wait before the waiter writes a timed_out resolution itself. Zero (the default) means never time out.
• OnExpiry - one of sw.ApprovalFail (default), sw.ApprovalDeny, or sw.ApprovalApprove. Unrecognized values panic at plan time.

Resolution paths:

• Dashboard: any node in approval_pending renders an indigo banner with a comment textarea and Approve / Deny buttons.
• CLI: sparkwing runs approvals approve --run ID --node ID, sparkwing runs approvals deny ....
• Programmatic: POST /api/v1/runs/{run}/approvals/{node} with {"resolution":"approved","comment":"..."}. The approver is recorded from the authenticated principal.

For a local run that must survive closing the submitting terminal, use sparkwing run <pipeline> --sw-detached. The resident consumer owns the run while it waits for approval.