Add Standard Schema output validation to flow steps

[jumski]

Summary

Add optional runtime output validation to flow handlers using the Standard Schema interface.

Proposed DSL:

• .step({ outputSchema }) validates the handler's complete output.
• .array({ outputSchema }) validates the complete returned array.
• .map({ itemOutputSchema }) validates each map task's returned item before aggregation.
• onOutputInvalid: 'retry' | 'fail' controls whether a schema violation uses normal retry handling or immediately hard-fails the step/run. Default: 'retry'.

Schemas stay in the runtime flow definition. They are not converted to JSON Schema, compiled into migrations, persisted in Postgres, or included in flow-shape comparison.

Flow input validation is intentionally out of scope for the initial implementation. Users can model it explicitly as a normal validation step.

Why

TypeScript checks handler types only during development. At runtime, handlers can return invalid data because of:

• malformed or semantically invalid LLM output;
• unexpected external API responses;
• unchecked casts or JavaScript callers;
• handler regressions;
• data-dependent branches that violate the declared return type.

pgflow persists handler results in step_states.output, uses them as dependent-step inputs, evaluates conditional patterns against them, and aggregates map results. Invalid output should be caught before it enters those paths.

Standard Schema support keeps pgflow independent of one validation library while preserving each library's behavior. Zod schemas still run through Zod itself, including refinements, transforms, defaults, and async validation. The same API can also accept Valibot, ArkType, and other Standard Schema-compatible libraries.

Proposed API

Regular step

import { z } from 'zod'
import { Flow } from '@pgflow/dsl'

const classificationSchema = z.strictObject({
  route: z.enum(['respond', 'ignore']),
  reason: z.string().min(1).max(280),
})

const flow = new Flow<{ text: string }>({ slug: 'classifyTicket' })
  .step(
    {
      slug: 'classify',
      outputSchema: classificationSchema,
      maxAttempts: 3,
    },
    async ({ text }) => classifyText(text),
  )
  .step(
    {
      slug: 'notify',
      dependsOn: ['classify'],
      if: { classify: { route: 'respond' } },
    },
    async ({ classify }) => notifyTeam(classify),
  )

The worker validates classifyText()'s return value before calling complete_task(). notify receives the schema's inferred output type, and its condition runs against validated data.

Array step

const articlesSchema = z.array(z.strictObject({
  id: z.string().uuid(),
  url: z.string().url(),
}))

const flow = new Flow<{ feedUrl: string }>({ slug: 'fetchArticles' })
  .array(
    {
      slug: 'articles',
      outputSchema: articlesSchema,
    },
    async ({ feedUrl }) => fetchArticles(feedUrl),
  )

outputSchema validates the complete array returned by the handler. It may enforce array length, uniqueness refinements, or relationships between items.

Map step

const processedArticleSchema = z.strictObject({
  id: z.string().uuid(),
  summary: z.string().min(1),
})

const flow = new Flow<{ ids: string[] }>({ slug: 'processArticles' })
  .array({ slug: 'ids' }, ({ ids }) => ids)
  .map(
    {
      slug: 'processArticle',
      array: 'ids',
      itemOutputSchema: processedArticleSchema,
      maxAttempts: 3,
    },
    async (id) => processArticle(id),
  )

itemOutputSchema applies to each map task result, not to the aggregated array. A violation fails or retries only that item's task. Successful parsed items are aggregated in task_index order as today. The step output type is InferOutput<typeof processedArticleSchema>[].

The explicit itemOutputSchema name avoids implying that the schema receives the map step's aggregated array. .map() should reject outputSchema; .step() and .array() should reject itemOutputSchema.

Transforms and defaults

const normalizedUserSchema = z.strictObject({
  email: z.string().email().transform((value) => value.toLowerCase()),
  tags: z.array(z.string()).default([]),
})

const flow = new Flow<{ userId: string }>({ slug: 'normalizeUser' })
  .step(
    {
      slug: 'user',
      outputSchema: normalizedUserSchema,
    },
    async ({ userId }) => loadUser(userId),
  )

The parsed value—not the raw handler return—is passed to complete_task(). Downstream steps and SQL queries see normalized/defaulted output.

Non-Zod schema

import * as v from 'valibot'

const resultSchema = v.object({
  ok: v.boolean(),
  value: v.string(),
})

const flow = new Flow<{ id: string }>({ slug: 'portableSchema' })
  .step(
    {
      slug: 'load',
      outputSchema: resultSchema,
    },
    async ({ id }) => loadResult(id),
  )

