Pipeline and Stage

`Pipeline`

class Pipeline {
    constructor(stages: readonly (Stage | PipelineStage)[])
    readonly stages: Stage[]
    readonly entries: readonly PipelineEntry[]

    transform(input: PipelineInput, options?: PipelineOptions & RunOptions): Promise<Row[]>  // stage order
    run(input: PipelineInput, options?: RunOptions & PipelineOptions): Promise<Row[]>        // page order
    events(input: PipelineInput, options?: Omit<RunOptions, 'onEvent'>): AsyncIterable<PipelineEvent> & { result: Promise<Row[]> }
    dispose(): Promise<void>
}

interface PipelineStage {
    stage: Stage
    id?: string          // default: the stage name, or `Name#index` when it repeats
    enabled?: boolean    // default true; RunOptions.enabled overrides per run
    when?: Condition     // run on matching rows only
    barrier?: boolean    // wait for every page before this stage
}

interface PipelineEntry {               // a PipelineStage, resolved
    readonly stage: Stage
    readonly id: string
    readonly index: number              // position, counting disabled stages
    readonly timingKey: string          // key in execution_time.stages
    readonly enabled: boolean
    readonly when?: Condition
    readonly barrier: boolean
}

A StageDescriptor with these fields, plus runtime, builds the same thing from JSON — see pipelineFromDescriptors.

interface PipelineOptions {
    signal?: AbortSignal
    onStage?: (name: string, ms: number, rows: number) => void
}

interface RunOptions {
    order?: 'page' | 'stage'                      // run(): 'page', transform(): 'stage'
    signal?: AbortSignal
    pagesInFlight?: number                        // default 1
    enabled?: Record<string, boolean>             // by stage id or name
    vars?: Record<string, Json>                   // for `{ var }` conditions
    stages?: Record<string, StageRunOptions>      // by stage id or name
    initRetry?: 'run' | 'row'                     // default 'run'
    rowUpdates?: { payload?: 'delta' | 'full' | 'none'; stages?: string[]; stripBinary?: boolean }
    onEvent?: (event: PipelineEvent) => void
    onRow?: (update: RowUpdate) => void
}

interface StageRunOptions {
    maxConcurrency?: number   // can only lower the stage's own
}

Both orders return the same rows in the same order; every returned row gets execution_time. A stage is initialised lazily, just before the first row that reaches it, and a failed init lands on that stage's rows rather than throwing. See Streaming runs for events, row updates, when and barriers.

entries is fixed when the pipeline is constructed, and that is what runs: stages is a copy taken from it, so pushing onto stages adds nothing to a run. To add an annotation pass, build a new pipeline — new Pipeline([...pipeline.entries, new ImageDrawBoxes()]) keeps every entry's settings. Use RunOptions.enabled to switch a stage off without rebuilding.

`Stage`

abstract class Stage<P extends BaseStageParams = BaseStageParams> {
    abstract readonly name: string
    readonly uid: number                           // unique per instance
    constructor(readonly params: P)

    async init(ctx?: InitContext): Promise<void>   // lazy, idempotent; ctx.onProgress for this stage
    protected abstract apply(input, row, ctx): Promise<unknown>
    protected expand(input, row, ctx): Promise<Row[] | null>
    protected abstract onError(message: string, row: Row): unknown
    async dispose(): Promise<void>

    transform(rows: Row[], ctx: StageContext): Promise<Row[]>

    // Scheduling hooks, all with safe defaults
    get scope(): 'row' | 'group' | 'document'   // 'row'; 'document' makes the stage a barrier
    get maxConcurrency(): number                // 1; raise for pure or network-bound stages
    resources(): ResourceClaim[]                // []; shared sessions a call holds
    planPages(row, ctx): Promise<number[] | null>   // null; PDF stages list their pages
    requiredModels(): Promise<ModelRequirement[]>   // []; the models init() will download
    writes(): string[]                          // [outputCol]; what a row update carries
    reads(): string[]                           // [inputCol, ...inputCols]
    runtime: StageRuntime                       // { executionProviders? }
    withRuntime(runtime: StageRuntime | undefined): this

    failRows(rows: readonly Row[], message: string, ms?: number): Row[]   // every row as if apply had failed
}

requiredModels() lists, from the params alone, the model files the stage will fetch when it first runs — nothing is downloaded. Engine-free stages return []. It is what pipelineModels gathers, and a custom stage that loads a model should override it so the pipeline's downloads can be shown, prefetched and evicted before a run. See Managing models.

runtime is read when the stage first builds its session, so set it (or call withRuntime) before the stage runs. Thread count is not in it: onnxruntime-web fixes one per realm, so stages that need different counts go in different workers — see Performance.

failRows is what the scheduler uses when a stage cannot start at all (a failed init): each row gets the stage's onError value, honouring keepInputData, and is charged ms spread across the rows.

name is also the StageDescriptor.type. The full contract, including why onError must return a well-formed empty schema instance, is in The stage lifecycle.

Rows and input

type Row = Record<string, unknown>

type PipelineInput =
    | Uint8Array | ArrayBuffer | Blob | File | string | Row | Row[]

interface InitContext { stageId?: string; signal?: AbortSignal; onProgress?: (p: ModelProgress) => void }
interface StageContext extends InitContext { index: number; runId?: string }

import { toRows } from '@stabrise/scaledp'

toRows is exported for anyone building rows by hand. Arrays pass through shallow-copied; a string is fetched and throws on a non-ok response; bytes become { content, path: 'memory' }; a File uses its name as the path.

Timing columns

const EXECUTION_TIME_COL = 'execution_time'
const ROW_TIME_COL = 'row_time'

interface ExecutionTime { stages: Record<string, number>; total: number }
interface RowTime { stages: Record<string, number>; total: number }

See Timings for what each measures and why they differ.

Params

import { BASE_STAGE_DEFAULTS, resolveParams, assertInRange, assertPositiveInt } from '@stabrise/scaledp'

interface BaseStageParams {
    inputCol: string        // 'content'
    outputCol: string       // 'output'
    pathCol: string         // 'path'
    pageCol: string         // 'page'
    keepInputData: boolean  // false
    propagateError: boolean // false
}

type Validator<T> = { [K in keyof T]?: (value: T[K], all: T) => void }
resolveParams<T>(defaults: Readonly<T>, options?: Partial<T>, validators?: Validator<T>): T

resolveParams ignores undefined values, so spreading an optional config never erases a default. Validators run against the resolved object and throw RangeError.

Errors

import {
    ScaleDpError, ImageError, OcrError, DetectionError, NerError, ConfigError,
    formatException,
} from '@stabrise/scaledp'

All extend ScaleDpError, which carries the stage it came from. formatException(stage, error) produces the string that lands in an output schema's exception. See The error contract.

Pipeline