#@shipplane/core

Workflow state machine for DAG task execution. Core owns workflow validation, scheduling, retries, timeouts, cancellation, logs, events, and snapshots.

Core does not know about workers, threads, HTTP, SSH, or persistence — see @shipplane/orchestrator for dispatching tasks to workers.

#Install

npm install @shipplane/core

#Usage

import { createRunner, createTaskRegistry, workflow } from "@shipplane/core";

const registry = createTaskRegistry()
  .register("fetch-user", {
    run: async () => "email:a@example.com",
  })
  .register("send-email", {
    run: async (ctx) => {
      await ctx.log("sending email", { to: ctx.outputs.fetch });
    },
  });

const wf = workflow("welcome")
  .task("fetch", { type: "fetch-user" })
  .task("send", { type: "send-email", dependsOn: ["fetch"] })
  .withConfig({
    concurrency: 2,
    retry: { maxAttempts: 2 },
    timeoutMs: 30_000,
  });

const runner = createRunner({ workflow: wf, registry });

await runner.start();

const snapshot = runner.snapshot();
console.log(snapshot.state.status);

#Public API

• workflow(id) — creates an immutable workflow builder.
• workflow.fromJSON(definition) — hydrates a workflow builder from JSON.
• validateWorkflowDefinition(definition) — validates task ids, dependencies, cycles, config, and JSON-shaped data.
• createTaskRegistry() — creates an immutable task implementation registry.
• createRunner({ workflow, registry, onEvent }) — creates a workflow runner.
• createRunner({ snapshot, registry, recovery, onEvent }) — resumes a runner from a snapshot.

#Runner methods

• start()
• resume()
• retryInterrupted()
• cancel()
• cancelTask(taskId)
• snapshot()
• on("event", handler)
• getLogs(taskId)
• getLastLog(taskId)

#Task context

Task handlers receive:

• runId, taskId, attempt
• params
• outputs from direct dependencies
• signal for cancellation and timeout
• log(message, meta?)
• call(fn, input) for abort-aware helper calls

Task output must be string | void.