@nightowlsdev/pipeline

Adapter/Storage

A separate, opt-in pipeline plane — a data-driven, fan-out-capable structured-extraction DAG whose stages can be executed by a NAMED, GOVERNED swarm agent (cost cap + tool gate + rules + metering), not just a raw LLM prompt.

What it does

The pipeline plane (FR-067): a data-driven structured-extraction DAG stored as data (`{ start, stages }`) and walked by a cursor that follows each stage's `onPass`/`onFail`/`guard` edges — a SEPARATE primitive from `defineWorkflow` (a workflow is a single-cursor agent/tool/human orchestration; a pipeline is a fan-out-capable extraction DAG for the ingest → extract-per-item → score → route shape). The differentiator over a generic prompt-DAG: a stage of `kind: "agent"` runs a REAL, named swarm agent through core's `runAgent`, so the step inherits the engine's cost cap, tool gate, rules, approval, and `swarm.turn_usage` metering — a governed extraction, not a raw prompt. Stage kinds are `agent` / `model` (an ephemeral governed prompt) / `tool` (a host executor) / `code` (a pure function) / `fan_out` (a bounded-concurrency map of a sub-stage over an array). `runPipeline({ pipeline, context, budgetUsd?, deps })` runs a graph headless (background/Trigger-friendly, no chat surface); `definePipeline(graph)` authors + validates a code-defined graph. `validatePipelineGraph` is the publish-time gate the workflow validator lacks: it checks that every stage/edge/agent a graph names actually EXISTS (referent existence) as well as acyclicity, so a broken graph is a 422 at author time, not a failure on the 10,000th run. `budgetUsd` is a HARD ceiling across all stages, fan-out spend included; a `schemaMode: "enforce"` output schema is validated and a violating reply fails the stage rather than flowing downstream. Library-shaped like `@nightowlsdev/metrics`/`tasks`: you own the `pg` Pool (an OPTIONAL peer — only `createPipelineStore` needs it), the package owns its `pipeline_0001_*` migration set and never runs DDL, and the four tables (`nightowls.pipelines`/`pipeline_versions`/`pipeline_runs`/`pipeline_stage_runs`) FK `nightowls.orgs`, so apply `@nightowlsdev/storage-supabase` through `0013_rename_schema` first. OPT-IN by construction: because the tables ship in THIS package's migration set (not storage-supabase's mandatory set), a host that never installs it never receives the DDL — the plane is additive to every existing engine adopter, with no change to `runs`/`run_steps`. Server-only posture: RLS on with no policy (the runner uses the service connection). The admin REST surface is `runner.pipelineRoutes()` (behind `adminRouteGate("pipelines")`, a default-deny scope — authoring/firing is refused server-side for a non-admin), and the Studio section is `pipelinesPlugin` from `@nightowlsdev/react` (a graph editor with a live validator + the agent stage-kind picker + a realtime run transcript). Engine-wall clean: peer `@nightowlsdev/core` + optional `pg`, no Mastra types in the public surface.

Install

pnpm add @nightowlsdev/pipeline

Key exports

runPipeline (headless, background/Trigger-friendly) / definePipeline (author + validate a code-defined graph — the sibling to defineWorkflow)
validatePipelineGraph — publish-time validation of referent EXISTENCE and acyclicity (the B10 gap defineWorkflow's validator lacks)
createPipelineStore({ pool }) (Postgres PipelineAdminStore) / InMemoryPipelineStore (the Postgres-free twin)
PIPELINE_MIGRATIONS / M_PIPELINE_0001 / PIPELINE_MIGRATION_SQL — the own, opt-in prefixed migration set (pipeline_0001_*) / nightOwlsPlugin manifest
validateJsonSchema (the schemaMode: 'enforce' output-schema validator) / resolvePath / renderTemplate
types: PipelineGraph, PipelineStage, StageKind, PipelineDeps, PipelineAdminStore, PipelineRunRow, PipelineStageRunRow

Usage

pipeline.ts

import { runPipeline, definePipeline } from "@nightowlsdev/pipeline";
import { openaiModels } from "@nightowlsdev/provider-openai";

// A pipeline is a GRAPH stored as data: { start, stages }. Each stage is one of agent | model | tool | code |
// fan_out, with onPass / onFail / guard edges. The differentiator: a "agent" stage runs a NAMED, GOVERNED swarm
// agent through runAgent (cost cap + tool gate + rules + turn_usage metering), not just a raw LLM prompt.
const graph = definePipeline({
  start: "extract",
  stages: {
    extract: {
      kind: "agent",                          // ← a governed swarm agent runs this stage
      agent: "invoice-extractor",
      outputSchema: { type: "object", required: ["items"] },
      schemaMode: "enforce",                  // the reply MUST satisfy the schema, or the stage retries then fails
      input: { text: { kind: "ref", path: "context.text" } },
      onPass: "score",
      onFail: { halt: true, reason: "extraction failed" },
    },
    score: {
      kind: "code",
      code: "score",
      input: { items: { kind: "ref", path: "stages.extract.output.items" } },
      onPass: null,                           // null ⇒ the run ends succeeded here
      onFail: { halt: true, reason: "scoring failed" },
    },
  },
}); // definePipeline VALIDATES referent existence AND acyclicity — the B10 gap defineWorkflow's validator lacks.

const result = await runPipeline({
  pipeline: graph,
  context: { text: invoiceText },
  budgetUsd: 2,                               // a REAL ceiling across ALL stages, fan-out spend included
  deps: {
    modelFactory: openaiModels(),
    resolveAgent: (slug) => agentRegistry[slug] ?? null,
    codeStages: { score: (input) => ({ score: (input.items as unknown[]).length }) },
    ctx: { tenantId: orgId, userId, agentSlug: "pipeline", runId: crypto.randomUUID(), threadId },
  },
});
// → { runId, status: "succeeded" | "failed" | "halted" | "cancelled" }

// Durable runs: createPipelineStore({ pool }) (own opt-in migrations, pipeline_0001_*). Admin REST:
// runner.pipelineRoutes() behind adminRouteGate("pipelines"). Studio: pipelinesPlugin. Full guide → /docs/pipelines