Chapter 10

pipe: reusable task groups

Pipelines accumulate common sequences. .pipe() extracts them into functions while keeping the chain flat and the types intact.

A pipe function takes a SequenceBuilder<Ctx> and returns one with states appended. The interesting part is the generic constraint: it declares what the context must already contain, without caring what else is in there.

const addCreateAtlas = <
  Ctx extends { 
extractFrames: {
    frameStorageRefs: StorageRefType[];
}extractFrames: { 
frameStorageRefs: {
    bucket: string;
    key: string;
}[]frameStorageRefs: 
type StorageRefType = {
    bucket: string;
    key: string;
}StorageRefType[] } }
>(
  b: SequenceBuilder<Ctx, Ctx, []>b: class SequenceBuilder<in out Ctx, Base = Ctx, E extends readonly StateEntry[] = []>
Builds a sequence of Step Function states with type-safe context accumulation.

Each `.task()` call appends a Lambda Task state and expands the context
type with that state's output. The payload callback receives a typed proxy
of the current context, so every ref is validated at compile time.

`.build()` wires up `Next`/`End` pointers and returns the ASL structure.
@example```ts
type Input = { bucket: string; key: string };

const result = new SequenceBuilder<Input>()
  .task('runMediaInfo', {
    inputSchema: RunMediainfoStepInput,
    outputSchema: RunMediainfoStepOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    bucket: ctx.bucket,
    key: ctx.key,
  }))
  .task('createVideo', {
    inputSchema: CreateVideoInput,
    outputSchema: CreateVideoOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    mediaInfo: ctx.runMediaInfo.mediaInfo,
  }))
  .build();
```SequenceBuilder<Ctx>
) =>
  b: SequenceBuilder<Ctx, Ctx, []>b.task(
    'createAtlas',
    {
      inputSchema: CreateAtlasInput,
      outputSchema: 
const CreateAtlasOutput: z.ZodObject<{
    atlasStorageRef: z.ZodObject<{
        bucket: z.ZodString;
        key: z.ZodString;
    }, z.core.$strip>;
}, z.core.$strip>CreateAtlasOutput,
      LambdaTaskConfig<I extends AnyZodObject, O extends AnyZodObject>.functionArn: stringfunctionArn: const LAMBDA_ARN: "${lambda_arn}"LAMBDA_ARN,
    },
    (ctx: Proxied<Ctx>ctx) => ({
      step: "create-atlas"step: 'create-atlas' as type const = "create-atlas"const,
      
frameStorageRefs: Proxied<{
    bucket: string;
    key: string;
}[]>frameStorageRefs: ctx: Proxied<Ctx>ctx.
extractFrames: Proxied<{
    frameStorageRefs: StorageRefType[];
}>extractFrames.
frameStorageRefs: Proxied<{
    bucket: string;
    key: string;
}[]>frameStorageRefs,
      outputFilename: stringoutputFilename: 'atlas.webp',
    })
  );

Ctx extends { extractFrames: { frameStorageRefs: StorageRefType[] } } says: usable in any pipeline that has already extracted frames. Nothing more.

Using it

const const asl: AslStateMachineasl = new new SequenceBuilder<Input, Input, []>(): SequenceBuilder<Input, Input, []>
Builds a sequence of Step Function states with type-safe context accumulation.

Each `.task()` call appends a Lambda Task state and expands the context
type with that state's output. The payload callback receives a typed proxy
of the current context, so every ref is validated at compile time.

`.build()` wires up `Next`/`End` pointers and returns the ASL structure.
@example```ts
type Input = { bucket: string; key: string };

const result = new SequenceBuilder<Input>()
  .task('runMediaInfo', {
    inputSchema: RunMediainfoStepInput,
    outputSchema: RunMediainfoStepOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    bucket: ctx.bucket,
    key: ctx.key,
  }))
  .task('createVideo', {
    inputSchema: CreateVideoInput,
    outputSchema: CreateVideoOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    mediaInfo: ctx.runMediaInfo.mediaInfo,
  }))
  .build();
```SequenceBuilder<
type Input = {
    bucket: string;
    key: string;
}Input>()
  .task(
    'extractFrames',
    {
      inputSchema: 
const ExtractFramesInput: z.ZodObject<{
    step: z.ZodLiteral<"extract-frames">;
    bucket: z.ZodString;
    key: z.ZodString;
}, z.core.$strip>ExtractFramesInput,
      outputSchema: 
const ExtractFramesOutput: z.ZodObject<{
    frameStorageRefs: z.ZodArray<z.ZodObject<{
        bucket: z.ZodString;
        key: z.ZodString;
    }, z.core.$strip>>;
}, z.core.$strip>ExtractFramesOutput,
      LambdaTaskConfig<I extends AnyZodObject, O extends AnyZodObject>.functionArn: stringfunctionArn: const LAMBDA_ARN: "${lambda_arn}"LAMBDA_ARN,
    },
    (ctx: Proxied<Input>ctx) => ({
      step: "extract-frames"step: 'extract-frames' as type const = "extract-frames"const,
      bucket: Ref<string>bucket: ctx: Proxied<Input>ctx.bucket: Ref<string>bucket,
      key: Ref<string>key: ctx: Proxied<Input>ctx.key: Ref<string>key,
    })
  )
  .pipe(addCreateAtlas)
  .pipe(addGenerateEmbedding)
  .build();

The chain stays flat — ExtractFrames → CreateAtlas → GenerateEmbedding — and createAtlas's payload still resolves to "$.extractFrames.frameStorageRefs". Nothing about the wiring changed; only where the code lives.

The context flows through

Piped functions widen the context exactly like inline calls, so later steps see everything:

