import {
	createWorkflowConnections,
	normalizeConnectionRequirements,
	validateTriggerConnections,
	type WorkflowConnectionDeclarations,
	type WorkflowConnectionRequirement,
	type WorkflowConnections,
	type WorkflowTriggerConnections,
} from "./connections.js";
import {
	hydrateWorkflowAccess,
	normalizeWorkflowAccess,
	type WorkflowAccess,
	type WorkflowAccessDeclarations,
	type WorkflowAccessRequirements,
} from "./workflow-access.js";
import { AsyncLocalStorage } from "node:async_hooks";
import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";
import { join } from "node:path";

import type { CapabilityContext } from "./context.js";
import { createCapabilityContext } from "./context.js";
import { ExecutionError, FatalError, RateLimitError, RetryableError } from "./error.js";
import { writeOutput } from "./output.js";
import { resolveRuntimeInput } from "./runtime-input.js";
import { readRunMetadata, type RunMetadata } from "./runtime-metadata.js";
import { createWorkflowTriggerContext, validateManualWorkflowInput } from "./events.generated.js";
import type { Infer, SchemaBuilder } from "./schema-builder.js";
import { compileWorkflowOutputSchema } from "./workflow-schema.js";
import { createWorkflowStepState, type WorkflowState } from "./workflow-state.js";
import type {
	ManualWorkflowTrigger,
	WebhookTrigger,
	WorkflowEventMap,
	WorkflowTrigger,
	WorkflowEventCreators,
	WorkflowTriggerForConnections,
} from "./events.generated.js";
import { createWorkflowSyncClient, type WorkflowSyncClient } from "./workflow-sync.js";

type HandlerOptions = {
	/** Return the concrete result instead of writing the output envelope. */
	concreteOutput?: true;
};

/** The only trigger whose deliveries a verify handler can answer. */
const WEBHOOK_TRIGGER_TYPE: WebhookTrigger["type"] = "webhooks.webhook";

type WorkflowWaitDurationValues = {
	milliseconds?: number;
	seconds?: number;
	minutes?: number;
	hours?: number;
	days?: number;
	weeks?: number;
};

export type WorkflowWaitDuration = {
	[Unit in keyof WorkflowWaitDurationValues]-?: Required<Pick<WorkflowWaitDurationValues, Unit>> &
		WorkflowWaitDurationValues;
}[keyof WorkflowWaitDurationValues];

type WorkflowWaitKey = string | string[];

type WorkflowInputFieldOptions = {
	optional?: boolean;
};

declare const workflowInputValueType: unique symbol;

export type WorkflowInputField<T> = {
	type: "text" | "number" | "checkbox" | "select" | "multi_select" | "date" | "person" | "files";
	optional?: boolean;
	options?: readonly string[];
	readonly [workflowInputValueType]?: T;
};

type WorkflowInputValue<
	T,
	TOptions extends WorkflowInputFieldOptions | undefined,
> = TOptions extends { optional: true } ? T | undefined : T;

/** Date-only or local date-time endpoints, with the selected IANA time zone. */
export type WorkflowInputDate = {
	start: string;
	end?: string;
	timeZone?: string;
};
export const input = {
	text<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<string, TOptions>> {
		return { type: "text", ...options };
	},
	number<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<number, TOptions>> {
		return { type: "number", ...options };
	},
	checkbox<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<boolean, TOptions>> {
		return { type: "checkbox", ...options };
	},
	select<
		const TOptions extends readonly [string, ...string[]],
		const TFieldOptions extends WorkflowInputFieldOptions | undefined,
	>(
		values: TOptions,
		options?: TFieldOptions,
	): WorkflowInputField<WorkflowInputValue<TOptions[number], TFieldOptions>> {
		return { type: "select", options: values, ...options };
	},
	date<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<WorkflowInputDate, TOptions>> {
		return { type: "date", ...options };
	},
	multiSelect<
		const TOptions extends readonly [string, ...string[]],
		const TFieldOptions extends WorkflowInputFieldOptions | undefined,
	>(
		values: TOptions,
		options?: TFieldOptions,
	): WorkflowInputField<WorkflowInputValue<TOptions[number][], TFieldOptions>> {
		return { type: "multi_select", options: values, ...options };
	},
	/** Returns the user ID of every selected person. */
	person<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<string[], TOptions>> {
		return { type: "person", ...options };
	},
	/** Returns each attachment's URL or Notion attachment block ID. */
	files<const TOptions extends WorkflowInputFieldOptions | undefined>(
		options?: TOptions,
	): WorkflowInputField<WorkflowInputValue<string[], TOptions>> {
		return { type: "files", ...options };
	},
};

export type WorkflowInputSchema = Record<string, WorkflowInputField<unknown>>;

export type WorkflowInputValues<TSchema extends WorkflowInputSchema> = {
	[Key in keyof TSchema]: TSchema[Key] extends WorkflowInputField<infer Value> ? Value : never;
};

export type WorkflowInputResponse<TSchema extends WorkflowInputSchema> = {
	value: WorkflowInputValues<TSchema>;
	respondedBy: string;
	respondedAt: string;
};

export type WorkflowWaitForInputOptions<TSchema extends WorkflowInputSchema> = {
	input: TSchema;
	key?: WorkflowWaitKey;
};

type WorkflowEventWaitNotionPropertyFilters = Record<string, unknown>;

type WorkflowEventWaitUserFilter = {
	/** Only events caused by these Notion users resolve the wait. */
	onlyTriggerForUsers?: { notionIds: string[] };
};

type WorkflowEventWaitMailFilter = {
	field: "body" | "domain" | "from" | "subject" | "to";
	operator:
		| "is_empty"
		| "is_not_empty"
		| "string_contains"
		| "string_does_not_contain"
		| "string_ends_with"
		| "string_is"
		| "string_is_not"
		| "string_starts_with";
	value: { type: "exact"; value?: string };
};

type WorkflowEventWaitCalendarScope = {
	calendars: Array<{ accountId: string; calendarId: string }>;
	filter?: Record<string, unknown>;
};

/**
 * Where each waitable event comes from. A wait has its own scope: a workflow
 * cannot both trigger on and wait for the same event.
 */
export type WorkflowEventWaitScopeMap = {
	"notion.page.created": {
		dataSourceId: string;
		viewId?: string;
		propertyFilters?: WorkflowEventWaitNotionPropertyFilters;
	};
	"notion.page.updated": {
		dataSourceId: string;
		/** Property IDs or names; changes to other properties are ignored. */
		properties?: string[];
		propertyFilters?: WorkflowEventWaitNotionPropertyFilters;
		shouldIgnorePageContentUpdates?: boolean;
	};
	"notion.page.deleted": {
		dataSourceId: string;
		viewId?: string;
		propertyFilters?: WorkflowEventWaitNotionPropertyFilters;
	};
	"notion.meetingNote.summary.completed": {
		dataSourceId: string;
		viewId?: string;
		propertyFilters?: WorkflowEventWaitNotionPropertyFilters;
	};
	"notion.page.discussion.comment.added": { dataSourceId: string };
	"notion.agent.mentioned": WorkflowEventWaitUserFilter;
	"slack.message": WorkflowEventWaitUserFilter & {
		channelIds: string[];
		filters?: {
			operator: "all" | "any";
			filters: Array<{
				type: "message.text.contains";
				operator: "all" | "any";
				values: string[];
			}>;
		};
		teamId?: string;
	};
	"slack.reaction.added": WorkflowEventWaitUserFilter & {
		channelIds?: string[];
		allChannels?: boolean;
		/** Emoji names without colons. */
		reactions: string[];
		teamId?: string;
	};
	"slack.app.mention": WorkflowEventWaitUserFilter & {
		channelIds: string[];
		teamId?: string;
	};
	"discord.interaction": { channelIds: string[] };
	"mail.email.received": {
		emailAccountId: string;
		filters?: WorkflowEventWaitMailFilter[];
		inboxOnly?: boolean;
	};
	"mail.email.sent": {
		emailAccountId: string;
		filters?: WorkflowEventWaitMailFilter[];
	};
	"mail.label.applied": { emailAccountId: string; labelNames: string[] };
	"calendar.event.created": WorkflowEventWaitCalendarScope;
	"calendar.event.updated": WorkflowEventWaitCalendarScope;
	"calendar.event.canceled": WorkflowEventWaitCalendarScope;
	"googleDriveOauth.filesChanged": {
		/** Direct children of a folder, or a shared drive. Polled hourly. */
		scope: { type: "folder"; folderId: string } | { type: "drive"; driveId: string };
	};
};

/** Triggers a workflow can wait on. */
export type WorkflowWaitEventTrigger = Extract<
	WorkflowTrigger,
	{ type: keyof WorkflowEventWaitScopeMap }
>;

type WorkflowEventWaitScope<TTrigger extends WorkflowWaitEventTrigger> =
	WorkflowEventWaitScopeMap[TTrigger["type"]];

type WorkflowWaitEventTriggerForConnections<TConnections extends WorkflowConnectionDeclarations> =
	Extract<WorkflowTriggerForConnections<TConnections>, WorkflowWaitEventTrigger>;

