feat(tasks): validate task payloads and return values at runtime

AGENTS.md

@@ -140,6 +140,7 @@ Tables live in the `pgconductor` schema with a `_private_` prefix (`_private_exe
 - Claiming returns each execution's undelivered signals, which prime its `SignalInbox` before the handler runs
 - The claim statement also reads the cancellation flag and newer undelivered signals of the worker's running executions (`running` in `getExecutions`), aborts cancelled ones and hands signals to their `SignalInbox`. While all slots are busy the worker still runs it every `pollIntervalMs`, with a batch size of 0, as long as an execution runs. Batch members are polled too and share the batch's controller; cancelling one releases the others without spending their attempt. After a drain's last empty claim the worker keeps these zero-sized polls until its pipeline ends, then joins the fetch loop before resetting
 - Handles graceful shutdown via AbortController; a stop during registration retries ends them and settles `stopped`
+- Validates the payload of every invocation (any origin, including SQL and dead letters) and the return value against the task's Standard Schemas (`Task.schemas`, resolved from `TaskSchemas` in `createTask`) and passes their output on. Issues fail the execution as `permanently_failed`; a throwing validator fails the attempt. Batches validate per member
 
 **TaskContext** (`task-context.ts`)
 - Provides API to task functions: `step()`, `sleep()`, `invoke()`, `invokeAndWait()`, `waitForResult()`, `waitForEvent()`, `subscribe()`, `waitForAny()`, `onSignal()`, `sendSignal()`, `signalWithStart()`, plus `executionId` and `queue`

docs/content/api/conductor.md

@@ -238,7 +238,7 @@ const executionId = await conductor.signalWithStart({
   queue: "default", // optional
   key: "thread-1",
   signal: { name: "input", id: requestId, payload },
-  input: {}, // used only when it starts an execution
+  input: {}, // used only when it starts an execution (default: {})
   priority: 0, // optional
   metadata: {}, // optional
 });

docs/content/crafting-tasks/defining-tasks.md

@@ -8,7 +8,7 @@ Schema definitions give you:
 
 - **Type safety** - TypeScript infers payload and return types automatically
 - **Shared schemas** - One source of truth for worker and client
-- **Runtime validation** - Optional validation with StandardSchema (Zod, Valibot, etc.)
+- **Runtime validation** - Workers validate payloads and return values with StandardSchema (Zod, Valibot, etc.)
 - **Better DX** - Autocomplete for payloads and return values
 
 ## Task Definitions
@@ -78,6 +78,20 @@ const conductor = Conductor.create({
 });
 ```
 
+## Runtime Validation
+
+Workers validate every execution of a task defined with `defineTask()` against its schemas, whichever way it was invoked: `conductor.invoke()`, `ctx.invoke()`, `signalWithStart()`, `pgconductor.invoke()` from SQL, or a dead-letter delivery.
+
+- The payload is validated before middleware and the handler run. The handler receives the schema's output, so transforms, defaults and stripped keys apply.
+- The return value is validated before the execution completes. The schema's output is stored as the result, serialized as JSON.
+- A payload or return value that does not match fails the execution without retries, with the issues as its error, for example `Invalid payload for task "send-email": to: Invalid email address`. Like any permanent failure it fails waiting parents and is delivered to a dead-letter queue.
+- A validator that throws instead of returning issues fails the attempt, which is retried.
+- Async validators are awaited.
+- Batch tasks validate each execution on its own: members with an invalid payload fail without reaching the handler, and the others run as the batch. Return values are validated per item.
+- Invoking does not validate, so the stored payload is the one you passed. Cron and custom event triggers carry no task payload.
+
+Type-only definitions (`DefineTask` with `TaskSchemas.fromUnion()`) have no runtime schema and are never validated.
+
 ## Event Definitions
 
 Events enable reactive, event-driven patterns.
@@ -140,7 +154,7 @@ const task = conductor.createTask(
   { name: "process-data" },
   { invocable: true },
   async (event, ctx) => {
-    // Return value must match schema
+    // The return value is validated against the schema
     return {
       processed: event.payload.items.length,
       errors: [],

docs/content/task-execution/batching.md

@@ -73,7 +73,7 @@ const processBatch = conductor.createTask(
 - `size` - Maximum number of events to batch together
 - `timeoutMs` - Maximum time to wait for a full batch before processing what's available
 
-Each batch event has `{ name, payload?, execution }`. The read-only `execution` object contains `id`, `queue`, `task_key`, and `locked_by`; the event payload remains unchanged. Batch handlers can use `event.execution.id` for their own idempotency keys.
+Each batch event has `{ name, payload?, execution }`. The read-only `execution` object contains `id`, `queue`, `task_key`, and `locked_by`. With a task schema, payloads and return values are validated per execution: one with an invalid payload fails on its own and the batch runs without it (see [Runtime Validation](../crafting-tasks/defining-tasks.md#runtime-validation)). Batch handlers can use `event.execution.id` for their own idempotency keys.
 
 ### Batch Tasks with Returns
 

docs/content/task-execution/dead-letter-queue.md

@@ -21,4 +21,4 @@ const chargeCard = conductor.createTask(
 
 The destination is a new execution with the original payload. Its execution row contains machine-readable source execution ID, source queue and task, final error, attempt count, and failure timestamp. The source remains a normal failed execution unless its retention policy removes it.
 
-Retries and cancellation do not deliver to a dead-letter queue. Delivery is transactional, claim-fenced, and idempotent. A destination may have its own retry, retention, and concurrency settings. Chains are supported, but a task cannot target itself directly. The destination task must accept the source payload.
+Retries and cancellation do not deliver to a dead-letter queue. Delivery is transactional, claim-fenced, and idempotent. A destination may have its own retry, retention, and concurrency settings. Chains are supported, but a task cannot target itself directly. The destination task must accept the source payload. A destination defined with a payload schema validates it like any execution, so a source that failed on an invalid payload also fails there; give such a destination a schema that accepts it, or a type-only definition.

docs/content/task-execution/signals.md

@@ -53,7 +53,7 @@ const executionId = await conductor.signalWithStart({
   task: "session",
   key: "slack:T1:C1:1712345678.000100",
   signal: { name: "input", id: requestId, payload },
-  input: {}, // payload of the execution, used only when it starts one
+  input: {}, // payload of the execution, used only when it starts one (default: {})
 });
 ```
 

packages/pgconductor-js/src/conductor.ts

@@ -277,6 +277,11 @@ export class Conductor<
 		TaskContext<Tasks, Events, Metadata> & ExtraContext & MiddlewareContext,
 		TaskEventFromTriggers<TTriggers, ResolvedPayload<Tasks, TDef>, Events>
 	> {
+		const queue = definition.queue || "default";
+		const schemas = this.options.tasks?.definitions.find(
+			(task: TaskDefinition<string, unknown, unknown, string>) =>
+				task.name === definition.name && task.queue === queue,
+		);
 		return Task.create<
 			TDef["name"],
 			ResolvedQueue<TDef>,
@@ -298,6 +303,7 @@ export class Conductor<
 			>,
 			this.events,
 			this.options.metadata,
+			schemas,
 		);
 	}
 
@@ -483,7 +489,7 @@ export class Conductor<
 					queue,
 					key: options.key,
 					signal: options.signal,
-					input: (options.input || null) as Payload | null,
+					input: (options.input || {}) as Payload,
 					priority: options.priority,
 					traceContext: this.telemetry.traceContext(),
 					metadata,

packages/pgconductor-js/src/lib/validate-schema.ts

@@ -1,20 +1,28 @@
 import type { StandardSchemaV1 } from "@standard-schema/spec";
 
+export async function parseSchema<T>(
+	schema: StandardSchemaV1<unknown, T> | undefined,
+	value: T,
+	label: string,
+): Promise<{ value: T } | { error: string }> {
+	if (!schema) return { value };
+	const result = await schema["~standard"].validate(value);
+	if (!result.issues) return { value: result.value };
+	const issues = result.issues.map((issue) => {
+		const path = issue.path
+			?.map((segment) => String(typeof segment === "object" ? segment.key : segment))
+			.join(".");
+		return path ? `${path}: ${issue.message}` : issue.message;
+	});
+	return { error: `Invalid ${label}: ${issues.join("; ")}` };
+}
+
 export async function validateSchema<T>(
 	schema: StandardSchemaV1<unknown, T> | undefined,
 	value: T,
 	label: string,
 ): Promise<T> {
-	if (!schema) return value;
-	const result = await schema["~standard"].validate(value);
-	if (result.issues) {
-		const issues = result.issues.map((issue) => {
-			const path = issue.path
-				?.map((segment) => String(typeof segment === "object" ? segment.key : segment))
-				.join(".");
-			return path ? `${path}: ${issue.message}` : issue.message;
-		});
-		throw new Error(`Invalid ${label}: ${issues.join("; ")}`);
-	}
-	return result.value;
+	const parsed = await parseSchema(schema, value, label);
+	if ("error" in parsed) throw new Error(parsed.error);
+	return parsed.value;
 }

packages/pgconductor-js/src/task-context.ts

@@ -745,7 +745,7 @@ export class TaskContext<
 					queue: options.queue || "default",
 					key: options.key,
 					signal: options.signal,
-					input: (options.input || null) as Payload | null,
+					input: (options.input || {}) as Payload,
 					priority: options.priority,
 					traceContext: this.opts.telemetry.traceContext(),
 					metadata: this.opts.execution.metadata,

packages/pgconductor-js/src/task.ts

@@ -77,6 +77,12 @@ export type ValidateDeadLetterConfiguration<T, TPayload extends object> = T exte
 
 export type RetentionSettings = boolean | { days: number };
 
+/** Standard Schemas the worker validates the payload and return value of every execution with. */
+export type TaskRuntimeSchemas = {
+	readonly payload?: StandardSchemaV1;
+	readonly returns?: StandardSchemaV1;
+};
+
 export type TaskEvent<P extends object = object> =
 	| { name: "pgconductor.cron" }
 	| { name: "pgconductor.invoke"; payload: P };
@@ -174,6 +180,7 @@ export class Task<
 		public readonly execute: ExecuteFunction<EventType, Returns, Context>,
 		public readonly eventDefinitions: readonly EventDefinition<string, any, any>[] = [],
 		public readonly metadataSchema?: StandardSchemaV1<unknown, object>,
+		public readonly schemas: TaskRuntimeSchemas = {},
 	) {
 		const { name, queue, ...config } = definition;
 		this.name = name;
@@ -212,13 +219,15 @@ export class Task<
 		execute: ExecuteFunction<EventType, Returns, Context>,
 		eventDefinitions: readonly EventDefinition<string, any, any>[] = [],
 		metadataSchema?: StandardSchemaV1<unknown, object>,
+		schemas: TaskRuntimeSchemas = {},
 	): Task<Key, Queue, Payload, Returns, Context, EventType> {
 		return new Task<Key, Queue, Payload, Returns, Context, EventType>(
 			definition,
 			triggers,
 			execute,
 			eventDefinitions,
 			metadataSchema,
+			schemas,
 		);
 	}
 }

packages/pgconductor-js/src/worker.ts

@@ -4,7 +4,9 @@ import type {
 	Execution,
 	ExecutionResult,
 	ExecutionSpec,
+	Payload,
 } from "./database-client.js";
+import type { StandardSchemaV1 } from "@standard-schema/spec";
 import type { AnyTask, BatchConfig } from "./task.js";
 import type { TaskDefinition } from "./task-definition.js";
 import { waitFor } from "./lib/wait-for.js";
@@ -25,6 +27,7 @@ import { Telemetry } from "./telemetry.js";
 import { noop } from "./lib/noop.js";
 import type { Middleware, MiddlewareResult } from "./middleware.js";
 import { SignalInbox } from "./signal-inbox.js";
+import { parseSchema } from "./lib/validate-schema.js";
 
 /**
  * The configuration options for the Worker.
@@ -69,6 +72,30 @@ function taskEventFor(execution: Execution): { name: string; payload?: unknown }
 	return { name: "pgconductor.invoke", payload: execution.payload };
 }
 
+// A value the schema rejects fails the execution without retries; a validator that throws fails the attempt
+async function parseOrFail(
+	schema: StandardSchemaV1 | undefined,
+	value: unknown,
+	label: string,
+): Promise<{ value: unknown } | { status: "failed" | "permanently_failed"; error: string }> {
+	try {
+		const parsed = await parseSchema(schema, value, label);
+		return "error" in parsed ? { status: "permanently_failed", error: parsed.error } : parsed;
+	} catch (err) {
+		return { status: "failed", error: coerceError(err).message };
+	}
+}
+
+// Invocations carry the payload of the task's schema, events and cron occurrences do not
+async function parsePayload(task: AnyTask, event: { name: string; payload?: unknown }) {
+	if (event.name !== "pgconductor.invoke") return { value: event.payload };
+	return parseOrFail(task.schemas.payload, event.payload, `payload for task "${task.name}"`);
+}
+
+function parseResult(task: AnyTask, result: unknown) {
+	return parseOrFail(task.schemas.returns, result, `return value of task "${task.name}"`);
+}
+
 /**
  * Worker implemented as async pipeline: fetch → execute → flush.
  * Uses async iterators for clean composition and natural backpressure.
@@ -603,6 +630,10 @@ export class Worker<
 		);
 
 		try {
+			const payload = await parsePayload(task, taskEvent);
+			if ("error" in payload) return { ...identity, ...payload };
+			taskEvent.payload = payload.value;
+
 			await this.scheduleNextExecution(exec);
 
 			const outcome = await this.telemetry.process({
@@ -681,10 +712,13 @@ export class Worker<
 				} as const;
 			}
 
+			const result = await parseResult(task, outcome.output);
+			if ("error" in result) return { ...identity, ...result };
+
 			return {
 				...identity,
 				status: "completed",
-				result: outcome.output,
+				result: result.value as Payload | undefined,
 				signal_names: signals.names,
 			} as const;
 		} catch (err) {
@@ -718,20 +752,52 @@ export class Worker<
 	private async executeBatchTask(
 		task: AnyTask,
 		taskKey: string,
-		executions: Execution[],
+		claimed: Execution[],
 	): Promise<ExecutionResult[]> {
-		const events = executions.map((exec) => {
-			return {
-				...taskEventFor(exec),
+		const parsed = await Promise.all(
+			claimed.map(async (exec) => {
+				const event = taskEventFor(exec);
+				return { exec, event, payload: await parsePayload(task, event) };
+			}),
+		);
+
+		// Members with an invalid payload settle on their own, the others run as the batch
+		const rejected: ExecutionResult[] = [];
+		const executions: Execution[] = [];
+		const events: object[] = [];
+		for (const { exec, event, payload } of parsed) {
+			if ("error" in payload) {
+				rejected.push({
+					execution_id: exec.id,
+					claim_token: exec.claim_token,
+					queue: exec.queue,
+					task_key: taskKey,
+					...payload,
+				});
+				continue;
+			}
+			executions.push(exec);
+			events.push({
+				...event,
+				payload: payload.value,
 				execution: {
 					id: exec.id,
 					queue: exec.queue,
 					task_key: exec.task_key,
 					locked_by: exec.locked_by,
 				},
-			};
-		});
+			});
+		}
+		if (executions.length === 0) return rejected;
+		return [...rejected, ...(await this.executeBatch(task, taskKey, executions, events))];
+	}
 
+	private async executeBatch(
+		task: AnyTask,
+		taskKey: string,
+		executions: Execution[],
+		events: object[],
+	): Promise<ExecutionResult[]> {
 		const taskAbortController = new TypedAbortController<TaskAbortReasons>();
 		const timeout = task.timeoutMs
 			? setTimeout(() => taskAbortController.abort({ reason: "timed-out" }), task.timeoutMs)
@@ -828,14 +894,23 @@ export class Worker<
 				}));
 			}
 
-			return executions.map((exec, i) => ({
-				execution_id: exec.id,
-				claim_token: exec.claim_token,
-				queue: exec.queue,
-				task_key: taskKey,
-				status: "completed" as const,
-				result: outcome.output?.[i],
-			}));
+			return Promise.all(
+				executions.map(async (exec, i) => {
+					const identity = {
+						execution_id: exec.id,
+						claim_token: exec.claim_token,
+						queue: exec.queue,
+						task_key: taskKey,
+					};
+					const result = await parseResult(task, outcome.output?.[i]);
+					if ("error" in result) return { ...identity, ...result };
+					return {
+						...identity,
+						status: "completed" as const,
+						result: result.value as Payload | undefined,
+					};
+				}),
+			);
 		} catch (err) {
 			if (this.signal.aborted) {
 				return executions.map((exec) => ({