No library adapter should be required. pgflow calls schema['~standard'].validate(value) and awaits either its synchronous or asynchronous result.

Type behavior

For a schema S extends StandardSchemaV1:

• handler return type should be assignable to StandardSchemaV1.InferInput<S>;
• the step's public output type should be StandardSchemaV1.InferOutput<S>;
• map step output should be StandardSchemaV1.InferOutput<S>[];
• current JSON-compatibility constraints should continue to apply to the value persisted after validation;
• without a schema, existing handler-return inference and behavior stay unchanged.

Using separate schema input/output types preserves transforms. For example, a schema can accept an unnormalized string and expose a normalized string to downstream steps.

Violation behavior

Default: retry

.step(
  {
    slug: 'extract',
    outputSchema: extractedDataSchema,
    maxAttempts: 3,
    // onOutputInvalid: 'retry' is the default
  },
  extractWithModel,
)

A schema violation becomes a normal task failure:

1. the handler returns;
2. validation fails;
3. the worker reports failure through the existing fail_task() path;
4. normal maxAttempts, retry delay, and whenExhausted behavior applies.

This default supports nondeterministic handlers such as LLM calls, where another attempt may return valid output.

Immediate hard failure

.step(
  {
    slug: 'normalize',
    outputSchema: normalizedSchema,
    onOutputInvalid: 'fail',
  },
  normalize,
)

'fail' means non-retryable failure: fail the step and run immediately, without applying whenExhausted. This matches the existing treatment of structural TYPE_VIOLATION failures.

Implementing this cleanly depends on a first-class permanent/non-retryable failure path rather than manually archiving PGMQ messages. This overlaps with the permanent/final failure use cases discussed in Discussion #240. If that mechanism is not ready, this feature can land in phases: support 'retry' first, then add 'fail' through the shared permanent-failure path.

Domain outcomes should remain normal return values, not permanent errors. For example, { status: 'resource_deleted' } is a valid, queryable completion; a schema contract violation is a workflow failure.

Error details and redaction

Introduce a typed worker error such as StepOutputValidationError.

Diagnostics should include only portable Standard Schema fields:

{
  stepSlug: 'extract',
  issues: [
    { path: ['items', 0, 'title'], message: 'Expected string' },
  ],
}

Do not include the invalid output or provider-specific received values in logs or persisted error messages. Outputs may contain secrets, personal data, or large payloads. Preserve the original issue objects on the in-memory error/cause if useful, but persist/log a redacted common representation.

Retry-safety caveat

Output validation happens after the handler returns. Any side effects performed by the handler may already have happened before a violation is found. If the task retries, those side effects may execute again.

Schemas do not make handlers exactly-once. Handlers must remain idempotent or retry-safe, especially for external side effects. Documentation should state this next to the default retry behavior.

Flow input validation: intentionally explicit for now

Do not add a hidden/ghost validation root step in the initial implementation.

A synthetic root step would:

• silently change the compiled flow shape;
• add one task and queue round-trip to every run;
• appear as unexpected monitoring noise;
• require rewiring all original root steps;
• change root handler signatures from direct flow input to dependency objects, or discard the parsed value and lose transforms/defaults.

Users can express input validation explicitly today:

const inputSchema = z.strictObject({
  email: z.string().email(),
})

const flow = new Flow<z.input<typeof inputSchema>>({ slug: 'sendEmail' })
  .step(
    { slug: 'validInput', maxAttempts: 1 },
    (input) => inputSchema.parse(input),
  )
  .step(
    { slug: 'send', dependsOn: ['validInput'] },
    ({ validInput }) => sendEmail(validInput.email),
  )

A future inputSchema option can validate at task claim time without changing flow shape, after the permanent-failure behavior is available. It should be a separate proposal.

Persistence and versioning

Schemas remain user-owned runtime objects:

• no JSON Schema conversion;
• no schema/hash columns;
• no compiled migration changes;
• no FlowShape or startup shape-comparison changes;
• no SQL Core validation;
• no automatic version enforcement when a schema changes.

Changing a schema for an existing flow slug is therefore an undetected behavior change. Documentation should tell users to version the flow slug when they make an incompatible input/output contract change.

Implementation plan

1. Standard Schema dependency and helper

Add @standard-schema/spec and implement one internal validation helper:

async function validateOutput<S extends StandardSchemaV1>(
  schema: S,
  value: unknown,
): Promise<StandardSchemaV1.InferOutput<S>> {
  const result = await schema['~standard'].validate(value)
  if (result.issues) throw new StepOutputValidationError(result.issues)
  return result.value
}