export type WorkflowWaitForEventOptions<TTrigger extends WorkflowWaitEventTrigger> = {
	/** The event to wait for, created with the same creators as workflow triggers. */
	event: TTrigger;
	/** How long to wait before resolving with `null`. Must be shorter than 7 days. */
	timeout: WorkflowWaitDuration;
	key?: WorkflowWaitKey;
} & ({} extends WorkflowEventWaitScope<TTrigger>
	? { scope?: WorkflowEventWaitScope<TTrigger> }
	: { scope: WorkflowEventWaitScope<TTrigger> });

export type WorkflowWaitUntilOptions =
	| {
			at: Date;
			key?: WorkflowWaitKey;
	  }
	| {
			after: WorkflowWaitDuration;
			key?: WorkflowWaitKey;
	  };

export type WorkflowWaitUntilResult = {
	status: "waiting";
	wait: {
		type: "until";
		name: string;
		resumeAtMs: number;
	};
};

export type WorkflowWaitForInputResult = {
	status: "waiting";
	wait: {
		type: "input";
		name: string;
		stepKey: string;
		input: WorkflowInputSchema;
	};
};

export type WorkflowWaitForEventResult = {
	status: "waiting";
	wait: {
		type: "event";
		name: string;
		stepKey: string;
		event: WorkflowWaitEventTrigger & { scope?: Record<string, unknown> };
		timeoutAtMs: number;
	};
};

type WorkflowWaitResult =
	| WorkflowWaitUntilResult
	| WorkflowWaitForInputResult
	| WorkflowWaitForEventResult;

type WorkflowHandlerResult =
	| { status: "success"; output?: Record<string, unknown> }
	| WorkflowWaitResult;

/** An object schema describing the value a workflow returns. */
export type WorkflowOutputSchema = SchemaBuilder<Record<string, unknown>, "object">;

/**
 * What a workflow handler returns: the value described by `outputSchema`, or
 * nothing when the workflow declares no output schema.
 */
export type WorkflowHandlerReturn<TOutputSchema extends WorkflowOutputSchema | undefined> =
	TOutputSchema extends WorkflowOutputSchema
		? Promise<Infer<TOutputSchema>> | Infer<TOutputSchema>
		: Promise<void> | void;

const MAX_WORKFLOW_WAIT_MS = 7 * 24 * 60 * 60 * 1_000;
export { FatalError, RetryableError } from "./error.js";

export type {
	WorkflowAccess,
	WorkflowAccessAgentLevel,
	WorkflowAccessBindings,
	WorkflowAccessDeclaration,
	WorkflowAccessDeclarations,
	WorkflowAccessLevel,
	WorkflowAccessLevelFor,
	WorkflowAccessResource,
	ResourceAccessLevel,
	WorkflowAccessResourceType,
	WorkflowAccessRequirement,
	WorkflowAccessRequirements,
	WorkflowAccessValue,
} from "./workflow-access.js";
export { access } from "./workflow-access.js";
export type WorkflowEvent = WorkflowEventMap[keyof WorkflowEventMap];

export type WorkflowEventForTrigger<T extends WorkflowTrigger> =
	T extends ManualWorkflowTrigger<infer TInput>
		? {
				type: "workflow.manual";
				input: TInput;
			}
		: T extends { type: keyof WorkflowEventMap }
			? WorkflowEventMap[T["type"]]
			: never;

export type WorkflowEventForTriggers<T extends readonly WorkflowTrigger[]> =
	WorkflowEventForTrigger<T[number]>;

export {
	connections,
	type WorkflowConnection,
	type WorkflowConnectionDeclarations,
	type WorkflowConnectionRequirement,
	type WorkflowConnections,
	type WorkflowTriggerConnections,
	type OAuthConfiguration,
	type OAuthConnection,
} from "./connections.js";

export type WorkflowDeadline = {
	/** Maximum wall-clock lifetime of one logical workflow run. */
	afterMs: number;
};

/**
 * An error class, or a predicate, that marks an error as retryable.
 * Classes match with `instanceof`; any other function is called with the error.
 */
export type WorkflowRetryableErrorMatcher =
	| (abstract new (...args: never[]) => Error)
	| ((error: unknown) => boolean);

export type WorkflowRetryPolicy = {
	/** Total attempts, including the initial attempt. At most 4. */
	maxAttempts: number;
	/** Delay before the first retry. Defaults to 1 second. */
	initialDelayMs?: number;
	/** Maximum delay between attempts. Defaults to 5 minutes. */
	maxDelayMs?: number;
	/** Exponential backoff multiplier. Defaults to 2. */
	backoffMultiplier?: number;
	/**
	 * Errors that may be retried, as one matcher or an array of matchers. When
	 * set, only matching errors and `RetryableError` are retried; every other
	 * error fails immediately. When omitted, every error except `FatalError` is
	 * retried.
	 */
	retryOn?: WorkflowRetryableErrorMatcher | readonly WorkflowRetryableErrorMatcher[];
};

/** Platform limit on `maxAttempts`. */
const MAX_WORKFLOW_RETRY_ATTEMPTS = 3;

/** Platform limit on deadlines and retry delays: seven days. */
const MAX_WORKFLOW_DURATION_MS = 7 * 24 * 60 * 60 * 1000;

/**
 * One workflow invocation runs in a sandbox that stops after five minutes, and
 * step retries wait inside that invocation. A step's worst-case retry delays
 * and timeouts must fit in this budget.
 */
const WORKFLOW_STEP_RETRY_BUDGET_MS = 5 * 60 * 1000;

/** Retry policy used when neither the workflow nor the step sets one. */
const DEFAULT_WORKFLOW_RETRY_POLICY = {
	maxAttempts: 3,
	initialDelayMs: 1_000,
	maxDelayMs: 5 * 60 * 1000,
	backoffMultiplier: 2,
} as const satisfies WorkflowRetryPolicy;

/**
 * The inbound HTTP request passed to a webhook verify handler.
 */
export type WebhookVerifyRequest = {
	/** Uppercase HTTP method. Deliveries are POSTs; GET/HEAD cover challenge probes. */
	method: "GET" | "HEAD" | "POST";
	/** The full webhook URL as received. */
	url: string;
	/** Query string parameters. */
	query: Record<string, string>;
	/** Request headers with lowercased names. */
	headers: Record<string, string>;
	/** Raw request body. Empty string for bodyless methods like GET and HEAD. */
	rawBody: string;
};

/**
 * The HTTP response a verify handler produces. `deliver`, when present,
 * explicitly controls whether the request is enqueued for the workflow
 * handler. Otherwise delivery is derived from the status: 2xx enqueues and 4xx
 * does not.
 */
export type WebhookVerifyResponse = {
	/** Response status code. Only 2xx and 4xx statuses are allowed. */
	status: number;
	/** Optional response body, at most 8KB. */
	body?: string;
	/** Response content type. Defaults to "text/plain". */
	contentType?: "application/json" | "text/plain";
	/**
	 * Whether to enqueue the request for the workflow handler. Defaults to true
	 * for 2xx responses and false for 4xx responses.
	 */
	deliver?: boolean;
};

/**
 * A webhook verify handler. Runs synchronously at Notion's webhook ingress,
 * before the HTTP response is sent, in the app's own sandbox — the same
 * runtime as the workflow handler, with the full Node.js standard library.
 * Read signing secrets from `process.env` and use `node:crypto` for
 * signature checks.
 *
 * Keep it fast: the run must fit a small wall-clock budget (a few seconds,
 * including sandbox startup), so avoid network calls — including
 * `context.notion` API calls, which typically won't fit the budget.
 */
export type WebhookVerifyHandler = (
	request: WebhookVerifyRequest,
	context: CapabilityContext,
) => WebhookVerifyResponse | Promise<WebhookVerifyResponse>;

/**
 * A verify invocation as dispatched by the platform: a single `request`
 * object rather than the event a workflow invocation carries.
 */
export type WebhookVerifyInvocation = {
	request: WebhookVerifyRequest;
};

/**
 * A trigger as the manifest records it. A declared `verify` handler is flagged
 * on the webhook trigger itself, which is where the platform reads it to decide
 * whether to run `verify` at webhook ingress.
 */
export type WorkflowManifestTrigger = WorkflowTrigger | (WebhookTrigger & { hasVerify: true });

/**
 * Configuration passed to {@link workflow}.
 */
export type WorkflowConfiguration<
	TTriggers extends readonly [WorkflowTrigger, ...WorkflowTrigger[]],
	TConnections extends WorkflowConnectionDeclarations = WorkflowConnectionDeclarations,
	TAccess extends WorkflowAccessDeclarations = WorkflowAccessDeclarations,
	TOutputSchema extends WorkflowOutputSchema | undefined = undefined,
