import { createHash } from "node:crypto";
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";

import { afterEach, beforeEach, describe, expect, expectTypeOf, it, vi, type Mock } from "vitest";

import { j } from "./schema-builder.js";
import { ExecutionError, FatalError, RateLimitError, RetryableError } from "./error.js";
import type { RunMetadata } from "./runtime-metadata.js";
import {
	events,
	type NotionPageCreatedEvent,
	type NotionPageUpdatedEvent,
	type SlackMessageEvent,
	type WebhookEvent,
} from "./events.generated.js";
import { customAgent } from "./notion-as-code/custom-agent.js";
import { database } from "./notion-as-code/database.js";
import { page } from "./notion-as-code/page.js";
import {
	access,
	bindWorkflowKey,
	connections,
	input,
	workflow,
	type StepContext,
	type WebhookVerifyHandler,
	type WorkflowContext,
	type WorkflowInputDate,
	type WorkflowRetryPolicy,
	type WorkflowWaitDuration,
} from "./workflow.js";

let stdoutSpy: Mock<typeof process.stdout.write>;
let checkpointDirectory: string | undefined;
let metadataDirectory: string | undefined;

beforeEach(async () => {
	stdoutSpy = vi.spyOn(process.stdout, "write").mockImplementation(() => true);
	delete process.env.NOTION_ACCESS_BINDINGS;
	delete process.env.NOTION_WORKFLOW_STEP_DIRECTORY;
	process.env.NOTION_WORKFLOW_CAPABILITY_KEY = "workflow-key";
	checkpointDirectory = undefined;
	metadataDirectory = await mkdtemp(join(tmpdir(), "notion-workflow-metadata-"));
	await writeRunMetadata();
});

afterEach(async () => {
	stdoutSpy.mockRestore();
	delete process.env.NOTION_ACCESS_BINDINGS;
	delete process.env.NOTION_WORKFLOW_STEP_DIRECTORY;
	delete process.env.NOTION_WORKFLOW_CAPABILITY_KEY;
	delete process.env.NOTION_INPUT_PAYLOAD_PATH;
	delete process.env.NOTION_RUN_METADATA_PATH;
	delete process.env.NOTION_API_BASE_URL;
	delete process.env.NOTION_API_TOKEN;
	vi.unstubAllGlobals();
	if (checkpointDirectory) {
		await rm(checkpointDirectory, { recursive: true, force: true });
	}
	if (metadataDirectory) {
		await rm(metadataDirectory, { recursive: true, force: true });
	}
});

const event = { type: "notion.page.created" } as NotionPageCreatedEvent;

describe("manual workflow triggers", () => {
	it("runs a manual trigger without arguments and rejects undeclared input", async () => {
		const handler = vi.fn();
		const definition = workflow({
			name: "No inputs",
			description: "Runs without form fields",
			triggers: [events.manual()],
			handler: async (event) => {
				handler(event.input);
			},
		});
		await definition.handler({ type: "workflow.manual", input: {} });
		expect(handler).toHaveBeenCalledWith({});
		await expect(
			definition.handler(JSON.parse('{"type":"workflow.manual","input":{"extra":true}}')),
		).rejects.toBeInstanceOf(ExecutionError);
		expect(handler).toHaveBeenCalledTimes(1);
	});

	it("infers and validates the manual event input", async () => {
		const handler = vi.fn();
		const definition = workflow({
			name: "Publish release",
			description: "Publishes a release from manually entered values.",
			triggers: [
				events.manual({
					inputSchema: j.object({ version: j.string(), dryRun: j.boolean() }),
				}),
			],
			handler: async (manualEvent) => {
				handler(manualEvent.input);
			},
		});

		await definition.handler({
			type: "workflow.manual",
			input: { version: "1.2.3", dryRun: true },
		});

		expect(handler).toHaveBeenCalledWith({
			version: "1.2.3",
			dryRun: true,
		});
	});

	it("rejects invalid manual input before invoking the handler", async () => {
		const handler = vi.fn();
		const definition = workflow({
			name: "Publish release",
			description: "Publishes a release from manually entered values.",
			triggers: [
				events.manual({
					inputSchema: j.object({ version: j.string() }),
				}),
			],
			handler,
		});

		await expect(
			definition.handler(JSON.parse('{"type":"workflow.manual","input":{"version":42}}')),
		).rejects.toBeInstanceOf(ExecutionError);

		expect(handler).not.toHaveBeenCalled();
	});
});

describe("workflow output schemas", () => {
	const outputSchema = j.object({ url: j.string(), publishedAt: j.datetime().nullable() });

	it("records the output JSON Schema in the workflow config", () => {
		const definition = workflow({
			name: "Publish release",
			description: "Publishes a release",
			triggers: [events.manual()],
			outputSchema,
			handler: () => ({ url: "https://example.com", publishedAt: null }),
		});

		expect(definition.config.outputJsonSchema).toEqual({
			type: "object",
			required: ["url", "publishedAt"],
			additionalProperties: false,
			properties: {
				url: { type: "string" },
				publishedAt: { anyOf: [{ type: "string", format: "date-time" }, { type: "null" }] },
			},
		});
	});

	it("omits the output schema and output when none is declared", async () => {
		const definition = workflow({
			name: "Log pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler: () => {},
		});

		expect(definition.config).not.toHaveProperty("outputJsonSchema");
		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
	});

	it("returns the output in the success envelope", async () => {
		const definition = workflow({
			name: "Publish release",
			description: "Publishes a release",
			triggers: [events.manual()],
			outputSchema,
			handler: async () => ({
				url: "https://example.com",
				publishedAt: "2026-09-28T00:00:00.000Z",
			}),
		});

		await definition.handler({ type: "workflow.manual", input: {} });

		expect(stdoutSpy).toHaveBeenLastCalledWith(
			`\n<__notion_output__>${JSON.stringify({
				_tag: "success",
				value: {
					status: "success",
					output: { url: "https://example.com", publishedAt: "2026-09-28T00:00:00.000Z" },
				},
			})}</__notion_output__>\n`,
		);
	});

	it.each([
		{ name: "a missing value", output: undefined },
		{ name: "a wrong type", output: { url: 1, publishedAt: null } },
		{ name: "an extra key", output: { url: "https://example.com", publishedAt: null, id: 1 } },
		{
			name: "more than 256 KB of JSON",
			output: { url: "x".repeat(256 * 1024), publishedAt: null },
		},
	])("fails without retrying when the handler returns $name", async ({ output }) => {
		const definition = workflow({
			name: "Publish release",
			description: "Publishes a release",
			triggers: [events.manual()],
			outputSchema,
			// SAFETY: Simulates untyped handler code returning a value outside the schema.
			handler: () => output as never,
		});

		await expect(
			definition.handler({ type: "workflow.manual", input: {} }),
		).rejects.toMatchObject({
			cause: expect.objectContaining({
				name: "FatalError",
				message: expect.stringContaining("Invalid workflow output"),
			}),
		});
		expect(String(stdoutSpy.mock.lastCall?.[0])).toContain('"_tag":"terminal"');
	});

	it("rejects an output schema that is not an object schema", () => {
		expect(() =>
			workflow({
				name: "Publish release",
				description: "Publishes a release",
				triggers: [events.manual()],
				// SAFETY: Simulates untyped app code passing a non-object schema.
				outputSchema: j.string() as never,
				handler: (() => {}) as never,
			}),
		).toThrow("workflow outputSchema: must be a j.object() schema");
	});
});

async function createWorkflowContext(
	retry?: WorkflowRetryPolicy,
): Promise<WorkflowContext<Record<never, never>>> {
	let context: WorkflowContext<Record<never, never>> | undefined;
	const definition = workflow({
		name: "Test workflow",
		description: "Captures its workflow context",
		triggers: [events.notionPageCreated()],
		...(retry === undefined ? {} : { retry }),
		handler: (_event, workflowContext) => {
			context = workflowContext;
		},
	});

	await definition.handler(event, { concreteOutput: true });

	if (!context) {
		throw new Error("Workflow context was not created");
	}

	return context;
}

function emittedStepEvents(): Array<{
	event: string;
	value: Record<string, unknown>;
}> {
	return stdoutSpy.mock.calls.map(([output]) => {
		const match = output
			.toString()
			.match(
				/^\n<__notion_step_(started|success|failure|replayed|completed)__>(.*)<\/__notion_step_\1__>\n$/,
			);

		if (!match) {
			throw new Error(`Unexpected step event: ${output.toString()}`);
		}

		return {
			event: match[1]!,
			value: JSON.parse(match[2]!) as Record<string, unknown>,
		};
	});
}