const builder = new new SequenceBuilder<Input, Input, []>(): SequenceBuilder<Input, Input, []>
Builds a sequence of Step Function states with type-safe context accumulation.

Each `.task()` call appends a Lambda Task state and expands the context
type with that state's output. The payload callback receives a typed proxy
of the current context, so every ref is validated at compile time.

`.build()` wires up `Next`/`End` pointers and returns the ASL structure.
@example```ts
type Input = { bucket: string; key: string };

const result = new SequenceBuilder<Input>()
  .task('runMediaInfo', {
    inputSchema: RunMediainfoStepInput,
    outputSchema: RunMediainfoStepOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    bucket: ctx.bucket,
    key: ctx.key,
  }))
  .task('createVideo', {
    inputSchema: CreateVideoInput,
    outputSchema: CreateVideoOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    mediaInfo: ctx.runMediaInfo.mediaInfo,
  }))
  .build();
```SequenceBuilder<
type Input = {
    bucket: string;
    key: string;
}Input>()
  .task(
    'extractFrames',
    {
      inputSchema: 
const ExtractFramesInput: z.ZodObject<{
    step: z.ZodLiteral<"extract-frames">;
    bucket: z.ZodString;
    key: z.ZodString;
}, z.core.$strip>ExtractFramesInput,
      outputSchema: 
const ExtractFramesOutput: z.ZodObject<{
    frameStorageRefs: z.ZodArray<z.ZodObject<{
        bucket: z.ZodString;
        key: z.ZodString;
    }, z.core.$strip>>;
}, z.core.$strip>ExtractFramesOutput,
      LambdaTaskConfig<I extends AnyZodObject, O extends AnyZodObject>.functionArn: stringfunctionArn: const LAMBDA_ARN: "${lambda_arn}"LAMBDA_ARN,
    },
    (ctx: Proxied<Input>ctx) => ({
      step: "extract-frames"step: 'extract-frames' as type const = "extract-frames"const,
      bucket: Ref<string>bucket: ctx: Proxied<Input>ctx.bucket: Ref<string>bucket,
      key: Ref<string>key: ctx: Proxied<Input>ctx.key: Ref<string>key,
    })
  )
  .pipe(addCreateAtlas);

type Ctx = 
type InferContext<B extends AnyBuilder> = B extends {
    _ctx: infer Ctx;
} ? Ctx : never
Extract the accumulated context type from a SequenceBuilder.
@example```ts
const builder = new SequenceBuilder<{ bucket: string }>()
  .task('runMediaInfo', config, ctx => ({ ... }))
  .task('createVideo', config, ctx => ({ ... }));

type Output = InferContext<typeof builder>;
// = { bucket: string; runMediaInfo: MediaInfoOutput; createVideo: VideoOutput }
```InferContext<typeof builder>;
    
type Ctx = {
    extractFrames: {
        frameStorageRefs: {
            bucket: string;
            key: string;
        }[];
    };
    bucket: string;
    key: string;
    createAtlas: {
        atlasStorageRef: {
            bucket: string;
            key: string;
        };
    };
}

The constraint is enforced

Piping a function into a builder that doesn't satisfy its constraint is a compile error, which is what makes these safe to move between pipelines:

new 
new SequenceBuilder<{
    bucket: string;
}, {
    bucket: string;
}, []>(): SequenceBuilder<{
    bucket: string;
}, {
    bucket: string;
}, []>
Builds a sequence of Step Function states with type-safe context accumulation.

Each `.task()` call appends a Lambda Task state and expands the context
type with that state's output. The payload callback receives a typed proxy
of the current context, so every ref is validated at compile time.

`.build()` wires up `Next`/`End` pointers and returns the ASL structure.
@example```ts
type Input = { bucket: string; key: string };

const result = new SequenceBuilder<Input>()
  .task('runMediaInfo', {
    inputSchema: RunMediainfoStepInput,
    outputSchema: RunMediainfoStepOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    bucket: ctx.bucket,
    key: ctx.key,
  }))
  .task('createVideo', {
    inputSchema: CreateVideoInput,
    outputSchema: CreateVideoOutput,
    functionArn: LAMBDA_ARN,
  }, ctx => ({
    mediaInfo: ctx.runMediaInfo.mediaInfo,
  }))
  .build();
```SequenceBuilder<{ bucket: stringbucket: string }>().pipe(addCreateAtlas);
Argument of type '<Ctx extends { extractFrames: { frameStorageRefs: StorageRefType[]; }; }>(b: SequenceBuilder<Ctx>) => Widened<Ctx, [], "createAtlas", { atlasStorageRef: { bucket: string; key: string; }; }>' is not assignable to parameter of type '(builder: SequenceBuilder<{ bucket: string; }, { bucket: string; }, []>) => SequenceBuilder<{ extractFrames: { frameStorageRefs: { bucket: string; key: string; }[]; }; createAtlas: { atlasStorageRef: { ...; }; }; }, any, any>'.
  Types of parameters 'b' and 'builder' are incompatible.
    Type 'SequenceBuilder<{ bucket: string; }, { bucket: string; }, []>' is not assignable to type 'SequenceBuilder<{ extractFrames: { frameStorageRefs: { bucket: string; key: string; }[]; }; }, { extractFrames: { frameStorageRefs: { bucket: string; key: string; }[]; }; }, []>'.
      Types of property '_ctx' are incompatible.
        Property 'extractFrames' is missing in type '{ bucket: string; }' but required in type '{ extractFrames: { frameStorageRefs: { bucket: string; key: string; }[]; }; }'.

The pipeline never extracted frames, so the atlas step can't run — caught at the call site rather than on the first execution.