> = {
	/**
	 * A human-readable name for the workflow, shown in the UI when viewing workflows
	 */
	name: string;

	/**
	 * A human-readable description of what the workflow does, shown in the UI when viewing workflows
	 */
	description: string;

	/**
	 * An array of triggers that can invoke this workflow.
	 *
	 * Each trigger defines a specific event or condition that causes the workflow to run.
	 */
	triggers: TTriggers;

	/**
	 * Notion-as-Code resources whose access is configured during cloud deployment.
	 * Deployment reconciles grants through the existing Notion-module permissions
	 * and supplies named runtime bindings on `context.access`.
	 */
	access?: TAccess;

	/**
	 * Connections that must be set up before this workflow can run.
	 */
	connections?: TConnections;

	/** Maximum wall-clock lifetime of one logical workflow run. */
	deadline?: WorkflowDeadline;

	/**
	 * Default retry policy for every step, and the retry policy for failures
	 * that happen outside a step. A step's own `retry` replaces it.
	 */
	retry?: WorkflowRetryPolicy;

	/**
	 * Optional synchronous verification handler, run at Notion's webhook
	 * ingress before the HTTP response is sent. Use it for providers that
	 * require a synchronous verification response: signature checks with
	 * specific status codes, challenge echoes, or handshake requests.
	 *
	 * Only workflows with an `events.webhook()` trigger may declare one.
	 *
	 * The handler controls the HTTP response. Return `deliver: true` to enqueue
	 * the request for the workflow handler, or `deliver: false` to answer the
	 * provider without enqueueing it. When omitted, 2xx enqueues and 4xx does
	 * not. `verify` runs in the app's own sandbox with full Node.js — read
	 * secrets from `process.env` and use `node:crypto` for signature checks. It
	 * receives the capability context as its second argument, without the run
	 * metadata and `step` a workflow run carries.
	 *
	 * Without `verify`, the webhook keeps the current behavior: every request
	 * is accepted with `202` and processed asynchronously.
	 *
	 * @example
	 * ```ts
	 * verify: (request) => {
	 *   const signature = createHmac("sha256", process.env.WEBHOOK_SECRET ?? "")
	 *     .update(request.rawBody)
	 *     .digest("hex");
	 *   const expected = Buffer.from(`sha256=${signature}`);
	 *   const actual = Buffer.from(request.headers["x-hub-signature-256"] ?? "");
	 *   const verified = expected.length === actual.length && timingSafeEqual(expected, actual);
	 *   return verified ? { status: 200 } : { status: 401, body: "Invalid signature" };
	 * },
	 * ```
	 */
	verify?: WebhookVerifyHandler;

	/**
	 * Optional schema for the object the handler returns, built with
	 * `@notionhq/apps/schema-builder`. The handler's return type is inferred
	 * from it, and the returned value is validated before the run succeeds.
	 *
	 * Without `outputSchema`, the workflow returns nothing.
	 *
	 * @example
	 * ```ts
	 * outputSchema: j.object({ releaseUrl: j.string() }),
	 * handler: async () => ({ releaseUrl: "https://example.com/releases/1" }),
	 * ```
	 */
	outputSchema?: TOutputSchema;

	handler: (
		event: WorkflowEventForTriggers<TTriggers>,
		context: WorkflowContext<TConnections, TAccess>,
	) => WorkflowHandlerReturn<NoInfer<TOutputSchema>>;
};

/** Configuration that checks trigger keys against declared connections. */
export type WorkflowTriggerConfiguration<
	TTriggers extends readonly [WorkflowTrigger, ...WorkflowTrigger[]],
	TConnections extends WorkflowConnectionDeclarations,
	TAccess extends WorkflowAccessDeclarations = WorkflowAccessDeclarations,
	TOutputSchema extends WorkflowOutputSchema | undefined = undefined,
> = Omit<WorkflowConfiguration<TTriggers, TConnections, TAccess, TOutputSchema>, "triggers"> & {
	triggers: (context: {
		events: WorkflowEventCreators<NoInfer<TConnections>>;
		connections: WorkflowTriggerConnections<NoInfer<TConnections>>;
	}) => TTriggers;
};

/**
 * A workflow capability as returned by {@link workflow}.
 *
 * Note that a workflow carries no key of its own — the build tool derives
 * the capability key from the file the workflow is default-exported from.
 */
export type Workflow<
	TTriggers extends readonly [WorkflowTrigger, ...WorkflowTrigger[]] = readonly [
		WorkflowTrigger,
		...WorkflowTrigger[],
	],
> = {
	_tag: "workflow";
	config: {
		name: string;
		description: string;
		triggers: WorkflowManifestTrigger[];
		connections?: readonly WorkflowConnectionRequirement[];
		access?: WorkflowAccessRequirements;
		deadline?: WorkflowDeadline;
		retry?: WorkflowRetryPolicy;
		/** JSON Schema for the handler's return value. Absent when the workflow returns nothing. */
		outputJsonSchema?: Record<string, unknown>;
	};
	handler: (
		input: WorkflowEventForTriggers<TTriggers> | WebhookVerifyInvocation,
		options?: HandlerOptions,
	) => Promise<WorkflowHandlerResult | WebhookVerifyResponse | undefined>;
};

/**
 * Create a workflow capability.
 *
 * Default-export the result from a file directly under `src/workflows/` —
 * the file's basename becomes the workflow's key.
 *
 * @example
 * ```ts
 * // src/workflows/onPageCreated.ts
 * import { events } from "@notionhq/apps/events";
 * import { workflow } from "@notionhq/apps/workflow";
 *
 * export default workflow({
 *   name: "Send Welcome Email",
 *   description: "Sends a welcome email when a new page is added",
 *   triggers: [events.notionPageCreated()],
 *   handler: async (event, context) => {
 *     console.log(event.page);
 *     await context.step("Process page", async ({ id }) => {
 *       console.log(`Running step ${id}`);
 *       return processPage(event.page);
 *     });
 *   },
 * });
 * ```
 */
export function workflow<
	const TTriggers extends readonly [
		WorkflowTriggerForConnections<NoInfer<TConnections>>,
		...WorkflowTriggerForConnections<NoInfer<TConnections>>[],
	],
	const TConnections extends WorkflowConnectionDeclarations = Record<never, never>,
	const TAccess extends WorkflowAccessDeclarations = Record<never, never>,
	const TOutputSchema extends WorkflowOutputSchema | undefined = undefined,
>(
	configuration: WorkflowTriggerConfiguration<TTriggers, TConnections, TAccess, TOutputSchema>,
): Workflow<TTriggers>;
export function workflow<
	const TTriggers extends readonly [WorkflowTrigger, ...WorkflowTrigger[]],
	const TConnections extends WorkflowConnectionDeclarations = Record<never, never>,
	const TAccess extends WorkflowAccessDeclarations = Record<never, never>,
	const TOutputSchema extends WorkflowOutputSchema | undefined = undefined,
>(
	configuration: WorkflowConfiguration<TTriggers, TConnections, TAccess, TOutputSchema>,
): Workflow<TTriggers>;
export function workflow<
	const TTriggers extends readonly [WorkflowTrigger, ...WorkflowTrigger[]],
	const TConnections extends WorkflowConnectionDeclarations = Record<never, never>,
	const TAccess extends WorkflowAccessDeclarations = Record<never, never>,
	const TOutputSchema extends WorkflowOutputSchema | undefined = undefined,
