Pipeline Lifecycle: runPipeline and advancePhase Explained

type RunPipelineArgs<TState> = {
  jobId: string;
  phases: Phase<TState>[];          // ordered array, must be non-empty
  storage: StorageAdapter<TState>;
  scheduler: SchedulerAdapter;
  retryMode?: "resume" | "full";    // default: "resume" if checkpoint exists
  initialState: TState;
  initialPhase?: string;            // override first phase name
};

import type { Phase } from "@claritylabs/cl-pipelines";
import {
  runPipeline,
  advancePhase,
  createMemoryStorage,
  createMemoryScheduler,
} from "@claritylabs/cl-pipelines";

type DocState = { text: string; wordCount?: number };

const extract: Phase<DocState> = {
  name: "extract",
  run: async (ctx) => {
    const words = ctx.checkpoint.state.text.split(/\s+/).length;
    await ctx.log(`extracted ${words} words`);
    return {
      kind: "next",
      nextPhase: "summarise",
      state: { ...ctx.checkpoint.state, wordCount: words },
    };
  },
};

const summarise: Phase<DocState> = {
  name: "summarise",
  run: async (ctx) => {
    await ctx.log(`summarising ${ctx.checkpoint.state.wordCount} words`);
    return { kind: "done" };
  },
};

const storage = createMemoryStorage<DocState>();
const scheduler = createMemoryScheduler();

scheduler._bind((jobId) =>
  advancePhase({ jobId, phases: [extract, summarise], storage, scheduler })
);

await runPipeline({
  jobId: "doc-42",
  phases: [extract, summarise],
  storage,
  scheduler,
  initialState: { text: "Hello world" },
});

await scheduler.drain();