describe("step", () => {
	it("exposes workflow state only while a step callback is running", async () => {
		const context = await createWorkflowContext();
		let escapedState: StepContext["state"] | undefined;

		await context.step("Capture state", ({ state }) => {
			escapedState = state;
		});

		await expect(escapedState?.get("cursor")).rejects.toThrow(
			"state.get() must be called inside context.step().",
		);
	});

	it("reads and writes durable workflow state inside a step", async () => {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi
			.fn<typeof fetch>()
			.mockResolvedValueOnce(
				new Response(JSON.stringify({ exists: true, value: "cursor-1" }), {
					status: 200,
					headers: { "content-type": "application/json" },
				}),
			)
			.mockResolvedValueOnce(
				new Response(JSON.stringify({}), {
					status: 200,
					headers: { "content-type": "application/json" },
				}),
			);
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();

		const cursor = await context.step("Update cursor", async ({ state }) => {
			const current = await state.get("cursor");
			await state.set("cursor", "cursor-2");
			return current;
		});

		expect(cursor).toBe("cursor-1");
		expect(fetchMock).toHaveBeenNthCalledWith(
			1,
			"https://api.example.com/api/v3/workersWorkflowState",
			expect.objectContaining({
				method: "POST",
				headers: expect.objectContaining({
					authorization: "Bearer workflow-token",
				}),
				body: JSON.stringify({
					operation: "get",
					key: "cursor",
					workerId: "worker-id",
					capabilityKey: "workflow-key",
				}),
			}),
		);
		expect(fetchMock).toHaveBeenNthCalledWith(
			2,
			"https://api.example.com/api/v3/workersWorkflowState",
			expect.objectContaining({
				method: "POST",
				body: expect.stringMatching(
					/^\{"operation":"commit","operationId":"run-group-id:[^"]+","mutations":\[\{"operation":"set","key":"cursor","value":"cursor-2"\}\],"workerId":"worker-id","capabilityKey":"workflow-key"\}$/,
				),
			}),
		);
	});

	it("discards buffered state when the step callback throws", async () => {
		const fetchMock = vi.fn<typeof fetch>();
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();

		await expect(
			context.step("Do something", ({ state }) => {
				state.set("foo", 42);
				throw new Error("crash");
			}),
		).rejects.toThrow("crash");
		expect(fetchMock).not.toHaveBeenCalled();
	});

	it("isolates buffered state between step retry attempts", async () => {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi.fn<typeof fetch>().mockImplementation(async (_input, init) => {
			const body = JSON.parse(String(init?.body)) as { operation: string };
			return new Response(
				JSON.stringify(body.operation === "get" ? { exists: false } : { success: true }),
				{ status: 200, headers: { "content-type": "application/json" } },
			);
		});
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();
		const seen: unknown[] = [];
		let attempts = 0;

		await context.step("Retry with state", { retry: { maxAttempts: 2 } }, async ({ state }) => {
			attempts += 1;
			seen.push(await state.get("failed-attempt"));
			if (attempts === 1) {
				state.set("failed-attempt", true);
				throw new RetryableError("try again");
			}
			state.set("succeeded", true);
		});

		expect(seen).toEqual([undefined, undefined]);
		const commits = fetchMock.mock.calls
			.map(([, init]) => String(init?.body))
			.filter((body) => body.includes('"operation":"commit"'));
		expect(commits).toHaveLength(1);
		expect(commits[0]).toContain(
			'"mutations":[{"operation":"set","key":"succeeded","value":true}]',
		);
	});

	it("reads buffered state before the step commits", async () => {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi.fn<typeof fetch>().mockResolvedValue(
			new Response(JSON.stringify({ success: true }), {
				status: 200,
				headers: { "content-type": "application/json" },
			}),
		);
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();

		const value = await context.step("Write then read", async ({ state }) => {
			state.set("foo", 42);
			return state.get("foo");
		});

		expect(value).toBe(42);
		expect(fetchMock).toHaveBeenCalledOnce();
	});

	it("coalesces repeated mutations of a key within the step transaction", async () => {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi.fn<typeof fetch>().mockResolvedValue(
			new Response(JSON.stringify({ success: true }), {
				status: 200,
				headers: { "content-type": "application/json" },
			}),
		);
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();

		const value = await context.step("Mutate repeatedly", async ({ state }) => {
			state.set("cursor", "first");
			state.set("cursor", undefined);
			state.set("cursor", "final");
			return state.get("cursor");
		});

		expect(value).toBe("final");
		expect(fetchMock).toHaveBeenCalledOnce();
		expect(fetchMock).toHaveBeenCalledWith(
			"https://api.example.com/api/v3/workersWorkflowState",
			expect.objectContaining({
				body: expect.stringContaining(
					'"mutations":[{"operation":"set","key":"cursor","value":"final"}]',
				),
			}),
		);
	});

	it("deletes a key by setting it to undefined", async () => {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi.fn<typeof fetch>().mockResolvedValue(
			new Response(JSON.stringify({ success: true }), {
				status: 200,
				headers: { "content-type": "application/json" },
			}),
		);
		vi.stubGlobal("fetch", fetchMock);
		const context = await createWorkflowContext();

		const isDeleted = await context.step("Delete state", async ({ state }) => {
			state.set("foo", 42);
			state.set("foo", undefined);
			return (await state.get("foo")) === undefined;
		});

		expect(isDeleted).toBe(true);
		expect(fetchMock).toHaveBeenCalledWith(
			"https://api.example.com/api/v3/workersWorkflowState",
			expect.objectContaining({
				body: expect.stringContaining('"mutations":[{"operation":"delete","key":"foo"}]'),
			}),
		);
	});

	it("provides required run metadata in the workflow context", async () => {
		const context = await createWorkflowContext();

		expect(context).toMatchObject({
			workspaceId: "workspace-id",
			workerId: "worker-id",
			runGroupId: "run-group-id",
			runId: "run-id",
			capabilityType: "workflow",
			attemptNumber: 2,
		});
		expectTypeOf(context.workspaceId).toEqualTypeOf<string>();
		expectTypeOf(context.workerId).toEqualTypeOf<string>();
		expectTypeOf(context.runGroupId).toEqualTypeOf<string>();
		expectTypeOf(context.runId).toEqualTypeOf<string>();
		expectTypeOf(context.capabilityType).toEqualTypeOf<"workflow">();
		expectTypeOf(context.attemptNumber).toEqualTypeOf<number>();
	});

	it("runs a synchronous callback and writes success events in order", async () => {
		const { step } = await createWorkflowContext();
		let context: StepContext | undefined;
		const resultPromise = step("Fetch page", (stepContext) => {
			context = stepContext;
			return { pageId: "page-1" } as const;
		});

		expectTypeOf(resultPromise).toEqualTypeOf<Promise<{ readonly pageId: "page-1" }>>();
		await expect(resultPromise).resolves.toEqual({ pageId: "page-1" });

		expect(context).toBeDefined();
		const { id } = context!;
		expect(id).toBe(`run-group-id:${workflowStepKey(["Fetch page"])}`);
		expect(emittedStepEvents()).toEqual([
			{
				event: "started",
				value: {
					id,
					name: "Fetch page",
					key: workflowStepKey(["Fetch page"]),
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
			{
				event: "success",
				value: {
					id,
					name: "Fetch page",
					key: workflowStepKey(["Fetch page"]),
					value: { pageId: "page-1" },
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
			{
				event: "completed",
				value: {
					id,
					name: "Fetch page",
					key: workflowStepKey(["Fetch page"]),
					type: "success",
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
		]);
	});

	it("uses stable, distinct ids for steps", async () => {
		const firstContext = await createWorkflowContext();
		const firstAttemptIds: string[] = [];

		await firstContext.step("First", ({ id }) => {
			firstAttemptIds.push(id);
		});
		await firstContext.step("Second", ({ id }) => {
			firstAttemptIds.push(id);
		});

		await writeRunMetadata({ runId: "retry-run-id", attemptNumber: 3 });
		const secondContext = await createWorkflowContext();
		let retryId = "";
		await secondContext.step("First", ({ id }) => {
			retryId = id;
		});

		await writeRunMetadata({
			runGroupId: "other-run-group-id",
			runId: "other-run-id",
			attemptNumber: 1,
		});
		const otherRunContext = await createWorkflowContext();
		let otherRunId = "";
		await otherRunContext.step("First", ({ id }) => {
			otherRunId = id;
		});

		expect(firstAttemptIds).toEqual([
			`run-group-id:${workflowStepKey(["First"])}`,
			`run-group-id:${workflowStepKey(["Second"])}`,
		]);
		expect(retryId).toBe(firstAttemptIds[0]);
		expect(otherRunId).toBe(`other-run-group-id:${workflowStepKey(["First"])}`);
	});

	it("uses an explicit key instead of the display name", async () => {
		const { step } = await createWorkflowContext();
		let id = "";

		await step("Fetch the customer page", { key: ["page", "customer"] }, (context) => {
			id = context.id;
			return "done";
		});

		const key = workflowStepKey(["page", "customer"]);
		expect(key).toHaveLength(43);
		expect(id).toBe(`run-group-id:${key}`);
		expect(emittedStepEvents()[0]).toEqual({
			event: "started",
			value: {
				id,
				name: "Fetch the customer page",
				key,
				timestampMs: expect.any(Number),
				durationMs: expect.any(Number),
			},
		});
	});

	it("wraps an explicit string key in one key segment", async () => {
		const { step } = await createWorkflowContext();
		let id = "";

		await step("Fetch the customer page", { key: "customer-page" }, (context) => {
			id = context.id;
			return "done";
		});

		const key = workflowStepKey(["customer-page"]);
		expect(id).toBe(`run-group-id:${key}`);
		expect(emittedStepEvents()[0]?.value.key).toBe(key);
	});

	it("normalizes implicit and explicit undefined results to null", async () => {
		const { step } = await createWorkflowContext();

		const implicitResult = step("Implicit void", async () => {});
		expectTypeOf(implicitResult).toEqualTypeOf<Promise<null>>();
		await expect(implicitResult).resolves.toBeNull();

		const explicitResult = step("Explicit undefined", () => undefined);
		expectTypeOf(explicitResult).toEqualTypeOf<Promise<null>>();
		await expect(explicitResult).resolves.toBeNull();

		const successValues = emittedStepEvents()
			.filter(({ event }) => event === "success")
			.map(({ value }) => value.value);
		expect(successValues).toEqual([null, null]);
	});

	it("timestamps and measures each event when it is emitted", async () => {
		const { step } = await createWorkflowContext();
		const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_721_234_567_890);
		const performanceNowSpy = vi
			.spyOn(performance, "now")
			.mockReturnValueOnce(101)
			.mockReturnValueOnce(102)
			.mockReturnValueOnce(103);

		try {
			await step("Timestamped", () => "done");
		} finally {
			nowSpy.mockRestore();
			performanceNowSpy.mockRestore();
		}

		expect(emittedStepEvents().map(({ value }) => value.timestampMs)).toEqual([
			performance.timeOrigin + 101,
			performance.timeOrigin + 102,
			performance.timeOrigin + 103,
		]);
		expect(emittedStepEvents().map(({ value }) => value.durationMs)).toEqual([0, 1, 2]);
	});

	it("writes failure events in order and rethrows the original error", async () => {
		const { step } = await createWorkflowContext();
		const originalError = new Error("boom");
		let id = "";
		const fn = vi.fn((context: StepContext) => {
			id = context.id;
			throw originalError;
		});

		await expect(step("Explode", { retry: { maxAttempts: 1 } }, fn)).rejects.toBe(
			originalError,
		);

		expect(fn).toHaveBeenCalledOnce();
		expect(stdoutSpy).toHaveBeenCalledTimes(3);
		expect(emittedStepEvents()).toEqual([
			{
				event: "started",
				value: {
					id,
					name: "Explode",
					key: workflowStepKey(["Explode"]),
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
			{
				event: "failure",
				value: {
					id,
					name: "Explode",
					key: workflowStepKey(["Explode"]),
					error: {
						name: "ExecutionError",
						message: "Error during app execution: Error: boom",
						trace: expect.any(String),
					},
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
			{
				event: "completed",
				value: {
					id,
					name: "Explode",
					key: workflowStepKey(["Explode"]),
					type: "failure",
					timestampMs: expect.any(Number),
					durationMs: expect.any(Number),
				},
			},
		]);
	});

	it("replays a saved step without calling its function", async () => {
		const { step } = await createWorkflowContext();
		const directory = await createCheckpointDirectory();
		await writeFile(
			checkpointPath(directory, workflowStepKey(["Fetch page"])),
			JSON.stringify({
				version: 0,
				key: workflowStepKey(["Fetch page"]),
				value: { pageId: "saved-page" },
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		);
		const fn = vi.fn(() => ({ pageId: "new-page" }));

		await expect(step("Fetch page", fn)).resolves.toEqual({
			pageId: "saved-page",
		});

		expect(fn).not.toHaveBeenCalled();
		const outputs = stdoutSpy.mock.calls.map(([output]) => String(output));
		expect(outputs).toHaveLength(3);
		expect(outputs[0]).toContain("<__notion_step_started__>");
		expect(outputs[0]).toContain(`"key":"${workflowStepKey(["Fetch page"])}"`);
		expect(outputs[1]).toContain("<__notion_step_replayed__>");
		expect(outputs[1]).toContain('"sourceRunId":"source-run"');
		expect(outputs[1]).toContain('"sourceStepId":"source-step"');
		expect(outputs[2]).toContain("<__notion_step_completed__>");
		expect(outputs[2]).toContain('"type":"success"');
	});

	it("replays a normalized void result as null", async () => {
		const { step } = await createWorkflowContext();
		const directory = await createCheckpointDirectory();
		await writeFile(
			checkpointPath(directory, workflowStepKey(["Side effect"])),
			JSON.stringify({
				version: 0,
				key: workflowStepKey(["Side effect"]),
				value: null,
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		);
		const fn = vi.fn(() => undefined);

		const result = step("Side effect", fn);
		expectTypeOf(result).toEqualTypeOf<Promise<null>>();
		await expect(result).resolves.toBeNull();
		expect(fn).not.toHaveBeenCalled();
	});

	it("runs a step when its checkpoint file is missing", async () => {
		await createCheckpointDirectory();
		const { step } = await createWorkflowContext();
		const fn = vi.fn(() => "new-value");

		await expect(step("Run me", fn)).resolves.toBe("new-value");

		expect(fn).toHaveBeenCalledOnce();
		expect(stdoutSpy.mock.calls.map(([output]) => String(output))).not.toEqual(
			expect.arrayContaining([expect.stringContaining("__notion_step_replayed__")]),
		);
	});

	it("mixes replayed steps with new work", async () => {
		const directory = await createCheckpointDirectory();
		await writeFile(
			checkpointPath(directory, workflowStepKey(["First"])),
			JSON.stringify({
				version: 0,
				key: workflowStepKey(["First"]),
				value: "saved",
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		);
		const { step } = await createWorkflowContext();
		const replayedFn = vi.fn(() => "new-first");
		const newFn = vi.fn(() => "new-second");

		await expect(step("First", replayedFn)).resolves.toBe("saved");
		await expect(step("Second", newFn)).resolves.toBe("new-second");

		expect(replayedFn).not.toHaveBeenCalled();
		expect(newFn).toHaveBeenCalledOnce();
		const outputs = stdoutSpy.mock.calls.map(([output]) => String(output));
		expect(
			outputs.filter((output) => output.includes("__notion_step_replayed__")),
		).toHaveLength(1);
		expect(outputs).toEqual(
			expect.arrayContaining([
				expect.stringContaining(
					`"name":"Second","key":"${workflowStepKey(["Second"])}","value":"new-second"`,
				),
			]),
		);
	});

	it.each([
		{
			name: "invalid JSON",
			contents: "{",
		},
		{
			name: "an invalid shape",
			contents: "[]",
		},
		{
			name: "wrong version",
			contents: JSON.stringify({
				version: 1,
				key: workflowStepKey(["Do not rerun"]),
				value: "saved",
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		},
		{
			name: "wrong key",
			contents: JSON.stringify({
				version: 0,
				key: "other",
				value: "saved",
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		},
		{
			name: "a missing value",
			contents: JSON.stringify({
				version: 0,
				key: workflowStepKey(["Do not rerun"]),
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		},
		{
			name: "missing source ids",
			contents: JSON.stringify({
				version: 0,
				key: workflowStepKey(["Do not rerun"]),
				value: "saved",
			}),
		},
	])("fails instead of rerunning a step for $name", async ({ contents }) => {
		const directory = await createCheckpointDirectory();
		await writeFile(checkpointPath(directory, workflowStepKey(["Do not rerun"])), contents);
		const { step } = await createWorkflowContext();
		const fn = vi.fn(() => "new-value");

		await expect(step("Do not rerun", fn)).rejects.toBeInstanceOf(Error);

		expect(fn).not.toHaveBeenCalled();
		expect(stdoutSpy.mock.calls.map(([output]) => String(output))).toEqual(
			expect.arrayContaining([
				expect.stringContaining("<__notion_step_failure__>"),
				expect.stringContaining('"type":"failure"'),
			]),
		);
	});

	it("fails instead of rerunning a step when its checkpoint cannot be read", async () => {
		const directory = await createCheckpointDirectory();
		await mkdir(checkpointPath(directory, workflowStepKey(["Do not rerun"])));
		const { step } = await createWorkflowContext();
		const fn = vi.fn(() => "new-value");

		await expect(step("Do not rerun", fn)).rejects.toBeInstanceOf(Error);

		expect(fn).not.toHaveBeenCalled();
		expect(stdoutSpy.mock.calls.map(([output]) => String(output))).toEqual(
			expect.arrayContaining([
				expect.stringContaining("<__notion_step_failure__>"),
				expect.stringContaining('"type":"failure"'),
			]),
		);
	});

	it("uses names as keys for overlapping steps regardless of completion order", async () => {
		const { step } = await createWorkflowContext();
		let finishFirst: (() => void) | undefined;
		const result = Promise.all([
			step(
				"First",
				() =>
					new Promise<string>((resolve) => {
						finishFirst = () => resolve("first");
					}),
			),
			step("Second", () => "second"),
		]);

		await Promise.resolve();
		finishFirst?.();
		await expect(result).resolves.toEqual(["first", "second"]);

		const started = stdoutSpy.mock.calls
			.map(([output]) => String(output))
			.filter((output) => output.includes("<__notion_step_started__>"));
		expect(started[0]).toContain(`"name":"First","key":"${workflowStepKey(["First"])}"`);
		expect(started[1]).toContain(`"name":"Second","key":"${workflowStepKey(["Second"])}"`);
	});

	it("rejects duplicate default and explicit keys", async () => {
		const { step } = await createWorkflowContext();
		const duplicateDefaultFn = vi.fn(() => "duplicate");
		const duplicateExplicitFn = vi.fn(() => "duplicate");

		await expect(step("Duplicate", () => "first")).resolves.toBe("first");
		await expect(step("Duplicate", duplicateDefaultFn)).rejects.toThrow(
			'Workflow step "Duplicate" has the same key as "Duplicate"',
		);
		await expect(
			step("Different name", { key: "Duplicate" }, duplicateExplicitFn),
		).rejects.toThrow('Workflow step "Different name" has the same key as "Duplicate"');

		expect(duplicateDefaultFn).not.toHaveBeenCalled();
		expect(duplicateExplicitFn).not.toHaveBeenCalled();
	});

	it("rejects empty explicit keys", async () => {
		const { step } = await createWorkflowContext();
		const fn = vi.fn(() => "value");

		await expect(step("Display name", { key: [] }, fn)).rejects.toThrow(
			"Workflow step keys must contain at least one segment.",
		);

		expect(fn).not.toHaveBeenCalled();
		expect(stdoutSpy).not.toHaveBeenCalled();
	});

	it("rejects empty explicit key segments", async () => {
		const { step } = await createWorkflowContext();
		const fn = vi.fn(() => "value");

		await expect(step("Display name", { key: [""] }, fn)).rejects.toThrow(
			"Workflow step key segments must not be empty.",
		);

		expect(fn).not.toHaveBeenCalled();
		expect(stdoutSpy).not.toHaveBeenCalled();
	});

	it("leaves step names unchanged for runtime validation", async () => {
		const { step } = await createWorkflowContext();
		const name = "Résumé\u0000\t日本語\n👨‍👩‍👧‍👦\u200b";

		await expect(step(name, { key: "raw-name" }, () => "value")).resolves.toBe("value");

		expect(emittedStepEvents()[0]?.value).toEqual(
			expect.objectContaining({
				name,
				key: workflowStepKey(["raw-name"]),
			}),
		);
	});

	it("allows steps to invoke other uniquely named steps", async () => {
		const { step } = await createWorkflowContext();

		await expect(
			step("Parent", async () => {
				return step("Child", () => "child-value");
			}),
		).resolves.toBe("child-value");

		expect(emittedStepEvents().map(({ event, value }) => [event, value.key])).toEqual([
			["started", workflowStepKey(["Parent"])],
			["started", workflowStepKey(["Child"])],
			["success", workflowStepKey(["Child"])],
			["completed", workflowStepKey(["Child"])],
			["success", workflowStepKey(["Parent"])],
			["completed", workflowStepKey(["Parent"])],
		]);
	});

	it("keeps nested and later step keys stable when a parent is replayed", async () => {
		const directory = await createCheckpointDirectory();
		for (const [key, value] of [
			["Parent", "saved-parent"],
			["Child", "saved-child"],
			["Sibling", "saved-sibling"],
		] as const) {
			const wireKey = workflowStepKey([key]);
			await writeFile(
				checkpointPath(directory, wireKey),
				JSON.stringify({
					version: 0,
					key: wireKey,
					value,
					sourceRunId: "source-run",
					sourceStepId: `source-${key.toLowerCase()}`,
				}),
			);
		}
		const { step } = await createWorkflowContext();
		const childFn = vi.fn(() => "new-child");
		const siblingFn = vi.fn(() => "new-sibling");

		await expect(
			step("Parent", async () => {
				await step("Child", childFn);
				return "new-parent";
			}),
		).resolves.toBe("saved-parent");
		await expect(step("Sibling", siblingFn)).resolves.toBe("saved-sibling");

		expect(childFn).not.toHaveBeenCalled();
		expect(siblingFn).not.toHaveBeenCalled();
		expect(
			emittedStepEvents()
				.filter(({ event }) => event === "started")
				.map(({ value }) => value.key),
		).toEqual([workflowStepKey(["Parent"]), workflowStepKey(["Sibling"])]);
	});

	it("resets duplicate key tracking for each workflow call", async () => {
		const keys: string[] = [];
		const definition = workflow({
			name: "Reuse step key",
			description: "Checks that a key can be reused in a new invocation",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step("First", () => undefined);
				const started = stdoutSpy.mock.calls
					.map(([output]) => String(output))
					.findLast((output) => output.includes("<__notion_step_started__>"));
				const match = started?.match(/"key":"([^"]+)"/);
				if (match?.[1]) {
					keys.push(match[1]);
				}
			},
		});

		await definition.handler(event, { concreteOutput: true });
		await definition.handler(event, { concreteOutput: true });

		expect(keys).toEqual([workflowStepKey(["First"]), workflowStepKey(["First"])]);
	});

	it("retries a failed callback using the step policy", async () => {
		const { step } = await createWorkflowContext();
		const callback = vi
			.fn<() => string>()
			.mockRejectedValueOnce(new Error("temporary"))
			.mockReturnValue("done");

		await expect(
			step("Retry request", { retry: { maxAttempts: 2, initialDelayMs: 0 } }, callback),
		).resolves.toBe("done");
		expect(callback).toHaveBeenCalledTimes(2);
	});

	it("uses the workflow retry policy for steps without their own", async () => {
		const { step } = await createWorkflowContext({ maxAttempts: 2, initialDelayMs: 0 });
		const callback = vi.fn<() => string>().mockRejectedValue(new Error("down"));

		await expect(step("Inherit policy", callback)).rejects.toThrow("down");
		expect(callback).toHaveBeenCalledTimes(2);
	});

	it("replaces the workflow retry policy with the step policy", async () => {
		const { step } = await createWorkflowContext({ maxAttempts: 3, initialDelayMs: 0 });
		const callback = vi.fn<() => string>().mockRejectedValue(new Error("down"));

		await expect(
			step("Override policy", { retry: { maxAttempts: 1 } }, callback),
		).rejects.toThrow("down");
		expect(callback).toHaveBeenCalledOnce();
	});

	it("makes three attempts with backoff when no policy is configured", async () => {
		vi.useFakeTimers();
		try {
			const { step } = await createWorkflowContext();
			const callback = vi.fn<() => string>().mockRejectedValue(new Error("down"));

			const result = step("Default policy", callback);
			const assertion = expect(result).rejects.toThrow("down");
			await vi.advanceTimersByTimeAsync(3_000);
			await assertion;
			expect(callback).toHaveBeenCalledTimes(3);
		} finally {
			vi.useRealTimers();
		}
	});

	it("retries only errors matched by retryOn", async () => {
		class TransientError extends Error {}
		const { step } = await createWorkflowContext({
			maxAttempts: 3,
			initialDelayMs: 0,
			retryOn: [TransientError, (error) => error instanceof Error && error.message === "429"],
		});

		const transient = vi.fn<() => string>().mockRejectedValue(new TransientError("blip"));
		await expect(step("Class match", transient)).rejects.toThrow("blip");
		expect(transient).toHaveBeenCalledTimes(3);

		const predicate = vi.fn<() => string>().mockRejectedValue(new Error("429"));
		await expect(step("Predicate match", predicate)).rejects.toThrow("429");
		expect(predicate).toHaveBeenCalledTimes(3);

		const other = vi.fn<() => string>().mockRejectedValue(new TypeError("bug"));
		await expect(step("No match", other)).rejects.toThrow("bug");
		expect(other).toHaveBeenCalledOnce();

		const explicit = vi.fn<() => string>().mockRejectedValue(new RetryableError("again"));
		await expect(step("Explicit retryable", explicit)).rejects.toThrow("again");
		expect(explicit).toHaveBeenCalledTimes(3);
	});

	it("accepts a single retryOn matcher", async () => {
		class TransientError extends Error {}
		const { step } = await createWorkflowContext({
			maxAttempts: 2,
			initialDelayMs: 0,
			retryOn: TransientError,
		});

		const transient = vi.fn<() => string>().mockRejectedValue(new TransientError("blip"));
		await expect(step("Class match", transient)).rejects.toThrow("blip");
		expect(transient).toHaveBeenCalledTimes(2);

		const other = vi.fn<() => string>().mockRejectedValue(new TypeError("bug"));
		await expect(step("No match", other)).rejects.toThrow("bug");
		expect(other).toHaveBeenCalledOnce();
	});

	it("never retries a FatalError matched by retryOn", async () => {
		const { step } = await createWorkflowContext({
			maxAttempts: 3,
			initialDelayMs: 0,
			retryOn: [Error],
		});
		const callback = vi.fn<() => string>().mockRejectedValue(new FatalError("stop"));

		await expect(step("Fatal", callback)).rejects.toThrow("stop");
		expect(callback).toHaveBeenCalledOnce();
	});

	it("rejects step retry policies that cannot fit in one invocation", async () => {
		const { step } = await createWorkflowContext();
		const callback = vi.fn(() => "done");

		await expect(
			step("Slow retries", { retry: { maxAttempts: 3, initialDelayMs: 200_000 } }, callback),
		).rejects.toThrow(
			'Workflow step "Slow retries" can spend up to 500000ms on retry delays and timeouts, which exceeds the 300000ms limit',
		);
		await expect(
			step("Long timeout", { timeoutMs: 120_000, retry: { maxAttempts: 3 } }, callback),
		).rejects.toThrow('Workflow step "Long timeout" can spend up to 363000ms');
		expect(callback).not.toHaveBeenCalled();
		expect(() =>
			workflow({
				name: "Slow workflow",
				description: "Its retry policy is the default for every step",
				triggers: [events.notionPageCreated()],
				retry: { maxAttempts: 3, initialDelayMs: 200_000 },
				handler: () => {},
			}),
		).toThrow("Workflow retry policy can spend up to 500000ms");
	});

	it("leaves a retry that no longer fits in the invocation to the run", async () => {
		const definition = workflow({
			name: "Deferred retry",
			description: "Hands a long explicit delay to the platform",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step("Wait for service", () => {
					throw new RetryableError("busy", { retryAfterMs: 10 * 60 * 1000 });
				});
			},
		});

		await expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"_tag":"retryable","name":"RetryableError","message":"busy"`,
			),
		);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(`"retryAfterMs":600000`),
		);
	});

	it("counts earlier work in the invocation against the step retry budget", async () => {
		vi.useFakeTimers();
		try {
			const callback = vi.fn(() => {
				throw new RetryableError("busy", { retryAfterMs: 60_000 });
			});
			const definition = workflow({
				name: "Late retry",
				description: "Fails a step near the end of the invocation",
				triggers: [events.notionPageCreated()],
				handler: async (_event, context) => {
					await new Promise((resolve) => setTimeout(resolve, 4.5 * 60 * 1000));
					await context.step("Call service", callback);
				},
			});

			const result = expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
			await vi.runAllTimersAsync();
			await result;
			expect(callback).toHaveBeenCalledOnce();
			expect(stdoutSpy).toHaveBeenLastCalledWith(
				expect.stringContaining(`"retryAfterMs":60000`),
			);
		} finally {
			vi.useRealTimers();
		}
	});

	it("writes a rate limit envelope for a RateLimitError", async () => {
		const definition = workflow({
			name: "Rate limited",
			description: "Reports the service's retry delay",
			triggers: [events.notionPageCreated()],
			handler: () => {
				throw new RateLimitError({ retryAfter: 30 });
			},
		});

		await expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`{"_tag":"error","error":{"_tag":"rate_limit","name":"RateLimitError"`,
			),
		);
		expect(stdoutSpy).toHaveBeenLastCalledWith(expect.stringContaining(`"retryAfter":30`));
	});

	it("waits for the delay requested by a RateLimitError", async () => {
		vi.useFakeTimers();
		try {
			const { step } = await createWorkflowContext();
			const callback = vi
				.fn<() => string>()
				.mockRejectedValueOnce(new RateLimitError({ retryAfter: 30 }))
				.mockReturnValue("done");

			const result = step("Rate limited", { retry: { maxAttempts: 2 } }, callback);
			await vi.advanceTimersByTimeAsync(29_000);
			expect(callback).toHaveBeenCalledOnce();
			await vi.advanceTimersByTimeAsync(1_000);
			await expect(result).resolves.toBe("done");
		} finally {
			vi.useRealTimers();
		}
	});

	it("does not retry a FatalError", async () => {
		const { step } = await createWorkflowContext();
		const error = new FatalError("invalid input");
		const callback = vi.fn(() => {
			throw error;
		});

		await expect(step("Validate input", { retry: { maxAttempts: 3 } }, callback)).rejects.toBe(
			error,
		);
		expect(callback).toHaveBeenCalledOnce();
	});

	it("times out a step callback", async () => {
		const { step } = await createWorkflowContext();

		await expect(
			step("Wait forever", { timeoutMs: 1 }, () => new Promise(() => {})),
		).rejects.toMatchObject({
			name: "StepTimeoutError",
			message: "Workflow step timed out after 1ms.",
		});
	});

	it("aborts a timed-out attempt before retrying it", async () => {
		const { step } = await createWorkflowContext();
		const signals: AbortSignal[] = [];
		const writes: number[] = [];

		const result = await step(
			"Write once",
			{ timeoutMs: 10, retry: { maxAttempts: 2, initialDelayMs: 0 } },
			async ({ signal }) => {
				signals.push(signal);
				const attempt = signals.length;
				if (attempt === 1) {
					await new Promise<void>((resolve, reject) => {
						const timer = setTimeout(resolve, 50);
						signal.addEventListener("abort", () => {
							clearTimeout(timer);
							reject(signal.reason);
						});
					});
				}
				writes.push(attempt);
				return attempt;
			},
		);
		await new Promise((resolve) => setTimeout(resolve, 60));

		expect(result).toBe(2);
		expect(writes).toEqual([2]);
		expect(signals[0]?.aborted).toBe(true);
		expect(signals[0]?.reason).toMatchObject({ name: "StepTimeoutError" });
		expect(signals[1]?.aborted).toBe(false);
	});

	it("writes step events under concreteOutput", async () => {
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step("Always visible", () => "done");
			},
		});

		await definition.handler(event, { concreteOutput: true });

		expect(stdoutSpy).toHaveBeenCalledTimes(3);
		expect(stdoutSpy.mock.calls[0]?.[0]).toEqual(
			expect.stringContaining("<__notion_step_started__>"),
		);
		expect(stdoutSpy.mock.calls[1]?.[0]).toEqual(
			expect.stringContaining("<__notion_step_success__>"),
		);
		expect(stdoutSpy.mock.calls[2]?.[0]).toEqual(expect.stringContaining('"type":"success"'));
	});
});

describe("wait.until", () => {
	it("stops the workflow and returns its absolute resume time", async () => {
		const now = Date.parse("2026-09-17T09:00:00.000Z");
		const resumeAt = new Date(now + 60_000);
		const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(now);
		let continued = false;
		const definition = workflow({
			name: "Wait for launch",
			description: "Waits until launch time",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.until("Launch time", { at: resumeAt });
				continued = true;
			},
		});

		try {
			await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
				status: "waiting",
				wait: {
					type: "until",
					name: "Launch time",
					resumeAtMs: resumeAt.getTime(),
				},
			});
			expect(continued).toBe(false);
		} finally {
			dateNowSpy.mockRestore();
		}
	});

	it("anchors a relative wait in its durable step checkpoint", async () => {
		const directory = await createCheckpointDirectory();
		const initialNow = Date.parse("2026-09-17T09:00:00.000Z");
		const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(initialNow);
		let continued = false;
		const definition = workflow({
			name: "Wait before following up",
			description: "Waits before continuing",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.until("Follow-up delay", {
					after: { minutes: 30, seconds: 15 },
				});
				continued = true;
			},
		});

		try {
			const firstResult = await definition.handler(event, { concreteOutput: true });
			expect(firstResult).toMatchObject({
				status: "waiting",
				wait: { resumeAtMs: initialNow + 1_815_000 },
			});

			const successEvent = emittedStepEvents().find(({ event }) => event === "success");
			const stepKey = successEvent?.value.key;
			const resumeAtMs = successEvent?.value.value;
			if (typeof stepKey !== "string" || typeof resumeAtMs !== "number") {
				throw new Error("Wait checkpoint event was not emitted");
			}
			await writeFile(
				checkpointPath(directory, stepKey),
				JSON.stringify({
					version: 0,
					key: stepKey,
					value: resumeAtMs,
					sourceRunId: "first-run",
					sourceStepId: "wait-step",
				}),
			);

			dateNowSpy.mockReturnValue(initialNow + 7_200_000);
			await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
				status: "success",
			});
			expect(continued).toBe(true);
		} finally {
			dateNowSpy.mockRestore();
		}
	});

	it("uses composite keys to distinguish waits in a loop", async () => {
		const now = Date.parse("2026-09-17T09:00:00.000Z");
		const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(now);
		const itemIds = ["first", "second"];
		const definition = workflow({
			name: "Wait between items",
			description: "Uses one stable display name for repeated waits",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				for (const itemId of itemIds) {
					await context.wait.until("Item delay", {
						key: ["item-delay", itemId],
						at: new Date(now),
					});
				}
			},
		});

		try {
			await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
				status: "success",
			});
			expect(
				emittedStepEvents()
					.filter(({ event }) => event === "success")
					.map(({ value }) => value.key),
			).toEqual([
				workflowStepKey(["wait", "until", "item-delay", "first"]),
				workflowStepKey(["wait", "until", "item-delay", "second"]),
			]);
		} finally {
			dateNowSpy.mockRestore();
		}
	});

	it("rejects waits that can outlive their durable checkpoint", async () => {
		const definition = workflow({
			name: "Wait too long",
			description: "Rejects an unsupported wait",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.until("Beyond retention", {
					after: { days: 7 },
				});
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"Workflow waits must be shorter than 7 days.",
		);
	});

	it.each([
		[{ hours: 0 }, "positive finite integers"],
		[{ minutes: 1.5 }, "positive finite integers"],
		[{}, "at least one unit"],
	])("rejects invalid relative durations", async (after, expectedMessage) => {
		const definition = workflow({
			name: "Wait for a valid duration",
			description: "Rejects an invalid duration",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.until("Invalid duration", {
					after: after as WorkflowWaitDuration,
				});
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			expectedMessage,
		);
	});
});

describe("wait.forInput", () => {
	it("stops the workflow with a typed form schema", async () => {
		let continued = false;
		const definition = workflow({
			name: "Collect launch details",
			description: "Waits for a person to provide launch details",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.forInput("Launch details", {
					input: {
						summary: input.text(),
						priority: input.select(["low", "medium", "high"]),
						notes: input.text({ optional: true }),
						teams: input.multiSelect(["Design", "Engineering"]),
					},
				});
				continued = true;
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "waiting",
			wait: {
				type: "input",
				name: "Launch details",
				stepKey: workflowStepKey(["wait", "input", "Launch details"]),
				input: {
					summary: { type: "text" },
					priority: {
						type: "select",
						options: ["low", "medium", "high"],
					},
					notes: { type: "text", optional: true },
					teams: { type: "multi_select", options: ["Design", "Engineering"] },
				},
			},
		});
		expect(continued).toBe(false);
	});

	it("returns the typed response stored in its durable checkpoint", async () => {
		const directory = await createCheckpointDirectory();
		const stepKey = workflowStepKey(["wait", "input", "Launch details"]);
		await writeFile(
			checkpointPath(directory, stepKey),
			JSON.stringify({
				version: 0,
				key: stepKey,
				value: {
					value: {
						summary: "Ship the new workflow",
						priority: "high",
						labels: ["Design", "Engineering"],
						owners: ["first-user-id", "second-user-id"],
						window: {
							start: "2026-09-22T09:00",
							end: "2026-09-23T17:00",
							timeZone: "America/Los_Angeles",
						},
						attachments: ["https://example.com/plan.pdf", "attachment-block-id"],
					},
					respondedBy: "user-id",
					respondedAt: "2026-09-18T12:00:00.000Z",
				},
				sourceRunId: "input-run",
				sourceStepId: "input-step",
			}),
		);

		let capturedPriority: "low" | "medium" | "high" | undefined;
		let capturedInputs: unknown;
		const definition = workflow({
			name: "Collect launch details",
			description: "Waits for a person to provide launch details",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				const response = await context.wait.forInput("Launch details", {
					input: {
						summary: input.text(),
						priority: input.select(["low", "medium", "high"]),
						labels: input.multiSelect(["Design", "Engineering"]),
						owners: input.person(),
						notes: input.text({ optional: true }),
						window: input.date(),
						attachments: input.files({ optional: true }),
					},
				});
				expectTypeOf(response.value.summary).toEqualTypeOf<string>();
				expectTypeOf(response.value.priority).toEqualTypeOf<"low" | "medium" | "high">();
				capturedPriority = response.value.priority;
				expectTypeOf(response.value.labels).toEqualTypeOf<("Design" | "Engineering")[]>();
				expectTypeOf(response.value.owners).toEqualTypeOf<string[]>();
				expectTypeOf(response.value.window).toEqualTypeOf<WorkflowInputDate>();
				expectTypeOf(response.value.attachments).toEqualTypeOf<string[] | undefined>();
				capturedInputs = response.value;
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(capturedPriority).toBe("high");
		expect(capturedInputs).toEqual({
			summary: "Ship the new workflow",
			priority: "high",
			labels: ["Design", "Engineering"],
			owners: ["first-user-id", "second-user-id"],
			window: {
				start: "2026-09-22T09:00",
				end: "2026-09-23T17:00",
				timeZone: "America/Los_Angeles",
			},
			attachments: ["https://example.com/plan.pdf", "attachment-block-id"],
		});
	});

	it.each([
		{ extra: { admin: true } },
		{ extra: { options: { admin: true } } },
		{ extra: { ["__proto__"]: { admin: true } } },
	])("rejects undeclared response fields: $extra", async ({ extra }) => {
		const directory = await createCheckpointDirectory();
		const stepKey = workflowStepKey(["wait", "input", "Review"]);
		await writeFile(
			checkpointPath(directory, stepKey),
			JSON.stringify({
				version: 0,
				key: stepKey,
				value: {
					value: { priority: "high", ...extra },
					respondedBy: "user-id",
					respondedAt: "2026-09-18T12:00:00.000Z",
				},
				sourceRunId: "input-run",
				sourceStepId: "input-step",
			}),
		);
		const definition = workflow({
			name: "Review",
			description: "Rejects undeclared input",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.forInput("Review", {
					input: {
						priority: input.select(["low", "high"]),
						notes: input.text({ optional: true }),
					},
				});
			},
		});
		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"does not match its schema",
		);
	});

	it.each([
		{ name: "unknown select", field: input.select(["low", "medium", "high"]), value: "urgent" },
		{
			name: "unknown multi-select",
			field: input.multiSelect(["Design", "Engineering"]),
			value: ["Design", "Other"],
		},
		{ name: "scalar multi-select", field: input.multiSelect(["Design"]), value: "Design" },
		{ name: "scalar person", field: input.person(), value: "user-id" },
		{ name: "invalid person array", field: input.person(), value: ["user-id", 123] },
		{ name: "date string", field: input.date(), value: "2026-09-22" },
		{
			name: "invalid date timezone",
			field: input.date(),
			value: { start: "2026-09-22", timeZone: 123 },
		},
	])("rejects a checkpoint with $name", async ({ field, value }) => {
		const directory = await createCheckpointDirectory();
		const stepKey = workflowStepKey(["wait", "input", "Launch details"]);
		await writeFile(
			checkpointPath(directory, stepKey),
			JSON.stringify({
				version: 0,
				key: stepKey,
				value: {
					value: { priority: value },
					respondedBy: "user-id",
					respondedAt: "2026-09-18T12:00:00.000Z",
				},
				sourceRunId: "input-run",
				sourceStepId: "input-step",
			}),
		);

		const definition = workflow({
			name: "Collect launch details",
			description: "Waits for a person to provide launch details",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.forInput("Launch details", {
					input: {
						priority: field,
					},
				});
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"does not match its schema",
		);
	});
});

describe("wait.forEvent", () => {
	const pageUpdated = { type: "notion.page.updated" } as NotionPageUpdatedEvent;
	const slackMessage = { type: "slack.message" } as SlackMessageEvent;
	const now = Date.parse("2026-09-17T09:00:00.000Z");
	const stepKey = workflowStepKey(["wait", "event", "Launch page created"]);

	function launchHandoff(onResume: (created: NotionPageCreatedEvent) => void) {
		return workflow({
			name: "Launch handoff",
			description: "Waits for the launch page to be created",
			triggers: [events.notionPageUpdated()],
			handler: async (_event, context) => {
				const created = await context.wait.forEvent("Launch page created", {
					event: events.notionPageCreated(),
					scope: { dataSourceId: "launches-data-source" },
					timeout: { days: 2 },
				});
				expectTypeOf(created).toEqualTypeOf<NotionPageCreatedEvent>();
				onResume(created);
			},
		});
	}

	async function writeEventCheckpoint(value: unknown): Promise<void> {
		const directory = await createCheckpointDirectory();
		await writeFile(
			checkpointPath(directory, stepKey),
			JSON.stringify({
				version: 0,
				key: stepKey,
				value,
				sourceRunId: "event-run",
				sourceStepId: "event-step",
			}),
		);
	}

	it("pauses with the trigger descriptor, scope, and timeout deadline", async () => {
		const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(now);
		const onResume = vi.fn();

		try {
			await expect(
				launchHandoff(onResume).handler(pageUpdated, { concreteOutput: true }),
			).resolves.toEqual({
				status: "waiting",
				wait: {
					type: "event",
					name: "Launch page created",
					stepKey,
					event: {
						type: "notion.page.created",
						scope: { dataSourceId: "launches-data-source" },
					},
					timeoutAtMs: now + 2 * 86_400_000,
				},
			});
			expect(onResume).not.toHaveBeenCalled();
		} finally {
			dateNowSpy.mockRestore();
		}
	});

	it("resumes with the event stored in its checkpoint", async () => {
		const created = { ...event, pageId: "launch-page-id" };
		await writeEventCheckpoint(created);
		const onResume = vi.fn();

		await expect(
			launchHandoff(onResume).handler(pageUpdated, { concreteOutput: true }),
		).resolves.toEqual({ status: "success" });
		expect(onResume).toHaveBeenCalledWith(created);
	});

	it("fails without retrying when the timeout checkpoint is written", async () => {
		await writeEventCheckpoint(null);
		const onResume = vi.fn();

		await expect(
			launchHandoff(onResume).handler(pageUpdated, { concreteOutput: true }),
		).rejects.toMatchObject({
			cause: expect.objectContaining({
				name: "FatalError",
				message: 'Workflow event wait "Launch page created" timed out.',
			}),
		});
		expect(onResume).not.toHaveBeenCalled();
	});

	it("rejects a checkpoint for a different event type", async () => {
		await writeEventCheckpoint({ type: "notion.page.deleted" });

		await expect(
			launchHandoff(() => {}).handler(pageUpdated, { concreteOutput: true }),
		).rejects.toThrow("does not match event type notion.page.created");
	});

	it("rejects a provider event without its declared connection from JavaScript", async () => {
		const definition = workflow({
			name: "Await calendar event",
			description: "Waits for a calendar event on a Slack connection",
			triggers: [events.notionPageCreated()],
			connections: { support: connections.slack() },
			handler: async (_event, context) => {
				await Reflect.apply(context.wait.forEvent, context.wait, [
					"Calendar event",
					{
						event: events.calendarEventCreated({
							calendars: { connectionKey: "support", target: "meetings" },
						}),
						scope: { calendars: [{ accountId: "account", calendarId: "calendar" }] },
						timeout: { hours: 1 },
					},
				]);
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"requires a declared calendar connection: support",
		);
	});

	it("rejects waiting for an event the workflow also triggers on", async () => {
		const definition = workflow({
			name: "Conflicting wait",
			description: "Triggers on and waits for page creation",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.forEvent("Next page", {
					event: events.notionPageCreated(),
					scope: { dataSourceId: "launches-data-source" },
					timeout: { hours: 1 },
				});
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"which is also a trigger of this workflow",
		);
	});

	it("allows waits on the same event type from a different connection", async () => {
		const definition = workflow({
			name: "Cross-connection wait",
			description: "Triggers on one Slack workspace and waits on another",
			triggers: ({ events }) => [events.slackMessage({ connectionKey: "support" })],
			connections: { support: connections.slack(), sales: connections.slack() },
			handler: async (_event, context) => {
				await context.wait.forEvent("Sales reply", {
					event: events.slackMessage({ connectionKey: "sales" }),
					scope: { channelIds: ["C123"] },
					timeout: { hours: 1 },
				});
			},
		});

		await expect(
			definition.handler(slackMessage, { concreteOutput: true }),
		).resolves.toMatchObject({
			status: "waiting",
			wait: { event: { type: "slack.message", connectionKey: "sales" } },
		});
	});

	it("requires a scope where the event needs one", () => {
		workflow({
			name: "Unscoped wait",
			description: "Waits for a page without naming its data source",
			triggers: [events.notionPageUpdated()],
			handler: async (_event, context) => {
				// @ts-expect-error Notion page waits need a data source.
				await context.wait.forEvent("Next page", {
					event: events.notionPageCreated(),
					timeout: { hours: 1 },
				});
				await context.wait.forEvent("Mention", {
					event: events.notionAgentMentioned(),
					timeout: { hours: 1 },
				});
			},
		});
	});

	it("rejects events that cannot resolve a wait", () => {
		workflow({
			name: "Invalid waits",
			description: "Waits on triggers that are not events",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.wait.forEvent("Schedule", {
					// @ts-expect-error Scheduled runs are not events.
					event: events.scheduled(),
					timeout: { hours: 1 },
				});
				await context.wait.forEvent("Webhook", {
					// @ts-expect-error Webhook waits need a callback URL first.
					event: events.webhook(),
					timeout: { hours: 1 },
				});
				await context.wait.forEvent("Button", {
					// @ts-expect-error Button presses are not delivered to Workers.
					event: events.notionButtonPressed(),
					timeout: { hours: 1 },
				});
			},
		});
	});
});

describe("context.workflows", () => {
	function createChildWorkflow() {
		const child = workflow({
			name: "Prepare releases",
			description: "Prepares the supplied releases",
			triggers: [events.manual({ inputSchema: j.object({ versions: j.array(j.string()) }) })],
			outputSchema: j.object({ prepared: j.integer() }),
			handler: (event) => ({ prepared: event.input.versions.length }),
		});
		bindWorkflowKey(child, "prepareReleases");
		return child;
	}

	function stubStartRun(runId = "child-run-id") {
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		const fetchMock = vi.fn<typeof fetch>().mockImplementation(
			async () =>
				new Response(JSON.stringify({ runId }), {
					status: 200,
					headers: { "content-type": "application/json" },
				}),
		);
		vi.stubGlobal("fetch", fetchMock);
		return fetchMock;
	}

	async function writeCheckpoint(stepKey: string, value: unknown): Promise<void> {
		const directory = checkpointDirectory ?? (await createCheckpointDirectory());
		await writeFile(
			checkpointPath(directory, stepKey),
			JSON.stringify({
				version: 0,
				key: stepKey,
				value,
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		);
	}

	it("starts a workflow without waiting for it", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		let handle: unknown;
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				handle = await context.workflows.start(child, { versions: ["1.0"] });
			},
		});

		await expect(parent.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(handle).toEqual({ runId: "child-run-id" });
		expect(fetchMock).toHaveBeenCalledWith(
			"https://api.example.com/api/v3/workersWorkflowStartRun",
			expect.objectContaining({
				method: "POST",
				headers: expect.objectContaining({ authorization: "Bearer workflow-token" }),
				body: JSON.stringify({
					workerId: "worker-id",
					runGroupId: "run-group-id",
					stepKey: workflowStepKey(["workflow", "start", "prepareReleases"]),
					targetCapabilityKey: "prepareReleases",
					input: { versions: ["1.0"] },
					capabilityKey: "workflow-key",
				}),
			}),
		);
	});

	it("replays a started run instead of starting another", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		await writeCheckpoint(workflowStepKey(["workflow", "start", "prepareReleases"]), {
			runId: "first-run-id",
		});
		let handle: unknown;
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				handle = await context.workflows.start(child, { versions: ["1.0"] });
			},
		});

		await parent.handler(event, { concreteOutput: true });

		expect(handle).toEqual({ runId: "first-run-id" });
		expect(fetchMock).not.toHaveBeenCalled();
	});

	it("starts a workflow inside a step as part of that step", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		let handle: unknown;
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				handle = await context.step("Kick off", () =>
					context.workflows.start(child, { versions: ["1.0"] }),
				);
			},
		});

		await expect(parent.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(handle).toEqual({ runId: "child-run-id" });
		expect(JSON.parse(String(fetchMock.mock.calls[0]?.[1]?.body))).toMatchObject({
			stepKey: workflowStepKey([
				workflowStepKey(["Kick off"]),
				"workflow",
				"start",
				"prepareReleases",
			]),
		});
		const output = stdoutSpy.mock.calls.map(([chunk]) => String(chunk)).join("");
		expect(output).not.toContain("Start Prepare releases");
	});

	it("sends the same start key when the enclosing step retries", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		let attempts = 0;
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step(
					"Kick off",
					{ retry: { maxAttempts: 2, initialDelayMs: 0 } },
					async () => {
						attempts += 1;
						await context.workflows.start(child, { versions: ["1.0"] });
						if (attempts === 1) {
							throw new RetryableError("transient");
						}
					},
				);
			},
		});

		await expect(parent.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(attempts).toBe(2);
		const stepKeys = fetchMock.mock.calls.map(
			([, init]) => JSON.parse(String(init?.body)).stepKey,
		);
		expect(stepKeys).toHaveLength(2);
		expect(stepKeys[0]).toBe(stepKeys[1]);
	});

	it("rejects the same start key twice within one step", async () => {
		const child = createChildWorkflow();
		stubStartRun();
		const parent = workflow({
			name: "Ship releases",
			description: "Starts releases",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step("Kick off", { retry: { maxAttempts: 1 } }, async () => {
					await context.workflows.start(child, { versions: ["1.0"] });
					await context.workflows.start(child, { versions: ["2.0"] });
				});
			},
		});

		await expect(parent.handler(event)).rejects.toThrow(
			'Workflow invocation "Prepare releases" has the same key as "Prepare releases"',
		);
	});

	it("uses explicit keys to invoke a workflow more than once", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		const parent = workflow({
			name: "Ship releases",
			description: "Starts releases",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				for (const version of ["1.0", "2.0"]) {
					await context.workflows.start(child, { versions: [version] }, { key: version });
				}
			},
		});

		await parent.handler(event, { concreteOutput: true });

		const stepKeys = fetchMock.mock.calls.map(
			(call) => JSON.parse(String(call[1]?.body)).stepKey,
		);
		expect(stepKeys).toEqual([
			workflowStepKey(["workflow", "start", "1.0"]),
			workflowStepKey(["workflow", "start", "2.0"]),
		]);
	});

	it("rejects invoking the same workflow twice with the same key", async () => {
		const child = createChildWorkflow();
		stubStartRun();
		const parent = workflow({
			name: "Ship releases",
			description: "Starts releases",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.workflows.start(child, { versions: ["1.0"] });
				await context.workflows.start(child, { versions: ["2.0"] });
			},
		});

		await expect(parent.handler(event)).rejects.toThrow(
			'Workflow invocation "Prepare releases" has the same key as "Prepare releases".',
		);
	});

	it("rejects input that does not match the target's input schema", async () => {
		const child = createChildWorkflow();
		const fetchMock = stubStartRun();
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				// SAFETY: exercises runtime validation of untyped input.
				await context.workflows.start(child, { versions: [1] } as never);
			},
		});

		await expect(parent.handler(event)).rejects.toThrow(
			'Invalid input for workflow "prepareReleases"',
		);
		expect(fetchMock).not.toHaveBeenCalled();
	});

	it("rejects workflows that are not part of the app", async () => {
		const unbound = workflow({
			name: "Unbound",
			description: "Not registered by the build",
			triggers: [events.manual()],
			handler: () => {},
		});
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.workflows.start(unbound, {});
			},
		});

		await expect(parent.handler(event)).rejects.toThrow(
			"context.workflows can only invoke workflows default-exported from src/workflows/",
		);
	});

	it("fails without retrying when the platform rejects the start", async () => {
		const child = createChildWorkflow();
		process.env.NOTION_API_BASE_URL = "https://api.example.com";
		process.env.NOTION_API_TOKEN = "workflow-token";
		vi.stubGlobal(
			"fetch",
			vi.fn<typeof fetch>().mockResolvedValue(
				new Response(JSON.stringify({ message: "Invocation depth limit reached" }), {
					status: 400,
					headers: { "content-type": "application/json" },
				}),
			),
		);
		const parent = workflow({
			name: "Ship releases",
			description: "Starts a release",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.workflows.start(child, { versions: ["1.0"] });
			},
		});

		await expect(parent.handler(event)).rejects.toThrow(
			'Starting workflow "prepareReleases" failed with status 400: Invocation depth limit reached',
		);
		expect(stdoutSpy).toHaveBeenCalledWith(expect.stringContaining('"_tag":"terminal"'));
	});

	it("types input from the target's manual trigger", () => {
		const child = createChildWorkflow();
		const withoutManualTrigger = workflow({
			name: "Page created",
			description: "Not invocable",
			triggers: [events.notionPageCreated()],
			handler: () => {},
		});
		const typeOnly = (context: WorkflowContext) => {
			void context.workflows.start(child, { versions: ["1.0"] });
			// @ts-expect-error versions must be strings
			void context.workflows.start(child, { versions: [1] });
			// @ts-expect-error workflows without a manual trigger cannot be invoked
			void context.workflows.start(withoutManualTrigger, {});
			expectTypeOf(context.workflows.start(child, { versions: [] })).resolves.toEqualTypeOf<{
				runId: string;
			}>();
		};
		void typeOnly;
	});
});

async function createCheckpointDirectory(): Promise<string> {
	checkpointDirectory = await mkdtemp(join(tmpdir(), "apps-step-checkpoints-"));
	process.env.NOTION_WORKFLOW_STEP_DIRECTORY = checkpointDirectory;
	return checkpointDirectory;
}

function checkpointPath(directory: string, key: string): string {
	const keyHash = createHash("sha256").update(key).digest("hex");
	return join(directory, `${keyHash}.json`);
}

function workflowStepKey(segments: readonly string[]): string {
	return createHash("sha256")
		.update(JSON.stringify(["notion-workflow-step-key", 1, ...segments]))
		.digest("base64url");
}

async function writeRunMetadata(overrides: Partial<RunMetadata> = {}): Promise<void> {
	if (!metadataDirectory) {
		throw new Error("Metadata directory was not created");
	}

	const metadataPath = join(metadataDirectory, "metadata.json");
	await writeFile(
		metadataPath,
		JSON.stringify({
			workspaceId: "workspace-id",
			workerId: "worker-id",
			runGroupId: "run-group-id",
			runId: "run-id",
			capabilityType: "workflow",
			attemptNumber: 2,
			...overrides,
		} satisfies RunMetadata),
	);
	process.env.NOTION_RUN_METADATA_PATH = metadataPath;
}

describe("workflow", () => {
	it("is tagged and exposes triggers in config", () => {
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler: () => {},
		});

		expect(definition._tag).toBe("workflow");
		expect(definition.config).toEqual({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [{ type: "notion.page.created" }],
		});
	});

	it("exposes deadline and retry configuration", () => {
		const deadline = { afterMs: 60_000 };
		const retry = {
			maxAttempts: 3,
			initialDelayMs: 1_000,
			maxDelayMs: 30_000,
			backoffMultiplier: 2,
		};
		const definition = workflow({
			name: "Reliable workflow",
			description: "Configures failure handling",
			triggers: [events.notionPageCreated()],
			deadline,
			retry,
			handler: () => {},
		});

		expect(definition.config.deadline).toEqual(deadline);
		expect(definition.config.retry).toEqual(retry);
		expect(
			workflow({
				name: "Classified workflow",
				description: "Keeps error matchers in the SDK",
				triggers: [events.notionPageCreated()],
				retry: { ...retry, retryOn: [TypeError] },
				handler: () => {},
			}).config.retry,
		).toEqual(retry);
		expect(definition.config.deadline).not.toBe(deadline);
		expect(definition.config.retry).not.toBe(retry);
	});

	it("rejects invalid workflow reliability configuration", () => {
		expect(() =>
			workflow({
				name: "Invalid workflow",
				description: "Has an invalid retry budget",
				triggers: [events.notionPageCreated()],
				deadline: { afterMs: 0 },
				retry: { maxAttempts: 0 },
				handler: () => {},
			}),
		).toThrow("Workflow deadline afterMs must be greater than zero.");
	});

	it("rejects reliability configuration beyond platform limits", () => {
		const define = (config: { deadline?: { afterMs: number }; retry?: WorkflowRetryPolicy }) =>
			workflow({
				name: "Limited workflow",
				description: "Exceeds platform limits",
				triggers: [events.notionPageCreated()],
				...config,
				handler: () => {},
			});

		expect(() => define({ retry: { maxAttempts: 4 } })).toThrow(
			"Workflow retry policy maxAttempts must be an integer between 1 and 3.",
		);
		expect(() => define({ deadline: { afterMs: 8 * 24 * 60 * 60 * 1000 } })).toThrow(
			"Workflow deadline afterMs must be at most 604800000.",
		);
		expect(() =>
			define({ retry: { maxAttempts: 2, retryOn: ["TypeError"] as unknown as [] } }),
		).toThrow(
			"Workflow retry policy retryOn must be an error class, a predicate, or an array of them.",
		);
	});

	it("declares a Calendar connection requirement", () => {
		const definition = workflow({
			name: "Schedule Follow-up",
			description: "Schedules a follow-up after a page is created",
			triggers: [events.notionPageCreated()],
			connections: {
				calendar: connections.calendar({
					targets: {
						calendar: { description: "Calendar", permissions: "read", multiple: true },
					},
				}),
			},
			handler: () => {},
		});

		expect(definition.config.connections).toEqual([
			{
				type: "calendar",
				key: "calendar",
				calendar: {
					targets: {
						calendar: { description: "Calendar", permissions: "read", multiple: true },
					},
				},
			},
		]);
	});

	it("keeps declaration keys stable after the input object changes", async () => {
		const declarations = { support: connections.slack() };
		const seen: string[][] = [];
		const definition = workflow({
			name: "Snapshot connections",
			description: "Captures declared keys",
			connections: declarations,
			triggers: [events.notionPageCreated()],
			handler(_event, context) {
				seen.push(Object.keys(context.connections));
			},
		});
		Object.assign(declarations, {
			support: connections.calendar({
				targets: {
					calendar: { description: "Calendar", permissions: "read", multiple: true },
				},
			}),
			team: connections.slack(),
		});
		expect(definition.config.connections).toEqual([{ type: "slack", key: "support" }]);
		await definition.handler(event, { concreteOutput: true });
		expect(seen).toEqual([["support"]]);
	});
	it("rejects inherited declaration objects before cloning", () => {
		const declarations = { support: connections.slack() };
		Object.setPrototypeOf(declarations, {
			extra: connections.calendar({
				targets: {
					calendar: { description: "Calendar", permissions: "read", multiple: true },
				},
			}),
		});
		expect(() =>
			workflow({
				name: "Inherited keys",
				description: "Reject prototype declarations",
				connections: declarations,
				triggers: [events.notionPageCreated()],
				handler() {},
			}),
		).toThrow("only own");
	});

	it("rejects invalid connection property names", () => {
		expect(() =>
			workflow({
				name: "Schedule Follow-up",
				description: "Schedules a follow-up after a page is created",
				triggers: [events.notionPageCreated()],
				connections: {
					"bad/key": connections.calendar({
						targets: {
							calendar: {
								description: "Calendar",
								permissions: "read",
								multiple: true,
							},
						},
					}),
				},
				handler: () => {},
			}),
		).toThrowError("Invalid workflow connection key: bad/key");
	});

	it("writes a success envelope after the handler runs", async () => {
		const handler = vi.fn();
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler,
		});

		await definition.handler(event);

		expect(handler).toHaveBeenCalledWith(event, expect.anything());
		expect(stdoutSpy).toHaveBeenCalledWith(
			`\n<__notion_output__>{"_tag":"success","value":{"status":"success"}}</__notion_output__>\n`,
		);
	});

	it("uses a file-backed event instead of the inline event", async () => {
		if (!metadataDirectory) {
			throw new Error("Metadata directory was not created");
		}
		const handler = vi.fn();
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler,
		});
		const fileEvent = { ...event, message: "File event" };
		const inputPath = join(metadataDirectory, "input.json");
		await writeFile(inputPath, JSON.stringify(fileEvent));
		process.env.NOTION_INPUT_PAYLOAD_PATH = inputPath;

		await definition.handler(event, { concreteOutput: true });

		expect(handler).toHaveBeenCalledWith(fileEvent, expect.anything());
	});

	it("returns the status without an envelope under concreteOutput", async () => {
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler: () => {},
		});

		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(stdoutSpy).not.toHaveBeenCalled();
	});

	it("writes retryable and terminal error envelopes", async () => {
		const retryable = workflow({
			name: "Retryable workflow",
			description: "Classifies temporary failures",
			triggers: [events.notionPageCreated()],
			handler: () => {
				throw new RetryableError("try later", { retryAfterMs: 5_000 });
			},
		});

		await expect(retryable.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"_tag":"retryable","name":"RetryableError","message":"try later"`,
			),
		);
		expect(stdoutSpy).toHaveBeenLastCalledWith(expect.stringContaining(`"retryAfterMs":5000`));

		stdoutSpy.mockClear();
		const fatal = workflow({
			name: "Fatal workflow",
			description: "Classifies permanent failures",
			triggers: [events.notionPageCreated()],
			handler: () => {
				throw new FatalError("stop now");
			},
		});

		await expect(fatal.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"_tag":"terminal","name":"FatalError","message":"stop now"`,
			),
		);
	});

	it("reports a step failure that used its retries as terminal", async () => {
		const definition = workflow({
			name: "Step failure",
			description: "Does not retry the run after a step exhausts its retries",
			triggers: [events.notionPageCreated()],
			handler: async (_event, context) => {
				await context.step("Fail", { retry: { maxAttempts: 1 } }, () => {
					throw new Error("step down");
				});
			},
		});

		await expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"_tag":"terminal","name":"Error","message":"step down"`,
			),
		);
	});

	it("reports failures outside steps not matched by the workflow retryOn as terminal", async () => {
		const definition = workflow({
			name: "Classified failure",
			description: "Applies retryOn to failures outside steps",
			triggers: [events.notionPageCreated()],
			retry: { maxAttempts: 2, retryOn: RangeError },
			handler: () => {
				throw new TypeError("bug");
			},
		});

		await expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenLastCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"_tag":"terminal","name":"TypeError","message":"bug"`,
			),
		);
	});

	it("writes an error envelope and throws ExecutionError on failure", async () => {
		const definition = workflow({
			name: "Log Pages",
			description: "Logs created pages",
			triggers: [events.notionPageCreated()],
			handler: () => {
				throw new Error("boom");
			},
		});

		await expect(definition.handler(event)).rejects.toBeInstanceOf(ExecutionError);
		expect(stdoutSpy).toHaveBeenCalledWith(
			expect.stringContaining(
				`<__notion_output__>{"_tag":"error","error":{"name":"ExecutionError","message":"Error during app execution: Error: boom"`,
			),
		);
	});
	it("normalizes NaC access and hydrates all supported resource kinds", async () => {
		const home = page({ resourceId: "home" });
		const table = database("table", {
			dataSourceResourceId: "table-source",
			name: "Table",
			schema: { Title: { resourceId: "title", type: "title" } },
		});
		const agent = customAgent({ resourceId: "agent", name: "Agent" });
		let resolvedAccess:
			| {
					readonly home: { readonly type: "page"; readonly id: string };
					readonly table: { readonly type: "database"; readonly id: string };
					readonly source: { readonly type: "dataSource"; readonly id: string };
					readonly agent: { readonly type: "customAgent"; readonly id: string };
			  }
			| undefined;
		const definition = workflow({
			name: "Access",
			description: "Hydrates access",
			triggers: [events.notionPageCreated()],
			access: {
				home: access.view(home),
				table: access.edit(table),
				source: access.comment(table.dataSource),
				agent: access.call(agent),
			},
			handler: (_event, context) => {
				expectTypeOf(context.access.home.type).toEqualTypeOf<"page">();
				expectTypeOf(context.access.table.type).toEqualTypeOf<"database">();
				expectTypeOf(context.access.source.type).toEqualTypeOf<"dataSource">();
				expectTypeOf(context.access.agent.type).toEqualTypeOf<"customAgent">();
				resolvedAccess = context.access;
			},
		});

		expect(definition.config.access).toEqual({
			home: { type: "page", resourceId: "home", level: "view" },
			table: { type: "database", resourceId: "table", level: "edit" },
			source: { type: "dataSource", resourceId: "table-source", level: "comment" },
			agent: { type: "customAgent", resourceId: "agent", level: "call" },
		});
		process.env.NOTION_ACCESS_BINDINGS = JSON.stringify({
			home: { type: "page", resourceId: "home", level: "view", id: "page-id" },
			table: { type: "database", resourceId: "table", level: "edit", id: "database-id" },
			source: {
				type: "dataSource",
				resourceId: "table-source",
				level: "comment",
				id: "data-source-id",
			},
			agent: { type: "customAgent", resourceId: "agent", level: "call", id: "agent-id" },
		});

		await definition.handler(event, { concreteOutput: true });

		expect(resolvedAccess).toEqual({
			home: { type: "page", id: "page-id" },
			table: { type: "database", id: "database-id" },
			source: { type: "dataSource", id: "data-source-id" },
			agent: { type: "customAgent", id: "agent-id" },
		});
		expect(Object.isFrozen(resolvedAccess)).toBe(true);
		expect(Object.isFrozen(resolvedAccess?.home)).toBe(true);
	});

	it("rejects missing or mismatched access bindings before the handler", async () => {
		const handler = vi.fn();
		const definition = workflow({
			name: "Access",
			description: "Requires access",
			triggers: [events.notionPageCreated()],
			access: { home: { resource: page({ resourceId: "home" }), level: "view" } },
			handler,
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"not bound",
		);
		expect(handler).not.toHaveBeenCalled();

		process.env.NOTION_ACCESS_BINDINGS = JSON.stringify({
			home: { type: "page", resourceId: "other", level: "view", id: "page-id" },
		});
		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"does not match",
		);
		expect(handler).not.toHaveBeenCalled();
	});

	it("does not parse unused access metadata when no access is declared", async () => {
		process.env.NOTION_ACCESS_BINDINGS = "{malformed";
		let contextAccess: unknown;
		const definition = workflow({
			name: "No Access",
			description: "Runs without access",
			triggers: [events.notionPageCreated()],
			handler: (_event, context) => {
				contextAccess = context.access;
			},
		});

		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});
		expect(contextAccess).toEqual({});
		expect(Object.isFrozen(contextAccess)).toBe(true);
	});
});

describe("connection-bound triggers", () => {
	it("resolves typed trigger callbacks once and preserves event inference", () => {
		let calls = 0;
		const definition = workflow({
			name: "Typed triggers",
			description: "Uses only declared provider keys",
			connections: {
				support: connections.slack(),
				calendar: connections.calendar({
					targets: {
						calendar: { description: "Calendar", permissions: "read", multiple: true },
					},
				}),
			},
			triggers: ({ events, connections }) => {
				calls += 1;
				expectTypeOf(events.slackMessage)
					.parameter(0)
					.toEqualTypeOf<{ connectionKey: "support" } | undefined>();
				expectTypeOf(
					connections.calendar.targets.calendar.connectionKey,
				).toEqualTypeOf<"calendar">();
				expectTypeOf(
					connections.calendar.targets.calendar.target,
				).toEqualTypeOf<"calendar">();
				expectTypeOf(events.discordInteraction)
					.parameter(0)
					.toEqualTypeOf<{ connectionKey: never } | undefined>();
				return [
					events.slackMessage({ connectionKey: "support" }),
					events.calendarEventCreated({
						calendars: connections.calendar.targets.calendar,
					}),
				];
			},
			handler: (event, context) => {
				expectTypeOf(event.type).toEqualTypeOf<
					"slack.message" | "calendar.event.created"
				>();
				expectTypeOf<keyof typeof context.connections>().toEqualTypeOf<
					"support" | "calendar"
				>();
			},
		});
		expect(calls).toBe(1);
		expect(definition.config.triggers).toEqual([
			{ type: "slack.message", connectionKey: "support" },
			{
				type: "calendar.event.created",
				connectionKey: "calendar",
				calendarTarget: "calendar",
			},
		]);
	});

	it("preserves the binding in the manifest and infers the event", () => {
		const definition = workflow({
			name: "Support",
			description: "Watch support messages",
			connections: { support: connections.slack() },
			triggers: [events.slackMessage({ connectionKey: "support" })],
			handler(event) {
				expectTypeOf(event.type).toEqualTypeOf<"slack.message">();
			},
		});
		expect(definition.config.triggers).toEqual([
			{ type: "slack.message", connectionKey: "support" },
		]);
	});
	it.each([
		{ requirements: {} },
		{
			requirements: {
				support: connections.calendar({
					targets: {
						calendar: { description: "Calendar", permissions: "read", multiple: true },
					},
				}),
			},
		},
		{ requirements: { other: connections.slack() } },
	])("rejects missing or incompatible declarations: %j", ({ requirements }) => {
		expect(() =>
			workflow({
				name: "Support",
				description: "",
				connections: requirements,
				triggers: [events.slackMessage({ connectionKey: "support" })],
				handler() {},
			}),
		).toThrow("requires a declared slack connection");
	});
	it("rejects duplicate bindings while allowing distinct connections", () => {
		const configuration = {
			name: "Support",
			description: "",
			connections: { support: connections.slack(), internal: connections.slack() },
			handler() {},
		};
		expect(() =>
			workflow({
				...configuration,
				triggers: [
					events.slackMessage({ connectionKey: "support" }),
					events.slackMessage({ connectionKey: "support" }),
				],
			}),
		).toThrow("Duplicate");
		expect(() =>
			workflow({
				...configuration,
				triggers: [
					events.slackMessage({ connectionKey: "support" }),
					events.slackMessage({ connectionKey: "internal" }),
				],
			}),
		).not.toThrow();
	});
	it("preserves existing unkeyed helpers", () => {
		expect(events.slackMessage()).toEqual({ type: "slack.message" });
		expect(() =>
			workflow({
				name: "Existing",
				description: "",
				triggers: [events.slackMessage()],
				handler() {},
			}),
		).not.toThrow();
	});
});

describe("step cancellation", () => {
	function cancellingWorkflow(
		handler: (
			event: NotionPageCreatedEvent,
			context: WorkflowContext<Record<never, never>, Record<never, never>>,
		) => Promise<void> | void,
	) {
		return workflow({
			name: "Cancellable workflow",
			description: "Observes run cancellation",
			triggers: [events.notionPageCreated()],
			handler,
		});
	}

	it("passes an unaborted signal in the handler context", async () => {
		let signal: AbortSignal | undefined;
		const definition = cancellingWorkflow((_event, context) => {
			signal = context.signal;
		});

		await expect(definition.handler(event, { concreteOutput: true })).resolves.toEqual({
			status: "success",
		});

		expect(signal).toBeInstanceOf(AbortSignal);
		expect(signal?.aborted).toBe(false);
	});

	it("aborts the handler's signal outside of any step", async () => {
		const definition = cancellingWorkflow(async (_event, { signal }) => {
			process.emit("SIGTERM", "SIGTERM");
			await Promise.resolve();
			signal.throwIfAborted();
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"Workflow run was cancelled (SIGTERM).",
		);
	});

	it("aborts a running step's signal on SIGTERM and fails the step", async () => {
		let reached!: () => void;
		const stepStarted = new Promise<void>((resolve) => {
			reached = resolve;
		});
		const definition = cancellingWorkflow(async (_event, context) => {
			const { signal } = context;
			await context.step("Wait for abort", () => {
				reached();
				return new Promise((_resolve, reject) => {
					signal.addEventListener("abort", () => reject(signal.reason), { once: true });
				});
			});
		});

		const run = definition.handler(event, { concreteOutput: true });
		await stepStarted;
		process.emit("SIGTERM", "SIGTERM");

		await expect(run).rejects.toThrow("Workflow run was cancelled (SIGTERM).");
		const failure = emittedStepEvents().find(({ event }) => event === "failure");
		expect(failure?.value).toMatchObject({
			name: "Wait for abort",
			error: {
				name: "ExecutionError",
				message:
					"Error during app execution: AbortError: Workflow run was cancelled (SIGTERM).",
			},
		});
	});

	it("aborts the step callback's signal on SIGTERM", async () => {
		let reached!: () => void;
		const stepStarted = new Promise<void>((resolve) => {
			reached = resolve;
		});
		let stepSignal: AbortSignal | undefined;
		const definition = cancellingWorkflow(async (_event, context) => {
			await context.step("Wait for abort", { timeoutMs: 60_000 }, ({ signal }) => {
				stepSignal = signal;
				reached();
				return new Promise((_resolve, reject) => {
					signal.addEventListener("abort", () => reject(signal.reason), { once: true });
				});
			});
		});

		const run = definition.handler(event, { concreteOutput: true });
		await stepStarted;
		process.emit("SIGTERM", "SIGTERM");

		await expect(run).rejects.toThrow("Workflow run was cancelled (SIGTERM).");
		expect(stepSignal?.aborted).toBe(true);
		expect(stepSignal?.reason).toMatchObject({ name: "AbortError" });
	});

	it("supports throwIfAborted between units of work", async () => {
		const processed: number[] = [];
		const definition = cancellingWorkflow(async (_event, context) => {
			const { signal } = context;
			await context.step("Process records", async () => {
				for (const record of [1, 2, 3]) {
					signal.throwIfAborted();
					processed.push(record);
					if (record === 2) {
						process.emit("SIGTERM", "SIGTERM");
					}
					await Promise.resolve();
				}
			});
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"Workflow run was cancelled (SIGTERM).",
		);
		expect(processed).toEqual([1, 2]);
	});

	it("does not start new steps after cancellation but still replays saved ones", async () => {
		const directory = await createCheckpointDirectory();
		await writeFile(
			checkpointPath(directory, workflowStepKey(["Saved"])),
			JSON.stringify({
				version: 0,
				key: workflowStepKey(["Saved"]),
				value: "saved-value",
				sourceRunId: "source-run",
				sourceStepId: "source-step",
			}),
		);
		const saved = vi.fn(() => "new-value");
		const next = vi.fn();
		let replayed: string | undefined;
		const definition = cancellingWorkflow(async (_event, context) => {
			await context.step("Cancel", () => {
				process.emit("SIGTERM", "SIGTERM");
			});
			replayed = await context.step("Saved", saved);
			await context.step("Next", next);
		});

		await expect(definition.handler(event, { concreteOutput: true })).rejects.toThrow(
			"Workflow run was cancelled (SIGTERM).",
		);
		expect(replayed).toBe("saved-value");
		expect(saved).not.toHaveBeenCalled();
		expect(next).not.toHaveBeenCalled();
	});

	it("removes its process signal listener when the run ends", async () => {
		const before = process.listenerCount("SIGTERM");
		let during: number | undefined;
		const definition = cancellingWorkflow(() => {
			during = process.listenerCount("SIGTERM");
		});

		await definition.handler(event, { concreteOutput: true });

		expect(during).toBe(before + 1);
		expect(process.listenerCount("SIGTERM")).toBe(before);
	});

	it("stops listening after the first signal so a second one uses the default behavior", async () => {
		const before = process.listenerCount("SIGTERM");
		let afterSignal: number | undefined;
		const definition = cancellingWorkflow(() => {
			process.emit("SIGTERM", "SIGTERM");
			afterSignal = process.listenerCount("SIGTERM");
		});

		await definition.handler(event, { concreteOutput: true });

		expect(afterSignal).toBe(before);
	});
});

describe("workflow verify", () => {
	const verifyRequest = {
		method: "POST" as const,
		url: "https://123456.webhook.notionusercontent.com/webhooks/app/s/w/u/incoming",
		query: {},
		headers: { "x-signature": "sig" },
		rawBody: '{"challenge":"abc"}',
	};

	it("marks the webhook trigger with hasVerify when a verify handler is set", () => {
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: () => ({ status: 200 }),
			handler: () => {},
		});

		expect(incoming.config).toEqual({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [{ type: "webhooks.webhook", hasVerify: true }],
		});
	});

	it("flags only the webhook trigger among a workflow's triggers", () => {
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.notionPageCreated(), events.webhook()],
			verify: () => ({ status: 200 }),
			handler: () => {},
		});

		expect(incoming.config.triggers).toEqual([
			{ type: "notion.page.created" },
			{ type: "webhooks.webhook", hasVerify: true },
		]);
	});

	it("omits hasVerify from the trigger when no verify handler is set", () => {
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			handler: () => {},
		});

		expect(incoming.config.triggers).toEqual([{ type: "webhooks.webhook" }]);
	});

	it("rejects a verify handler on a workflow without a webhook trigger", () => {
		expect(() =>
			workflow({
				name: "Log Pages",
				description: "Logs created pages",
				triggers: [events.notionPageCreated()],
				verify: () => ({ status: 200 }),
				handler: () => {},
			}),
		).toThrow(/has no webhook trigger/);
	});

	it("dispatches a verify invocation to the verify handler, not the workflow handler", async () => {
		const handler = vi.fn();
		const verify: WebhookVerifyHandler = (request) => ({
			status: 200,
			body: (JSON.parse(request.rawBody) as { challenge: string }).challenge,
			contentType: "text/plain",
		});
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify,
			handler,
		});

		const response = await incoming.handler(
			{ request: verifyRequest },
			{ concreteOutput: true },
		);

		expect(response).toEqual({ status: 200, body: "abc", contentType: "text/plain" });
		expect(handler).not.toHaveBeenCalled();
	});

	it("returns an explicit delivery decision from the verify handler", async () => {
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: () => ({ status: 200, deliver: false }),
			handler: () => {},
		});

		const response = await incoming.handler(
			{ request: verifyRequest },
			{ concreteOutput: true },
		);

		expect(response).toEqual({ status: 200, deliver: false });
	});

	it("passes the capability context to the verify handler", async () => {
		let receivedContext: unknown;
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: (_request, context) => {
				receivedContext = context;
				return { status: 200 };
			},
			handler: () => {},
		});

		await incoming.handler({ request: verifyRequest }, { concreteOutput: true });

		expect(receivedContext).toHaveProperty("notion");
		expect(receivedContext).not.toHaveProperty("step");
	});

	it("verifies without workflow run metadata", async () => {
		delete process.env.NOTION_RUN_METADATA_PATH;
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: () => ({ status: 200 }),
			handler: () => {},
		});

		await expect(
			incoming.handler({ request: verifyRequest }, { concreteOutput: true }),
		).resolves.toEqual({ status: 200 });
	});

	it("writes the verify response as a success envelope without concreteOutput", async () => {
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: () => ({ status: 401 }),
			handler: () => {},
		});

		await incoming.handler({ request: verifyRequest });

		expect(stdoutSpy).toHaveBeenCalledWith(
			expect.stringContaining('"_tag":"success","value":{"status":401}'),
		);
	});

	it("runs in the app bundle, so verify handlers may capture module scope", async () => {
		const signingSecret = "shh";
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: (request) => ({
				status: request.headers["x-signature"] === signingSecret ? 200 : 401,
			}),
			handler: () => {},
		});

		const response = await incoming.handler(
			{ request: { ...verifyRequest, headers: { "x-signature": "shh" } } },
			{ concreteOutput: true },
		);

		expect(response).toEqual({ status: 200 });
	});

	it("uses a file-backed verify request instead of the inline invocation", async () => {
		if (!metadataDirectory) {
			throw new Error("Metadata directory was not created");
		}
		const payloadPath = join(metadataDirectory, "verify-input.json");
		await writeFile(
			payloadPath,
			JSON.stringify({ request: { ...verifyRequest, rawBody: '{"challenge":"file"}' } }),
		);
		process.env.NOTION_INPUT_PAYLOAD_PATH = payloadPath;

		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: (request) => ({
				status: 200,
				body: (JSON.parse(request.rawBody) as { challenge: string }).challenge,
			}),
			handler: () => {},
		});

		const response = await incoming.handler(
			{ request: verifyRequest },
			{ concreteOutput: true },
		);

		expect(response).toEqual({ status: 200, body: "file" });
	});

	it("still dispatches trigger events to the workflow handler when verify is configured", async () => {
		const handler = vi.fn();
		const webhookEvent = { type: "webhooks.webhook" } as WebhookEvent;
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			verify: () => ({ status: 200 }),
			handler,
		});

		const result = await incoming.handler(webhookEvent, { concreteOutput: true });

		expect(result).toEqual({ status: "success" });
		expect(handler).toHaveBeenCalledWith(webhookEvent, expect.anything());
	});

	it("rejects a verify invocation when no verify handler is declared", async () => {
		const handler = vi.fn();
		const incoming = workflow({
			name: "Incoming",
			description: "Processes an incoming webhook",
			triggers: [events.webhook()],
			handler,
		});

		await expect(
			incoming.handler({ request: verifyRequest }, { concreteOutput: true }),
		).rejects.toThrow(/does not declare a verify handler/);
		expect(handler).not.toHaveBeenCalled();
	});
});