>(
	configuration:
		| WorkflowConfiguration<TTriggers, TConnections, TAccess, TOutputSchema>
		| WorkflowTriggerConfiguration<TTriggers, TConnections, TAccess, TOutputSchema>,
): Workflow<TTriggers> {
	const requirements = normalizeConnectionRequirements(configuration.connections ?? {});
	const accessRequirements = normalizeWorkflowAccess(configuration.access);
	const connectionDeclarations =
		configuration.connections === undefined
			? undefined
			: structuredClone(configuration.connections);
	const triggers =
		typeof configuration.triggers === "function"
			? configuration.triggers(createWorkflowTriggerContext(configuration.connections))
			: configuration.triggers;
	validateTriggerConnections(triggers, requirements);
	validateWorkflowDeadline(configuration.deadline);
	validateWorkflowRetryPolicy(configuration.retry, "Workflow retry policy");
	if (configuration.retry !== undefined) {
		validateStepRetryBudget(configuration.retry, undefined, "Workflow retry policy");
	}
	const deadline =
		configuration.deadline === undefined ? undefined : structuredClone(configuration.deadline);
	const retry =
		configuration.retry === undefined ? undefined : manifestRetryPolicy(configuration.retry);
	const outputSchema =
		configuration.outputSchema === undefined
			? undefined
			: compileWorkflowOutputSchema(configuration.outputSchema);

	const manualTriggers = triggers.filter(
		(trigger): trigger is ManualWorkflowTrigger => trigger.type === "workflow.manual",
	);
	if (manualTriggers.length > 1) {
		throw new Error("Workflows support at most one manual trigger");
	}

	if (
		configuration.verify !== undefined &&
		!triggers.some((trigger) => trigger.type === WEBHOOK_TRIGGER_TYPE)
	) {
		throw new Error(
			`Workflow "${configuration.name}" declares a verify handler but has no webhook trigger. ` +
				`Add events.webhook() or remove verify.`,
		);
	}

	return {
		_tag: "workflow",
		config: {
			name: configuration.name,
			description: configuration.description,
			triggers: manifestTriggers(triggers, configuration.verify !== undefined),
			...(configuration.connections === undefined ? {} : { connections: requirements }),
			...(configuration.access === undefined ? {} : { access: accessRequirements }),
			...(deadline === undefined ? {} : { deadline }),
			...(retry === undefined ? {} : { retry }),
			...(outputSchema === undefined ? {} : { outputJsonSchema: outputSchema.jsonSchema }),
		},
		async handler(
			input: WorkflowEventForTriggers<TTriggers> | WebhookVerifyInvocation,
			options?: HandlerOptions,
		): Promise<WorkflowHandlerResult | WebhookVerifyResponse | undefined> {
			const invocationStartedAt = Date.now();
			// Errors from steps that already used their retry policy. The run is not
			// retried for these, so they are reported as terminal.
			const exhaustedStepErrors = new WeakSet<object>();
			let cancellation: RunCancellation | undefined;
			try {
				input = await resolveRuntimeInput(input);

				if (isVerifyInvocation(input)) {
					if (configuration.verify === undefined) {
						throw new Error(
							`Workflow "${configuration.name}" received a verify request but does not ` +
								`declare a verify handler`,
						);
					}

					const response = await configuration.verify(
						input.request,
						createCapabilityContext(),
					);

					if (options?.concreteOutput) {
						return response;
					}

					writeOutput({ _tag: "success", value: response });
					return undefined;
				}

				const event = input;
				// Verify requests keep Node's default exit on SIGTERM.
				cancellation = listenForRunCancellation();
				const runMetadata = readRunMetadata();
				const baseContext = createCapabilityContext();
				const step = createStep(runMetadata, {
					defaultRetry: configuration.retry,
					exhaustedErrors: exhaustedStepErrors,
					invocationStartedAt,
					signal: cancellation.signal,
				});
				const capabilityContext: WorkflowContext<TConnections, TAccess> = {
					...baseContext,
					access: hydrateWorkflowAccess<TAccess>(accessRequirements),
					connections: createWorkflowConnections(
						baseContext.notion,
						connectionDeclarations,
					),
					...runMetadata,
					step,
					wait: createWait<TConnections>(step, requirements, triggers),
					signal: cancellation.signal,
					sync: createWorkflowSyncClient(),
					workflows: createWorkflowInvocations(runMetadata, step),
				};
				if (event.type === "workflow.manual") {
					const manualTrigger = manualTriggers[0];
					if (manualTrigger === undefined) {
						throw new Error(
							"Received workflow.manual for a workflow without a manual trigger",
						);
					}
					validateManualWorkflowInput(manualTrigger, event.input);
				}
				const handlerOutput: unknown = await configuration.handler(
					event,
					capabilityContext,
				);
				const result: WorkflowHandlerResult =
					outputSchema === undefined
						? { status: "success" }
						: {
								status: "success",
								output: serializeWorkflowOutput(
									handlerOutput,
									outputSchema.validate,
								),
							};

				if (options?.concreteOutput) {
					return result;
				}

				writeOutput({ _tag: "success", value: result });
			} catch (err) {
				if (err instanceof WorkflowWaitInterrupt) {
					if (options?.concreteOutput) {
						return err.result;
					}

					writeOutput({
						_tag: "wait",
						wait: err.result.wait,
					});
					return;
				}

				const error = new ExecutionError(err);
				const terminal =
					err instanceof FatalError ||
					(isObject(err) && exhaustedStepErrors.has(err)) ||
					(!(err instanceof RetryableError) &&
						!isRetryableError(err, configuration.retry));

				if (!options?.concreteOutput) {
					writeOutput({
						_tag: "error",
						error:
							!terminal && err instanceof RateLimitError
								? {
										_tag: "rate_limit",
										name: err.name,
										message: err.message,
										...(err.retryAfter === undefined
											? {}
											: { retryAfter: err.retryAfter }),
									}
								: !terminal && err instanceof RetryableError
									? {
											_tag: "retryable",
											name: err.name,
											message: err.message,
											trace: err.stack,
											...(err.retryAfterMs === undefined
												? {}
												: { retryAfterMs: err.retryAfterMs }),
										}
									: terminal
										? {
												_tag: "terminal",
												name: err instanceof Error ? err.name : error.name,
												message:
													err instanceof Error
														? err.message
														: error.message,
												trace:
													err instanceof Error ? err.stack : error.stack,
											}
										: {
												name: error.name,
												message: error.message,
												trace: error.stack,
											},
					});
				}

				throw error;
			} finally {
				cancellation?.dispose();
			}
		},
	};
}

/**
 * The triggers as the manifest records them. The manifest only records that a
 * verify handler exists, flagged on the webhook trigger; the handler itself
 * ships in the app bundle and runs via a verify invocation.
 */
function manifestTriggers(
	triggers: readonly WorkflowTrigger[],
	hasVerify: boolean,
): WorkflowManifestTrigger[] {
	return triggers.map((trigger) =>
		hasVerify && trigger.type === WEBHOOK_TRIGGER_TYPE
			? { ...trigger, hasVerify: true as const }
			: trigger,
	);
}

/** The platform stores workflow output with its run, so it must stay small. */
const MAX_WORKFLOW_OUTPUT_BYTES = 256 * 1024;

/**
 * Validates the JSON the platform will receive rather than the in-memory value,
 * so values that change during serialization, such as `Date` or `toJSON`
 * objects, are checked in the form callers will see.
 *
 * A mismatch is a bug in the handler, so it fails the run without retrying.
 */
function serializeWorkflowOutput(
	value: unknown,
	validate: (value: unknown) => void,
): Record<string, unknown> {
	try {
		const json = value === undefined ? undefined : JSON.stringify(value);
		const bytes = json === undefined ? 0 : Buffer.byteLength(json, "utf8");
		if (bytes > MAX_WORKFLOW_OUTPUT_BYTES) {
			throw new Error(
				`Invalid workflow output: must be at most ${MAX_WORKFLOW_OUTPUT_BYTES} bytes as JSON, got ${bytes}.`,
			);
		}
		const serialized: unknown = json === undefined ? undefined : JSON.parse(json);
		validate(serialized);
		// SAFETY: validate only accepts values matching the j.object() output schema.
		return serialized as Record<string, unknown>;
	} catch (error) {
		throw new FatalError(error instanceof Error ? error.message : String(error));
	}
}

/**
 * Whether an invocation is the platform's synchronous verify dispatch. The
 * platform sends workflow invocations as a trigger event and verify
 * invocations as a `{ request }` object, so the shape disambiguates.
 */
function isVerifyInvocation(input: unknown): input is WebhookVerifyInvocation {
	return typeof input === "object" && input !== null && "request" in input;
}

/** Context passed to a workflow step. */
export type StepContext = {
	/**
	 * An identifier that is stable for this step across attempts of the workflow run.
	 * Use this to make side effects idempotent when the workflow is retried.
	 */
	id: string;
	/**
	 * Durable key-value state shared by every run of this workflow.
	 *
	 * State is exposed on the step context because every operation participates
	 * in the active step's durability and replay boundary.
	 */
	state: WorkflowState;
	/**
	 * Aborts when the workflow run is asked to stop, for example because it was
	 * cancelled or reached its time limit, or when this attempt exceeds the
	 * step's `timeoutMs`. Pass it to `fetch` and other abortable APIs, or call
	 * `signal.throwIfAborted()` between units of work. Each attempt gets a new
	 * signal.
	 *
	 * Cancellation is cooperative: the step keeps running until the callback
	 * observes the signal. A step that throws because of the signal fails and
	 * is not saved, so it runs again if the workflow is retried. A timed-out
	 * attempt that ignores the signal can keep running alongside the retry.
	 */
	signal: AbortSignal;
};

type WorkflowStepResult<T> = T extends void ? null : T;

/** Options for a workflow step. */
export type WorkflowStepOptions = {
	/**
	 * A stable key used to identify this step across attempts of a workflow run.
	 * Pass a string for a single key segment or an array of strings for a composite
	 * key. Defaults to the step name. Changing the key creates a new logical step.
	 */
	key?: string | string[];
	/**
	 * Maximum duration of one invocation of the step callback. When it elapses,
	 * the attempt fails and its `signal` is aborted.
	 */
	timeoutMs?: number;
	/** Retry policy for failures from this step callback. */
	retry?: WorkflowRetryPolicy;
};

type WorkflowStep = {
	<T>(name: string, fn: (context: StepContext) => T | Promise<T>): Promise<WorkflowStepResult<T>>;
	<T>(
		name: string,
		options: WorkflowStepOptions,
		fn: (context: StepContext) => T | Promise<T>,
	): Promise<WorkflowStepResult<T>>;
};

type WorkflowWait<
	TConnections extends WorkflowConnectionDeclarations = WorkflowConnectionDeclarations,
> = {
	/**
	 * Suspend the workflow until a relative duration has elapsed or an absolute
	 * date has been reached.
	 */
	until(name: string, options: WorkflowWaitUntilOptions): Promise<void>;
	/**
	 * Suspend the workflow until a person submits values matching the input
	 * schema. The returned value is restored from the durable workflow
	 * checkpoint when the request is fulfilled.
	 */
	forInput<const TSchema extends WorkflowInputSchema>(
		name: string,
		options: WorkflowWaitForInputOptions<TSchema>,
	): Promise<WorkflowInputResponse<TSchema>>;
	/**
	 * Suspend the workflow until an event of the specified type occurs after the
	 * wait starts. Resolves with the event. If the timeout elapses first, it
	 * throws a `FatalError` and the run fails without retrying.
	 */
	forEvent<const TTrigger extends WorkflowWaitEventTriggerForConnections<TConnections>>(
		name: string,
		options: WorkflowWaitForEventOptions<TTrigger>,
	): Promise<WorkflowEventForTrigger<TTrigger>>;
};

