The stage lifecycle

Every stage extends Stage<P> and implements a small contract.

abstract class Stage<P extends BaseStageParams = BaseStageParams> {
    abstract readonly name: string
    constructor(readonly params: P)

    async init(ctx?: InitContext): Promise<void>   // lazily, once per run
    protected abstract apply(input, row, ctx): Promise<unknown>
    protected expand(input, row, ctx): Promise<Row[] | null>  // default: null
    protected abstract onError(message: string, row: Row): unknown
    async dispose(): Promise<void>

    transform(rows, ctx): Promise<Row[]>           // final
}

init()

Called lazily, just before the first row that actually reaches the stage, and once per run. This is where a model is downloaded and a session created, which is why it is separate from apply: forty pages should not mean forty downloads. It runs on every run, so it must be idempotent.

A stage no row reaches never initialises: one that is disabled, one whose when matches nothing, or one whose every input row already failed upstream. Its model is never downloaded.

init() runs inside the error contract. A failure (a model that will not download) is recorded on that stage's rows, through failRows, and the rest of the pipeline completes; propagateError: true still throws. See The error contract.

It is also why per-row timings do not sum to the run total — init() is charged to the stage's run time, not to any row. See Timings.

apply() — one row in, one value out

The common case. Return the value to be written to outputCol; the base class handles reading inputCol, dropping it when keepInputData is off, catching errors, and recording time.

protected async apply(input: ScaleDpImage, row: Row): Promise<Document> {
    const boxes = await this.read(input)
    return createDocument({ path: input.path, type: 'ocr', text: …, bboxes: boxes })
}

expand() — one row in, several out

Return an array of rows and the multi-row path is taken instead; return null (the default) and apply is used. This is how PDF page explosion and box cropping work, and it is also how a stage writes more than one column.

Stages that implement expand leave apply throwing "unreachable" — the base class never calls both.

The elapsed time of one expand call is split evenly across the rows it produced. No finer attribution exists from outside the stage.

onError() — the load-bearing one

protected onError(message: string): Document {
    return createDocument({ exception: message })
}

Return a well-formed, empty instance of the output schema, carrying the message. Not null, not a partial object. Downstream stages and display helpers read .bboxes, .entities, .text without guarding, and the whole non-throwing contract rests on the failed column still having the right shape.

The message is already formatted by the time it arrives — formatException(name, error) produces 'StageName: OcrError: …' with the stack appended.

dispose()

Release sessions, workers and services. The base implementation does nothing; Pipeline.dispose() calls every stage's.

Scheduling hooks

The base class also carries what the scheduler needs to stream pages. Every one has a default, so a simple stage overrides none of them.

How the scheduler uses them is in Streaming; the signatures are in the Pipeline API.

Parameters

Stages never read options directly. They merge over a frozen defaults constant:

export const MY_STAGE_DEFAULTS: MyStageParams = Object.freeze({
    ...BASE_STAGE_DEFAULTS,
    inputCol: 'image',
    outputCol: 'text',
    threshold: 0.5,
})

export class MyStage extends Stage<MyStageParams> {
    readonly name = 'MyStage'

    constructor(options: Partial<MyStageParams> = {}) {
        super(resolveParams(MY_STAGE_DEFAULTS, options, {
            threshold: (value) => assertInRange('threshold', value, 0, 1),
        }))
    }
}

resolveParams ignores undefined values, so spreading an optional config never erases a default. Validators run against the fully resolved object and throw RangeError — in the constructor, where the message can name the offending field, rather than several stages into a run.

That defaults constant is also what the registry points at: spec.defaults is the frozen object itself, not a copy, so documentation and the builder's controls cannot describe a default the stage does not have.

The six every stage has