A pipeline your agents can run. Structured extraction as a governed DAG, not a prompt chain.

pnpm add @nightowlsdev/pipeline @nightowlsdev/core
pnpm add pg                # optional — only the durable Postgres store needs it
owl install pipeline       # ejects pipeline_0001_* into supabase/migrations/
supabase db push           # you apply it, with your own tooling

import { runPipeline, definePipeline } from "@nightowlsdev/pipeline";
import type { PipelineDeps, PipelineGraph } from "@nightowlsdev/pipeline";
import type { ModelFactory, AgentDef, SwarmContext } from "@nightowlsdev/core";

declare const modelFactory: ModelFactory;             // openaiModels() / createModelFactory({ ... })
declare const agentRegistry: Record<string, AgentDef>;
declare const ctx: SwarmContext;                      // the AUTHENTICATED tenant / user / agentSlug / run / thread
declare const invoiceText: string;

// A pipeline is a GRAPH stored as data. definePipeline VALIDATES referent existence AND acyclicity — a broken
// graph throws HERE (author time), never on the 10,000th run (the gap defineWorkflow's validator lacks).
const graph: PipelineGraph = definePipeline({
  start: "extract",
  stages: {
    extract: {
      kind: "agent",                                  // ← a NAMED, GOVERNED swarm agent runs this stage
      agent: "invoice-extractor",
      outputSchema: { type: "object", required: ["items"] },
      schemaMode: "enforce",                          // a reply that violates the schema retries, then FAILS the stage
      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" },
    },
  },
});

const deps: PipelineDeps = {
  modelFactory,
  resolveAgent: (slug) => agentRegistry[slug] ?? null,      // an "agent" stage runs THIS agent via runAgent
  codeStages: { score: (input) => ({ score: (input.items as unknown[]).length }) },
  ctx,
};

const result = await runPipeline({
  pipeline: graph,
  context: { text: invoiceText },
  budgetUsd: 2,                                       // a REAL ceiling across ALL stages, fan-out spend included
  deps,
});
// result.status ∈ "succeeded" | "failed" | "halted" | "cancelled"

import { createPipelineStore, InMemoryPipelineStore } from "@nightowlsdev/pipeline";
import type { PipelineAdminStore } from "@nightowlsdev/pipeline";

declare const pool: import("pg").Pool;

// Durable: you own the Pool (pg is an OPTIONAL peer — only this needs it). The tables are RLS-on with no policy
// (server-only), so the Pool must OWN them or hold BYPASSRLS — the runner uses the service connection.
const store: PipelineAdminStore = createPipelineStore({ pool });

// The Postgres-free twin — the SAME admin contract, for dev and tests.
const memory: PipelineAdminStore = new InMemoryPipelineStore();

// Reads are org-scoped: list an org's pipelines, or one run + its per-stage transcript.
const pipelines = await store.listPipelines("00000000-0000-4000-8000-00000000abcd");
const run = await store.getRun("00000000-0000-4000-8000-00000000abcd", "run-uuid");

// app/api/swarm/pipelines/route.ts
import { createNextjsRunner } from "@nightowlsdev/runner-nextjs";
import { createPipelineStore, validatePipelineGraph, runPipeline } from "@nightowlsdev/pipeline";

const runner = createNextjsRunner({
  engine, auth, storage,
  pipelinePlane: {
    store: createPipelineStore({ pool }),
    validateGraph: validatePipelineGraph,          // the B10 gate, enforced SERVER-SIDE (a broken graph → 422)
    // OPTIONAL — firing a run. Wire your own runPipeline call (or a durable enqueue). Absent ⇒ fire answers 404.
    fire: async (ctx, { slug, context, budgetUsd, dryRun }) => {
      const r = await runPipeline({ pipeline: slug, context, budgetUsd, dryRun, deps: pipelineDeps(ctx) });
      return { runId: r.runId, status: r.status };
    },
  },
});

// Each route file mounts one family — all behind adminRouteGate("pipelines") (a default-deny scope):
export const { GET, POST } = runner.pipelineRoutes().library;    // /api/swarm/pipelines
export const { GET }       = runner.pipelineRoutes().runs;       // /api/swarm/pipelines/runs
export const { GET }       = runner.pipelineRoutes().runDetail;  // /api/swarm/pipelines/runs/[id]

import { AIStudio, pipelinesPlugin } from "@nightowlsdev/react";

// "Pulled in on demand" — the section only appears when you register it (tree-shaken otherwise), and only when
// its routes are mounted (discover only-404-degrades). Authoring/firing is refused SERVER-SIDE for a non-admin.
<AIStudio plugins={[pipelinesPlugin]} pluginClientOptions={{ base: "/api/swarm", getToken }} />