/** Options for invoking another workflow from a workflow. */
export type WorkflowInvokeOptions = {
	/**
	 * A stable key for this invocation within the calling run. Defaults to the
	 * target workflow's key. Pass one when invoking the same workflow more than
	 * once, such as in a loop. Changing the key starts a new run.
	 */
	key?: string | string[];
};

/** The input a workflow accepts through its manual trigger. */
export type WorkflowInvokeInput<TWorkflow> =
	TWorkflow extends Workflow<infer TTriggers>
		? ManualWorkflowInput<Extract<TTriggers[number], ManualWorkflowTrigger>>
		: never;

type ManualWorkflowInput<TTrigger> = [TTrigger] extends [never]
	? never
	: TTrigger extends ManualWorkflowTrigger<infer TInput>
		? TInput
		: never;

/** A handle to a workflow run started with `start()`. */
export type WorkflowRunHandle = {
	/** The ID of the started run. */
	runId: string;
};

/** Invoke other workflows in the same app. The target must have a manual trigger. */
export type WorkflowInvocations = {
	/**
	 * Start another workflow and continue without waiting for it. Resolves once
	 * the run is created. The started run's success or failure does not affect
	 * this workflow.
	 *
	 * Starting is durable: a retry or replay of this workflow returns the same
	 * run instead of starting another. Called directly from the handler, the
	 * start is its own step. Called inside `context.step()`, it is part of that
	 * step, and retries of that step return the same run.
	 */
	start<TWorkflow extends Workflow>(
		workflow: TWorkflow,
		input: WorkflowInvokeInput<TWorkflow>,
		options?: WorkflowInvokeOptions,
	): Promise<WorkflowRunHandle>;
};

/** Context passed to a workflow handler. */
export type WorkflowContext<
	TConnections extends WorkflowConnectionDeclarations = WorkflowConnectionDeclarations,
	TAccess extends WorkflowAccessDeclarations = WorkflowAccessDeclarations,
> = CapabilityContext & {
	/** Resolved NaC bindings for this invocation; UI-granted resources are not added here. */
	readonly access: WorkflowAccess<TAccess>;
	connections: WorkflowConnections<TConnections>;
} & RunMetadata & {
		/**
		 * Run an observable unit of work within the workflow.
		 *
		 * The callback may be synchronous or asynchronous, but `step` always returns
		 * a promise. Its return value must be JSON-serializable so it can be included
		 * in the success event written to stdout. A callback that returns `undefined`
		 * is normalized to `null`, which keeps first runs and saved replays consistent.
		 *
		 * The step name is used to derive its stable replay key by default. Step key
		 * inputs must be unique within a workflow invocation. Pass an explicit string
		 * or array of strings when display names are duplicated or may change.
		 *
		 * @example
		 * ```ts
		 * const page = await context.step("Fetch page", async ({ id }) => {
		 *   console.log(`Running step ${id}`);
		 *   return context.notion.pages.retrieve({ page_id: pageId });
		 * });
		 * ```
		 */
		step: WorkflowStep;
		/** Suspend this workflow without consuming compute resources. */
		wait: WorkflowWait<TConnections>;
		/**
		 * Aborts when the workflow run is asked to stop, for example because it was
		 * cancelled or reached its time limit. Pass it to `fetch` and other
		 * abortable APIs, or call `signal.throwIfAborted()` between units of work,
		 * inside or outside a step.
		 *
		 * Cancellation is cooperative: code keeps running until it observes the
		 * signal. A step that throws because of the signal fails and is not saved,
		 * so it runs again if the workflow is retried. New steps do not start once
		 * the signal has aborted.
		 */
		readonly signal: AbortSignal;
		/** Trigger declared syncs or apply event-driven sync changes. */
		sync: WorkflowSyncClient;
		/** Invoke other workflows in the same app. */
		workflows: WorkflowInvocations;
	};

type StepCompletionType = "success" | "failure";
type StepEventType = "started" | "success" | "failure" | "replayed" | "completed";

type WorkflowStepCheckpoint = {
	version: 0;
	key: string;
	value: unknown;
	sourceRunId: string;
	sourceStepId: string;
};

const WORKFLOW_STEP_DIRECTORY_ENV_KEY = "NOTION_WORKFLOW_STEP_DIRECTORY";
const MAX_WORKFLOW_STEP_KEY_LENGTH = 43;
const WORKFLOW_STEP_KEY_PATTERN = /^[A-Za-z0-9_-]+$/;
const WORKFLOW_STEP_KEY_HASH_DOMAIN = "notion-workflow-step-key";
const WORKFLOW_STEP_KEY_HASH_VERSION = 1;

function writeStepEvent(
	event: StepEventType,
	value: Record<string, unknown>,
	startedAt: number,
	emittedAt = performance.now(),
) {
	const tag = `__notion_step_${event}__`;
	process.stdout.write(
		`\n<${tag}>${JSON.stringify({
			...value,
			timestampMs: performance.timeOrigin + emittedAt,
			durationMs: emittedAt - startedAt,
		})}</${tag}>\n`,
	);
}

function createStep(
	metadata: Pick<RunMetadata, "runGroupId" | "workerId">,
	options: {
		defaultRetry?: WorkflowRetryPolicy | undefined;
		exhaustedErrors?: WeakSet<object>;
		/** When the handler started. Step retries share one invocation's budget. */
		invocationStartedAt: number;
		/** Aborts when the run is cancelled. */
		signal: AbortSignal;
	},
): WorkflowContext["step"] {
	const { runGroupId } = metadata;
	const { defaultRetry, exhaustedErrors, invocationStartedAt, signal } = options;
	const stepNameByKey = new Map<string, string>();

	async function step<T>(
		name: string,
		optionsOrFn: WorkflowStepOptions | ((context: StepContext) => T | Promise<T>),
		maybeFn?: (context: StepContext) => T | Promise<T>,
	): Promise<WorkflowStepResult<T>> {
		const stepOptions = typeof optionsOrFn === "function" ? undefined : optionsOrFn;
		const fn = typeof optionsOrFn === "function" ? optionsOrFn : maybeFn;
		if (!fn) {
			throw new Error(`Workflow step "${name}" is missing its callback.`);
		}

		validateWorkflowStepOptions(stepOptions);
		const retry = stepOptions?.retry ?? defaultRetry ?? DEFAULT_WORKFLOW_RETRY_POLICY;
		validateStepRetryBudget(retry, stepOptions?.timeoutMs, `Workflow step "${name}"`);

		const keyInput = stepOptions?.key ?? name;
		const key = createWorkflowStepKey(typeof keyInput === "string" ? [keyInput] : keyInput);
		const existingStepName = stepNameByKey.get(key);
		if (existingStepName !== undefined) {
			throw new Error(
				`Workflow step "${name}" has the same key as "${existingStepName}". Step names are used as keys by default; provide a unique name or an explicit key.`,
			);
		}
		stepNameByKey.set(key, name);

		const id = `${runGroupId}:${key}`;
		const event = { id, name, key };
		const startedAt = performance.now();

		writeStepEvent("started", event, startedAt, startedAt);

		try {
			const checkpoint = await readWorkflowStepCheckpoint(key);
			if (checkpoint) {
				writeStepEvent(
					"replayed",
					{
						...event,
						sourceRunId: checkpoint.sourceRunId,
						sourceStepId: checkpoint.sourceStepId,
					},
					startedAt,
				);
				writeStepEvent(
					"completed",
					{
						...event,
						type: "success" satisfies StepCompletionType,
					},
					startedAt,
				);
				return checkpoint.value as WorkflowStepResult<T>;
			}

			// Replays above stay allowed after cancellation because they do no new work.
			signal.throwIfAborted();
			const value = normalizeWorkflowStepResult(
				await runWorkflowStepCallback({
					metadata,
					id,
					key,
					fn,
					timeoutMs: stepOptions?.timeoutMs,
					retry,
					invocationStartedAt,
					signal,
				}),
			);
			writeStepEvent("success", { ...event, value }, startedAt);
			writeStepEvent(
				"completed",
				{
					...event,
					type: "success" satisfies StepCompletionType,
				},
				startedAt,
			);
			return value;
		} catch (caught) {
			// A retry that no longer fits in this invocation is left to the run's
			// retry policy, so the step is not reported as having used its retries.
			const deferred = caught instanceof DeferredStepRetry;
			const err = deferred ? caught.error : caught;
			const error = new ExecutionError(err);
			writeStepEvent(
				"failure",
				{
					...event,
					error: {
						name: error.name,
						message: error.message,
						trace: error.stack,
					},
				},
				startedAt,
			);
			writeStepEvent(
				"completed",
				{
					...event,
					type: "failure" satisfies StepCompletionType,
				},
				startedAt,
			);
			// A cancelled step did not use up its retries, so the run stays retryable.
			if (!deferred && !signal.aborted && isObject(err)) {
				exhaustedErrors?.add(err);
			}
			throw err;
		}
	}

	return step;
}