Standard Schema support should not convert or wrap the schema before validation. Calling the vendor's own validate preserves Zod refinements/transforms/defaults and equivalent features in other libraries.

2. DSL types (pkgs/dsl/src/dsl.ts)

• Extend StepDefinition with a runtime-only internal task-result schema (for example taskOutputSchema?: StandardSchemaV1).
• Expose it as outputSchema on .step() and .array().
• Expose it as itemOutputSchema on .map() and normalize both public names to the same internal field.
• Add onOutputInvalid?: 'retry' | 'fail' alongside that runtime-only schema metadata.
• Do not put schema objects in StepRuntimeOptions; they are not SQL/runtime database options.
• Keep schemas out of extractFlowShape() and compilation.
• Preserve schemas when fluent calls return a new immutable Flow instance.
• Infer handler input and step output from InferInput/InferOutput without changing existing no-schema overload behavior.

dsl.ts currently has multiple overloads for root/dependent/conditional/skippable variants. Avoid doubling every overload if possible; introduce reusable schema option/inference helpers and cover inference with type tests before choosing the final overload shape.

.array() currently delegates to .step(), so make sure schema metadata survives delegation. .map() builds its own StepDefinition, so normalize itemOutputSchema there.

3. Worker execution (pkgs/edge-worker/src/flow/StepTaskExecutor.ts)

After handler execution and before success logging/completeTask():

1. find the task-result schema on the step definition;
2. await Standard Schema validation;
3. pass the parsed value to completeTask();
4. on violation, route the typed error through normal failure handling or the permanent-failure path based on onOutputInvalid.

Validation time should count as task execution time. taskCompleted should log only after validation succeeds.

No extra database round-trip is needed. Validation runs once per task attempt inside the worker process.

4. Permanent-failure integration

Coordinate onOutputInvalid: 'fail' with the shared permanent-failure design related to Discussion #240.

The clean implementation likely needs the worker/client/Core failure API to distinguish retryable from non-retryable failures. Avoid schema-specific queue manipulation or direct pgmq.archive() calls.

5. Documentation

Update:

• concepts/understanding-flows — schemas and parsed outputs;
• concepts/map-steps — itemOutputSchema and per-task failure isolation;
• build/retrying-steps — output violations and idempotency caveat;
• build/validation-steps — explicit input validation remains the pattern;
• tutorials/use-cases/structured-output — show schema validation at the pgflow boundary.

Test plan

DSL type tests

• .step({ outputSchema }) exposes InferOutput<S> to dependents.
• Handler return accepts InferInput<S> for transformed schemas.
• .array({ outputSchema }) requires a schema whose output is an array.
• .map({ itemOutputSchema }) produces InferOutput<S>[].
• .map() rejects outputSchema.
• .step() and .array() reject itemOutputSchema.
• No-schema flows preserve all existing inference.
• Conditional/skippable step typing still works with schema-derived outputs.
• Zod and at least one other Standard Schema library pass the same inference tests.

DSL runtime tests

• Schema metadata is retained in StepDefinition across fluent Flow copies.
• Schema metadata is absent from extracted/compiled flow shape.
• .array() delegation retains outputSchema.
• .map() normalizes itemOutputSchema to the internal task-result schema.

Edge Worker tests

• Valid output calls completeTask() with the parsed value.
• Zod transforms/defaults appear in the stored output.
• Invalid output does not call completeTask().
• Default invalid output calls normal failTask() and follows retries.
• 'fail' uses the permanent-failure path once available.
• Map validation applies independently to each task result.
• Async Standard Schema validation is awaited.
• Handler exceptions are not reclassified as schema failures.
• Diagnostics omit the invalid output and normalize nested paths safely.

Compatibility tests

• Zod refinements, transforms, defaults, and async validation retain their normal behavior.
• Valibot (or another Standard Schema implementation) works without an adapter.
• Existing flows without schemas require no code or migration changes.

Acceptance criteria

• .step() and .array() accept optional outputSchema.
• .map() accepts optional itemOutputSchema with per-item semantics.
• Standard Schema-compatible libraries work without adapters.
• Schema output types drive downstream TypeScript inference.
• The parsed value is stored and passed downstream.
• Default violations use existing retry/whenExhausted behavior.
• onOutputInvalid: 'fail' uses a first-class permanent-failure path, not direct queue manipulation.
• Validation errors do not expose invalid output values.
• Schemas do not change compiled SQL, database schema, or flow shape.
• Existing no-schema flows behave exactly as before.
• Documentation explains retry-safety and user-managed flow versioning.