Pipelines

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

sparkwing pipeline new pins the running CLI's SDK release. Source builds without a release stamp use a published fallback. QA builds pin their QA SDK and resolve it through the QA Go proxy during dependency setup. Production builds use public SDK pins.

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 invoked pipeline name (sparkwing run build-deploy), matching ^[A-Za-z0-9][A-Za-z0-9._-]*$
• entrypoint - the SDK registration implementing it (required). Register the implementation under this name; several pipeline names may share one entrypoint.
• 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), pre_commit, pre_push, post_commit. Absent means "manual only" (a command).
• guards - gate dispatch on placement, args and git branch (when / unless token lists). Generated pipeline configuration uses these same keys.
• args - per-arg default values, keyed by CLI flag name
• profile - the named controller connection in the machine config. A pinned pipeline runs through pipeline trigger --profile.
• requires - runner-label requirements inherited by every job
• prefers - soft runner-label preferences
• source - the revision supplying pipeline code
• 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.

For name: deploy-prod with entrypoint: Deploy, use deploy-prod in commands. Register the SDK implementation as sw.Register("Deploy", factory). Scaffolds register the configured entrypoint. A run of deploy-prod whose entrypoint has no registration fails with unknown_pipeline, and its hint names the entrypoint the host looked up. Help, flag completion, secrets and risk declarations come from that registration. Help, argument errors and RunContext.Pipeline keep the invoked name.

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

  # 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:
        - name: nightly
          cron: "0 2 * * *"
          where: local

branches / paths / actions record intent: the controller does not read your sparkwing.yaml, so it dispatches the pipelines a team subscribed to the repository through the GitHub App. To require a checked-out branch before any step runs, add guards: {when: [git:branch=main]} with its literal name. For pull requests this compares the head branch, not the base branch. git:branch=default matches the repository's default branch, which a GitHub App trigger records and any other run reads from its checkout; a run that cannot resolve it is refused. 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.

GitHub delivery is handled by the controller through the GitHub App. 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, Memo, 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 its own Job 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 before dispatch so inspection can show it.

Cost grid

Only Plan-layer Job* adders create Jobs. A unit that needs Job-only modifiers (Retry, Requires, distinct runner) is its own Job; a set of Jobs decided by an earlier Job's output uses source.Generate.

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: generated jobs

job.Generate declares work after a source result becomes available. Read the source's typed output inside the generator callback and add jobs to its supplied plan. A downstream job uses NeedsGenerated to wait for the accepted expansion. See Splitting work into Jobs for an example.

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 Memo 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.

Splitting work into Jobs

When part of a job needs its own retry, runner or memo key, declare it as a separate job and order it with Needs. When jobs depend on runtime output, register a generator on the source:

analyze := sw.Job(plan, "analyze", &AnalyzeJob{})
targets := sw.RefTo[[]string](analyze)
generated := analyze.Generate(func(ctx context.Context, expansion *sw.Plan) error {
    for _, target := range targets.Get(ctx) {
        sw.Job(expansion, "scan-"+target, &ComplianceJob{Target: target})
    }
    return nil
})
sw.Job(plan, "finish", func(context.Context) error { return nil }).
    NeedsGenerated(generated)

Each generated job has its own state, attempts and placement. The host validates and persists expansions before scheduling them. See Generated jobs.

Modifier scope discipline

A Step that needs Retry / Timeout / Requires is the canonical signal to promote it to a Job of its own.

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.

Generated topology

A generator's callback decides its jobs from the source output. Initial plan inspection includes the registered generator, while accepted expansions carry the actual jobs. The host persists empty expansions and checks declaration hashes when reconstructing code.

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 before any node runs. What stays runtime-dynamic is bounded:

• Which jobs execute (Plan-time branching on in, Job SkipIf).
• Which Steps execute (intra-Work SkipIf).
• Generator 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 plan --pipeline 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 its inner steps.

Cache

.Memo(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{}).Memo(
    func(ctx context.Context) (sparkwing.MemoValue, error) {
        return sparkwing.MemoKey("build", "v1"), nil
    },
    sparkwing.MemoTTL(24*time.Hour),
)

sparkwing.MemoKey(parts...) collects ordered JSON values for the executing host to hash. Pass an inputs.Compose(...) recipe directly to Memo for built-in file, tree, environment and literal inputs. Return sparkwing.NoMemo, nil to bypass memoization for one invocation. Return errors when inputs cannot be resolved. Errors, panics, empty keys, and resolution deadlines fail on the executing host before the body runs. 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.

Secrets

A pipeline declares its secrets with a Secrets() struct, each field required unless its sw tag ends in ,optional (see Declaring secrets). The declaration is checked before the first node runs, so a run never fails halfway through a deploy for a secret it could have named at the start.

• On a controller, the run's plan is refused when a required secret is neither set for the run's repository and pipeline (read only by runs whose code Sparkwing fetched from the repository) nor set team-wide and shared with runs. A child at its parent's exact revision reads its parent pipeline's secrets, and any other child only shared ones (see secrets). The run fails before any of its nodes runs. Its error names the declared secrets the run cannot read, and sparkwing runs status --run RUN_ID --profile PROFILE shows the team which scope each is missing from and the command that fixes it: sparkwing secrets set --secret NAME --file <path> --pipeline PIPELINE --profile PROFILE for this pipeline only, or --shared instead of --pipeline for a team-wide secret every pipeline may read. That detail is a secret_preflight_refused run event, read with the team's runs.read access and never by the run's claim tokens, because it names secrets and pipelines the run has no access to.
• On a laptop, sparkwing run checks the same declarations against the local controller and reports every missing secret at once with the command or setting that supplies it.

Optional declarations stay optional during planning and plan acceptance. A missing optional secret never stops a run; its field stays empty and Secret returns sparkwing.ErrSecretMissing for it. Optional secrets use the same scope checks as required secrets. An optional secret outside the run's repository or pipeline scope, or unavailable to its source provenance, also reads as missing (undefined in TypeScript). The result does not reveal whether another pipeline has a value. On a controller a node may read only the secrets its pipeline declares, and reading any other name fails with an error naming the sw tag to add.

Approval gates

Pause a run and wait for a human decision by registering a gate via plan.Approval. 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 := plan.Approval("approve-prod", sw.ApprovalConfig{
    Message:  fmt.Sprintf("Promote %s to prod?", git.SHA),
    Timeout:  2 * time.Hour,
    OnTimeout: sw.ApprovalFail,
}).Needs(integStg)
sw.Job(plan, "deploy-prod", &Deploy{Env: "prod"}).Needs(approve)

plan.Approval returns *Approval, a narrower handle than *JobNode -- only the modifiers that make sense for a human gate are methods on it (Needs, NeedsIfPresent, NeedsGenerated, OnFailure, SkipIf, Optional, ContinueOnError). Modifiers that don't apply to gates -- Retry, Timeout, Memo, 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.
• OnTimeout - 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 --job 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. The local controller checks approval deadlines every second. Cancelling the run removes its gate from the pending approval list. sparkwing runs approvals list --run ID still includes its request history.

Repository fleet history comes from the local controller's public run projection. A failed history read is an error. Repository commands do not open the runs store directly or treat an unavailable history source as an empty result.