/** Signals that a step retry must be deferred to a later workflow invocation. */
class DeferredStepRetry {
	constructor(readonly error: unknown) {}
}

type RunCancellation = { signal: AbortSignal; dispose: () => void };

/**
 * Converts the first SIGTERM into an abort of the run's signal.
 *
 * The platform sends SIGTERM shortly before it stops a workflow's sandbox.
 * Without a listener, Node exits immediately with no output. With one, the
 * process keeps running: the current step sees the abort and the run reports
 * its failure, then the entry script exits once the handler settles. A step
 * that ignores the abort keeps running until the sandbox is stopped.
 *
 * The listener is removed after the first signal and when the handler ends,
 * so a second SIGTERM, or one outside a run, uses Node's default exit.
 */
function listenForRunCancellation(): RunCancellation {
	const controller = new AbortController();
	const onSignal = () => {
		dispose();
		controller.abort(new DOMException("Workflow run was cancelled (SIGTERM).", "AbortError"));
	};
	const dispose = () => {
		process.off("SIGTERM", onSignal);
	};
	process.on("SIGTERM", onSignal);
	return { signal: controller.signal, dispose };
}

class WorkflowWaitInterrupt extends Error {
	constructor(readonly result: WorkflowWaitResult) {
		super(workflowWaitMessage(result));
		this.name = "WorkflowWaitInterrupt";
	}
}

function workflowWaitMessage(result: WorkflowWaitResult): string {
	if (result.wait.type === "until") {
		return `Workflow is waiting until ${new Date(result.wait.resumeAtMs).toISOString()}`;
	}
	if (result.wait.type === "event") {
		return `Workflow is waiting for event: ${result.wait.name}`;
	}
	return `Workflow is waiting for input: ${result.wait.name}`;
}

function createWait<TConnections extends WorkflowConnectionDeclarations>(
	step: WorkflowStep,
	connectionRequirements: readonly WorkflowConnectionRequirement[],
	triggers: readonly WorkflowTrigger[],
): WorkflowWait<TConnections> {
	const inputWaitNameByKey = new Map<string, string>();
	const eventWaitNameByKey = new Map<string, string>();

	return {
		async until(name, options) {
			const keyInput = options.key ?? name;
			const keySegments = [
				"wait",
				"until",
				...(typeof keyInput === "string" ? [keyInput] : keyInput),
			];
			const resumeAtMs = await step(name, { key: keySegments }, () =>
				resolveWaitUntilMs(options),
			);

			if (Date.now() >= resumeAtMs) {
				return;
			}

			throw new WorkflowWaitInterrupt({
				status: "waiting",
				wait: {
					type: "until",
					name,
					resumeAtMs,
				},
			});
		},
		async forInput(name, options) {
			const keyInput = options.key ?? name;
			const stepKey = createWorkflowStepKey([
				"wait",
				"input",
				...(typeof keyInput === "string" ? [keyInput] : keyInput),
			]);
			const existingName = inputWaitNameByKey.get(stepKey);
			if (existingName !== undefined) {
				throw new Error(
					`Workflow input wait "${name}" has the same key as "${existingName}". Provide a unique name or an explicit key.`,
				);
			}
			inputWaitNameByKey.set(stepKey, name);

			const checkpoint = await readWorkflowStepCheckpoint(stepKey);
			if (checkpoint) {
				if (!isWorkflowInputResponse(checkpoint.value, options.input)) {
					throw new Error(
						`Workflow input checkpoint ${hashWorkflowStepKey(stepKey)} does not match its schema.`,
					);
				}
				return checkpoint.value;
			}

			throw new WorkflowWaitInterrupt({
				status: "waiting",
				wait: {
					type: "input",
					name,
					stepKey,
					input: options.input,
				},
			});
		},
		async forEvent<TTrigger extends WorkflowWaitEventTriggerForConnections<TConnections>>(
			name: string,
			options: WorkflowWaitForEventOptions<TTrigger>,
		): Promise<WorkflowEventForTrigger<TTrigger>> {
			const keyInput = options.key ?? name;
			const keySegments = typeof keyInput === "string" ? [keyInput] : keyInput;
			const stepKey = createWorkflowStepKey(["wait", "event", ...keySegments]);
			const existingName = eventWaitNameByKey.get(stepKey);
			if (existingName !== undefined) {
				throw new Error(
					`Workflow event wait "${name}" has the same key as "${existingName}". Provide a unique name or an explicit key.`,
				);
			}
			eventWaitNameByKey.set(stepKey, name);
			validateTriggerConnections([options.event], connectionRequirements);
			if (triggers.some((trigger) => isSameTriggerEvent(trigger, options.event))) {
				throw new Error(
					`Workflow event wait "${name}" waits for ${options.event.type}, which is also a trigger of this workflow. An event can start a workflow or resume a wait, not both.`,
				);
			}

			const timeoutAtMs = await step(
				name,
				{ key: ["wait", "event-timeout", ...keySegments] },
				() => resolveWaitUntilMs({ after: options.timeout }),
			);

			const checkpoint = await readWorkflowStepCheckpoint(stepKey);
			if (checkpoint) {
				if (checkpoint.value === null) {
					throw new FatalError(`Workflow event wait "${name}" timed out.`);
				}
				if (!isRecord(checkpoint.value) || checkpoint.value.type !== options.event.type) {
					throw new Error(
						`Workflow event checkpoint ${hashWorkflowStepKey(stepKey)} does not match event type ${options.event.type}.`,
					);
				}
				return checkpoint.value as WorkflowEventForTrigger<TTrigger>;
			}

			throw new WorkflowWaitInterrupt({
				status: "waiting",
				wait: {
					type: "event",
					name,
					stepKey,
					event: { ...options.event, ...(options.scope ? { scope: options.scope } : {}) },
					timeoutAtMs,
				},
			});
		},
	};
}

function isSameTriggerEvent(left: WorkflowTrigger, right: WorkflowTrigger): boolean {
	const connectionKey = (trigger: WorkflowTrigger) =>
		"connectionKey" in trigger ? trigger.connectionKey : undefined;
	return left.type === right.type && connectionKey(left) === connectionKey(right);
}

const workflowKeys = new WeakMap<object, string>();

/**
 * Records the capability key of a workflow. The build tool calls this from the
 * generated entry, because a workflow's key comes from its file name.
 *
 * @internal
 */
export function bindWorkflowKey(workflow: object, key: string): void {
	workflowKeys.set(workflow, key);
}

const WORKFLOW_START_RUN_PATH = "/api/v3/workersWorkflowStartRun";

/** The step attempt whose callback is running, if any. */
type ActiveStep = {
	key: string;
	/** Invocation keys used in this attempt. A retried attempt starts empty. */
	invocationNameByKey: Map<string, string>;
};

const activeStep = new AsyncLocalStorage<ActiveStep>();

function createWorkflowInvocations(
	metadata: Pick<RunMetadata, "runGroupId" | "workerId">,
	step: WorkflowStep,
): WorkflowInvocations {
	const invocationNameByKey = new Map<string, string>();

	return {
		async start(target, input, options) {
			const { targetKey, name } = resolveInvocationTarget(target, input);
			const keyOption = options?.key ?? targetKey;
			const keyInput = typeof keyOption === "string" ? [keyOption] : [...keyOption];
			const enclosingStep = activeStep.getStore();
			// Inside a step, the start key is scoped to that step so the same
			// invocation in two steps starts two runs.
			const startKey =
				enclosingStep === undefined
					? ["workflow", "start", ...keyInput]
					: [enclosingStep.key, "workflow", "start", ...keyInput];
			const stepKey = createWorkflowStepKey(startKey);
			claimInvocationKey(enclosingStep?.invocationNameByKey ?? invocationNameByKey, {
				stepKey,
				name,
			});

			const request = () =>
				requestWorkflowStartRun({
					workerId: metadata.workerId,
					runGroupId: metadata.runGroupId,
					stepKey,
					targetCapabilityKey: targetKey,
					input,
				});
			if (enclosingStep !== undefined) {
				// The platform returns the run it already started for this key, so
				// a retried step does not start another.
				return request();
			}
			return step(`Start ${name}`, { key: startKey }, request);
		},
	};
}

function resolveInvocationTarget(
	target: Workflow,
	input: unknown,
): { targetKey: string; name: string } {
	const targetKey = workflowKeys.get(target);
	if (targetKey === undefined) {
		throw new FatalError(
			"context.workflows can only invoke workflows default-exported from src/workflows/ in this app.",
		);
	}
	const manualTrigger = target.config.triggers.find(
		(trigger): trigger is ManualWorkflowTrigger => trigger.type === "workflow.manual",
	);
	if (manualTrigger === undefined) {
		throw new FatalError(
			`Workflow "${targetKey}" has no manual trigger, so it cannot be invoked. Add events.manual() to its triggers.`,
		);
	}
	try {
		validateManualWorkflowInput(manualTrigger, input);
	} catch (error) {
		throw new FatalError(
			`Invalid input for workflow "${targetKey}": ${error instanceof Error ? error.message : String(error)}`,
		);
	}
	return { targetKey, name: target.config.name };
}

/**
 * Rejects two invocations with the same key in one scope. The second would
 * silently return the first one's run.
 */
function claimInvocationKey(
	invocationNameByKey: Map<string, string>,
	invocation: { stepKey: string; name: string },
): void {
	const existingName = invocationNameByKey.get(invocation.stepKey);
	if (existingName !== undefined) {
		throw new Error(
			`Workflow invocation "${invocation.name}" has the same key as "${existingName}". Pass a unique key when invoking a workflow more than once.`,
		);
	}
	invocationNameByKey.set(invocation.stepKey, invocation.name);
}

async function requestWorkflowStartRun(request: {
	workerId: string;
	runGroupId: string;
	stepKey: string;
	targetCapabilityKey: string;
	input: unknown;
}): Promise<WorkflowRunHandle> {
	const baseUrl = process.env.NOTION_API_BASE_URL;
	const token = process.env.NOTION_API_TOKEN;
	const capabilityKey = process.env.NOTION_WORKFLOW_CAPABILITY_KEY;
	if (!baseUrl || !token || !capabilityKey) {
		throw new FatalError(
			"Invoking workflows is only available when the workflow runtime provides NOTION_API_BASE_URL, NOTION_API_TOKEN, and NOTION_WORKFLOW_CAPABILITY_KEY.",
		);
	}

	const response = await fetch(`${baseUrl.replace(/\/$/, "")}${WORKFLOW_START_RUN_PATH}`, {
		method: "POST",
		headers: {
			authorization: `Bearer ${token}`,
			"content-type": "application/json",
		},
		body: JSON.stringify({ ...request, capabilityKey }),
	});
	if (!response.ok) {
		const message = await readErrorMessage(response);
		const description = `Starting workflow "${request.targetCapabilityKey}" failed with status ${response.status}${message === undefined ? "" : `: ${message}`}`;
		// Rejected requests fail the same way on every attempt.
		if (response.status >= 400 && response.status < 500 && response.status !== 429) {
			throw new FatalError(description);
		}
		throw new RetryableError(description);
	}
	const body: unknown = await response.json();
	if (!isRecord(body) || typeof body.runId !== "string") {
		throw new Error("Starting a workflow returned an unexpected response.");
	}
	return { runId: body.runId };
}

async function readErrorMessage(response: Response): Promise<string | undefined> {
	try {
		const body: unknown = await response.json();
		return isRecord(body) && typeof body.message === "string" ? body.message : undefined;
	} catch {
		return undefined;
	}
}

function isWorkflowInputResponse<TSchema extends WorkflowInputSchema>(
	value: unknown,
	schema: TSchema,
): value is WorkflowInputResponse<TSchema> {
	if (
		!isRecord(value) ||
		!isRecord(value.value) ||
		typeof value.respondedBy !== "string" ||
		typeof value.respondedAt !== "string"
	) {
		return false;
	}

	if (Object.keys(value.value).some((key) => !Object.hasOwn(schema, key))) {
		return false;
	}

	for (const [key, field] of Object.entries(schema)) {
		const fieldValue = value.value[key];
		if (fieldValue === undefined && field.optional === true) {
			continue;
		}
		if (!isWorkflowInputFieldValue(fieldValue, field)) {
			return false;
		}
	}
	return true;
}

function isWorkflowInputFieldValue(value: unknown, field: WorkflowInputField<unknown>): boolean {
	switch (field.type) {
		case "text":
			return typeof value === "string";
		case "date":
			return (
				isRecord(value) &&
				typeof value.start === "string" &&
				(value.end === undefined || typeof value.end === "string") &&
				(value.timeZone === undefined || typeof value.timeZone === "string")
			);
		case "number":
			return typeof value === "number" && Number.isFinite(value);
		case "checkbox":
			return typeof value === "boolean";
		case "select":
			return (
				typeof value === "string" &&
				field.options !== undefined &&
				field.options.includes(value)
			);
		case "person":
		case "files":
			return Array.isArray(value) && value.every((item) => typeof item === "string");
		case "multi_select": {
			const options = field.options;
			return (
				Array.isArray(value) &&
				options !== undefined &&
				value.every((item) => typeof item === "string" && options.includes(item))
			);
		}
	}
}

function resolveWaitUntilMs(options: WorkflowWaitUntilOptions): number {
	const now = Date.now();
	if ("at" in options) {
		const resumeAtMs = options.at.getTime();
		if (!Number.isFinite(resumeAtMs)) {
			throw new Error("Workflow wait date must be valid.");
		}
		assertWaitFitsCheckpointRetention(resumeAtMs - now);
		return resumeAtMs;
	}

	const durationMs = workflowWaitDurationMs(options.after);
	assertWaitFitsCheckpointRetention(durationMs);
	return now + durationMs;
}

function workflowWaitDurationMs(duration: WorkflowWaitDuration): number {
	const entries = Object.entries(duration);
	if (entries.length === 0) {
		throw new Error("Workflow wait duration must specify at least one unit.");
	}

	let durationMs = 0;
	for (const [unit, value] of entries) {
		if (!Number.isFinite(value) || !Number.isInteger(value) || value <= 0) {
			throw new Error("Workflow wait duration values must be positive finite integers.");
		}

		durationMs += value * waitDurationUnitMultiplier(unit);
		if (!Number.isSafeInteger(durationMs)) {
			throw new Error("Workflow wait duration is too large.");
		}
	}

	return durationMs;
}

function assertWaitFitsCheckpointRetention(durationMs: number): void {
	if (durationMs >= MAX_WORKFLOW_WAIT_MS) {
		throw new Error("Workflow waits must be shorter than 7 days.");
	}
}

function waitDurationUnitMultiplier(unit: string): number {
	switch (unit) {
		case "milliseconds":
			return 1;
		case "seconds":
			return 1_000;
		case "minutes":
			return 60_000;
		case "hours":
			return 3_600_000;
		case "days":
			return 86_400_000;
		case "weeks":
			return 604_800_000;
		default:
			throw new Error(`Unsupported workflow wait duration unit: ${unit}`);
	}
}

async function runWorkflowStepCallback<T>(args: {
	metadata: Pick<RunMetadata, "workerId">;
	id: string;
	key: string;
	fn: (context: StepContext) => T | Promise<T>;
	timeoutMs: number | undefined;
	retry: WorkflowRetryPolicy;
	invocationStartedAt: number;
	signal: AbortSignal;
}): Promise<T> {
	let attempt = 1;

	while (true) {
		// Each attempt gets its own state scope so a failed or timed-out attempt
		// never commits, and its pending mutations are not visible to the retry.
		const stepState = createWorkflowStepState(args.metadata, args.id);
		// The attempt's signal aborts on run cancellation or this attempt's timeout.
		const timeoutController = new AbortController();
		const context: StepContext = {
			id: args.id,
			signal: AbortSignal.any([args.signal, timeoutController.signal]),
			state: stepState.state,
		};
		try {
			return await activeStep.run({ key: args.key, invocationNameByKey: new Map() }, () =>
				stepState.run(() =>
					runWithTimeout(args.fn(context), args.timeoutMs, timeoutController),
				),
			);
		} catch (error) {
			// A cancelled run stops here instead of retrying the step.
			if (
				args.signal.aborted ||
				attempt >= args.retry.maxAttempts ||
				!isRetryableError(error, args.retry)
			) {
				throw error;
			}

			const delayMs =
				explicitRetryDelay(error) ?? calculateRetryDelay(args.retry, attempt - 1);
			// Earlier work in this invocation shares the sandbox, so measure from
			// the handler's start rather than the step's.
			const nextAttemptBudgetMs = delayMs + (args.timeoutMs ?? 0);
			if (
				Date.now() - args.invocationStartedAt + nextAttemptBudgetMs >
				WORKFLOW_STEP_RETRY_BUDGET_MS
			) {
				throw new DeferredStepRetry(error);
			}
			if (delayMs > 0) {
				await new Promise((resolve) => setTimeout(resolve, delayMs));
			}
			args.signal.throwIfAborted();
			attempt += 1;
		}
	}
}

/**
 * Whether a policy permits retrying an error. `FatalError` never retries and
 * `RetryableError` always does; `retryOn`, when set, limits everything else.
 */
function isRetryableError(error: unknown, policy: WorkflowRetryPolicy | undefined): boolean {
	if (error instanceof FatalError) {
		return false;
	}
	if (error instanceof RetryableError || policy?.retryOn === undefined) {
		return true;
	}
	const matchers: readonly WorkflowRetryableErrorMatcher[] =
		typeof policy.retryOn === "function" ? [policy.retryOn] : policy.retryOn;
	return matchers.some((matcher) =>
		isErrorClass(matcher)
			? error instanceof matcher
			: (matcher as (error: unknown) => boolean)(error),
	);
}

function isErrorClass(
	matcher: WorkflowRetryableErrorMatcher,
): matcher is abstract new (...args: never[]) => Error {
	return matcher === Error || matcher.prototype instanceof Error;
}

/** A delay requested by the error itself, which takes precedence over backoff. */
function explicitRetryDelay(error: unknown): number | undefined {
	if (error instanceof RetryableError) {
		return error.retryAfterMs;
	}
	if (error instanceof RateLimitError && error.retryAfter !== undefined) {
		return error.retryAfter * 1_000;
	}
	return undefined;
}

async function runWithTimeout<T>(
	value: T | Promise<T>,
	timeoutMs: number | undefined,
	timeoutController: AbortController,
): Promise<T> {
	if (timeoutMs === undefined) {
		return await value;
	}

	let timer: NodeJS.Timeout | undefined;
	try {
		return await Promise.race([
			value,
			new Promise<never>((_resolve, reject) => {
				timer = setTimeout(() => {
					const error = new Error(`Workflow step timed out after ${timeoutMs}ms.`);
					error.name = "StepTimeoutError";
					timeoutController.abort(error);
					reject(error);
				}, timeoutMs);
			}),
		]);
	} finally {
		if (timer !== undefined) {
			clearTimeout(timer);
		}
	}
}

/**
 * Exponential backoff with "equal jitter": half the delay is fixed and half is
 * random, so retries from many runs spread out without retrying too early.
 */
function calculateRetryDelay(policy: WorkflowRetryPolicy, retryIndex: number): number {
	const delayMs = maxRetryDelay(policy, retryIndex);
	return delayMs / 2 + Math.random() * (delayMs / 2);
}

/** The backoff delay before jitter, which is also its upper bound. */
function maxRetryDelay(policy: WorkflowRetryPolicy, retryIndex: number): number {
	const initialDelayMs = policy.initialDelayMs ?? DEFAULT_WORKFLOW_RETRY_POLICY.initialDelayMs;
	const maxDelayMs = policy.maxDelayMs ?? DEFAULT_WORKFLOW_RETRY_POLICY.maxDelayMs;
	const backoffMultiplier =
		policy.backoffMultiplier ?? DEFAULT_WORKFLOW_RETRY_POLICY.backoffMultiplier;
	return Math.min(initialDelayMs * backoffMultiplier ** retryIndex, maxDelayMs);
}

/**
 * Rejects retry policies whose longest possible retry delays plus step timeouts
 * cannot fit in one workflow invocation.
 */
function validateStepRetryBudget(
	policy: WorkflowRetryPolicy,
	timeoutMs: number | undefined,
	label: string,
): void {
	let worstCaseMs = (timeoutMs ?? 0) * policy.maxAttempts;
	for (let retryIndex = 0; retryIndex < policy.maxAttempts - 1; retryIndex++) {
		worstCaseMs += maxRetryDelay(policy, retryIndex);
	}
	if (worstCaseMs > WORKFLOW_STEP_RETRY_BUDGET_MS) {
		throw new Error(
			`${label} can spend up to ${worstCaseMs}ms on retry delays and timeouts, which exceeds the ${WORKFLOW_STEP_RETRY_BUDGET_MS}ms limit for one workflow invocation. Lower maxAttempts, the retry delays, or timeoutMs, or use context.wait.until() for longer waits.`,
		);
	}
}

/** The manifest carries only serializable policy fields; `retryOn` runs in the SDK. */
function manifestRetryPolicy(policy: WorkflowRetryPolicy): Omit<WorkflowRetryPolicy, "retryOn"> {
	const { retryOn: _retryOn, ...rest } = policy;
	return structuredClone(rest);
}

function validateWorkflowDeadline(deadline: WorkflowDeadline | undefined): void {
	if (deadline === undefined) {
		return;
	}
	assertPositiveFiniteNumber(deadline.afterMs, "Workflow deadline afterMs");
	assertAtMost(deadline.afterMs, MAX_WORKFLOW_DURATION_MS, "Workflow deadline afterMs");
}

function validateWorkflowRetryPolicy(policy: WorkflowRetryPolicy | undefined, label: string): void {
	if (policy === undefined) {
		return;
	}
	if (
		!Number.isInteger(policy.maxAttempts) ||
		policy.maxAttempts < 1 ||
		policy.maxAttempts > MAX_WORKFLOW_RETRY_ATTEMPTS
	) {
		throw new Error(
			`${label} maxAttempts must be an integer between 1 and ${MAX_WORKFLOW_RETRY_ATTEMPTS}.`,
		);
	}
	if (policy.initialDelayMs !== undefined) {
		assertNonNegativeFiniteNumber(policy.initialDelayMs, `${label} initialDelayMs`);
		assertAtMost(policy.initialDelayMs, MAX_WORKFLOW_DURATION_MS, `${label} initialDelayMs`);
	}
	if (policy.maxDelayMs !== undefined) {
		assertNonNegativeFiniteNumber(policy.maxDelayMs, `${label} maxDelayMs`);
		assertAtMost(policy.maxDelayMs, MAX_WORKFLOW_DURATION_MS, `${label} maxDelayMs`);
	}
	if (
		policy.retryOn !== undefined &&
		typeof policy.retryOn !== "function" &&
		(!Array.isArray(policy.retryOn) ||
			policy.retryOn.some((matcher) => typeof matcher !== "function"))
	) {
		throw new Error(
			`${label} retryOn must be an error class, a predicate, or an array of them.`,
		);
	}
	if (
		policy.initialDelayMs !== undefined &&
		policy.maxDelayMs !== undefined &&
		policy.maxDelayMs < policy.initialDelayMs
	) {
		throw new Error(`${label} maxDelayMs must be at least initialDelayMs.`);
	}
	if (
		policy.backoffMultiplier !== undefined &&
		(!Number.isFinite(policy.backoffMultiplier) || policy.backoffMultiplier < 1)
	) {
		throw new Error(`${label} backoffMultiplier must be at least one.`);
	}
}

function validateWorkflowStepOptions(options: WorkflowStepOptions | undefined): void {
	if (options?.timeoutMs !== undefined) {
		assertPositiveFiniteNumber(options.timeoutMs, "Workflow step timeoutMs");
	}
	validateWorkflowRetryPolicy(options?.retry, "Workflow step retry policy");
}

function assertAtMost(value: number, max: number, label: string): void {
	if (value > max) {
		throw new Error(`${label} must be at most ${max}.`);
	}
}

function isObject(value: unknown): value is object {
	return (typeof value === "object" && value !== null) || typeof value === "function";
}

function assertPositiveFiniteNumber(value: number, label: string): void {
	if (!Number.isFinite(value) || value <= 0) {
		throw new Error(`${label} must be greater than zero.`);
	}
}

function assertNonNegativeFiniteNumber(value: number, label: string): void {
	if (!Number.isFinite(value) || value < 0) {
		throw new Error(`${label} must be zero or greater.`);
	}
}

function createWorkflowStepKey(segments: readonly unknown[]): string {
	if (segments.length === 0) {
		throw new Error("Workflow step keys must contain at least one segment.");
	}
	for (const segment of segments) {
		if (typeof segment !== "string") {
			throw new Error("Workflow step key segments must be strings.");
		}
		if (segment.length === 0) {
			throw new Error("Workflow step key segments must not be empty.");
		}
	}

	const serialized = JSON.stringify([
		WORKFLOW_STEP_KEY_HASH_DOMAIN,
		WORKFLOW_STEP_KEY_HASH_VERSION,
		...segments,
	]);
	const key = createHash("sha256").update(serialized).digest("base64url");
	if (key.length !== MAX_WORKFLOW_STEP_KEY_LENGTH || !WORKFLOW_STEP_KEY_PATTERN.test(key)) {
		throw new Error(
			`Workflow step keys must be ${MAX_WORKFLOW_STEP_KEY_LENGTH} Base64url characters after hashing.`,
		);
	}

	return key;
}

function normalizeWorkflowStepResult<T>(value: T): WorkflowStepResult<T> {
	return (value === undefined ? null : value) as WorkflowStepResult<T>;
}

async function readWorkflowStepCheckpoint(
	key: string,
): Promise<WorkflowStepCheckpoint | undefined> {
	const directory = process.env[WORKFLOW_STEP_DIRECTORY_ENV_KEY];
	if (!directory) {
		return undefined;
	}

	// The key may contain slashes, private data, or more bytes than a file name
	// allows. Hashing gives both sides a safe path while the key inside the file
	// still proves that this checkpoint belongs to the requested step.
	const path = join(directory, `${hashWorkflowStepKey(key)}.json`);
	let serialized: string;
	try {
		serialized = await readFile(path, "utf8");
	} catch (error) {
		if (isNodeError(error) && error.code === "ENOENT") {
			return undefined;
		}
		throw error;
	}

	const parsed: unknown = JSON.parse(serialized);
	if (
		!isRecord(parsed) ||
		parsed.version !== 0 ||
		parsed.key !== key ||
		!Object.hasOwn(parsed, "value") ||
		typeof parsed.sourceRunId !== "string" ||
		typeof parsed.sourceStepId !== "string"
	) {
		throw new Error(`Workflow step checkpoint ${hashWorkflowStepKey(key)} is invalid.`);
	}

	return {
		version: 0,
		key,
		value: parsed.value,
		sourceRunId: parsed.sourceRunId,
		sourceStepId: parsed.sourceStepId,
	};
}

function hashWorkflowStepKey(key: string): string {
	return createHash("sha256").update(key).digest("hex");
}

function isRecord(value: unknown): value is Record<string, unknown> {
	return typeof value === "object" && value !== null && !Array.isArray(value);
}

function isNodeError(error: unknown): error is NodeJS.ErrnoException {
	return error instanceof Error && "code" in error;
}