export type WorkflowStatus = "pending" | "running" | "waiting-approval" | "completed" | "failed" | "cancelled"; export interface WorkflowStep { name: string; dependsOn?: string[]; approval?: boolean; run(input: I, context: WorkflowRunContext): O | Promise; } export interface WorkflowRunContext { workflowId: string; step: string; results: Readonly>; signal: AbortSignal; progress(value: number, message?: string): void; } export interface WorkflowSnapshot { id: string; name: string; status: WorkflowStatus; input: unknown; results: Record; completed: string[]; waitingFor?: string; progress: number; message?: string; error?: string; updatedAt: number; } export interface WorkflowStore { get(id: string): Promise; put(snapshot: WorkflowSnapshot): Promise; list(): Promise; } export function memoryWorkflowStore(): WorkflowStore { const values = new Map(); return { async get(id) { const value = values.get(id); return value ? structuredClone(value) : null; }, async put(value) { values.set(value.id, structuredClone(value)); }, async list() { return [...values.values()].map((value) => structuredClone(value)); }, }; } export interface WorkflowDefinition { name: string; steps: WorkflowStep[]; /** Compile-time input marker; definitions do not store runtime input values. */ readonly __input?: I; } export interface WorkflowEngine { start(definition: WorkflowDefinition, input: I, id?: string): Promise; resume(definition: WorkflowDefinition, id: string): Promise; approve( definition: WorkflowDefinition, id: string, step: string, actor: string, ): Promise; cancel(id: string): Promise; get(id: string): Promise; list(): Promise; } function validate(definition: WorkflowDefinition): void { const names = new Set(definition.steps.map((step) => step.name)); if (names.size !== definition.steps.length) throw new Error("WRN-WORKFLOW-DUPLICATE-STEP"); for (const step of definition.steps) for (const dependency of step.dependsOn ?? []) if (!names.has(dependency)) throw new Error( `WRN-WORKFLOW-DEPENDENCY: '${step.name}' depends on missing '${dependency}'.`, ); const visit = (name: string, path: Set): void => { if (path.has(name)) throw new Error(`WRN-WORKFLOW-CYCLE: ${[...path, name].join(" -> ")}`); const next = new Set(path).add(name); const step = definition.steps.find((value) => value.name === name)!; for (const dependency of step.dependsOn ?? []) visit(dependency, next); }; for (const step of definition.steps) visit(step.name, new Set()); } export function createWorkflowEngine(store: WorkflowStore = memoryWorkflowStore()): WorkflowEngine { const controllers = new Map(); async function execute( definition: WorkflowDefinition, snapshot: WorkflowSnapshot, ): Promise { validate(definition); const controller = new AbortController(); controllers.set(snapshot.id, controller); snapshot.status = "running"; await store.put(snapshot); try { while (snapshot.completed.length < definition.steps.length) { const ready = definition.steps.filter( (step) => !snapshot.completed.includes(step.name) && (step.dependsOn ?? []).every((dependency) => snapshot.completed.includes(dependency)), ); if (!ready.length) throw new Error("WRN-WORKFLOW-BLOCKED: no runnable steps."); const step = ready[0]!; if (step.approval && snapshot.waitingFor !== `approved:${step.name}`) { snapshot.status = "waiting-approval"; snapshot.waitingFor = step.name; snapshot.updatedAt = Date.now(); await store.put(snapshot); return structuredClone(snapshot); } snapshot.waitingFor = undefined; const dependencies = step.dependsOn ?? []; const value = dependencies.length === 1 ? snapshot.results[dependencies[0]!] : dependencies.length ? Object.fromEntries(dependencies.map((name) => [name, snapshot.results[name]])) : snapshot.input; snapshot.results[step.name] = await step.run(value, { workflowId: snapshot.id, step: step.name, results: snapshot.results, signal: controller.signal, progress(value, message) { snapshot.progress = Math.max(0, Math.min(100, value)); snapshot.message = message; snapshot.updatedAt = Date.now(); void store.put(snapshot); }, }); snapshot.completed.push(step.name); snapshot.progress = Math.round((snapshot.completed.length / definition.steps.length) * 100); snapshot.updatedAt = Date.now(); await store.put(snapshot); } snapshot.status = "completed"; snapshot.progress = 100; } catch (error) { snapshot.status = controller.signal.aborted ? "cancelled" : "failed"; snapshot.error = error instanceof Error ? error.message : String(error); } finally { snapshot.updatedAt = Date.now(); controllers.delete(snapshot.id); await store.put(snapshot); } return structuredClone(snapshot); } return { async start(definition, input, id = `workflow-${crypto.randomUUID()}`) { if (await store.get(id)) throw new Error(`WRN-WORKFLOW-ID: '${id}' already exists.`); return execute(definition, { id, name: definition.name, status: "pending", input, results: {}, completed: [], progress: 0, updatedAt: Date.now(), }); }, async resume(definition, id) { const snapshot = await store.get(id); if (!snapshot) throw new Error(`WRN-WORKFLOW-NOT-FOUND: '${id}'.`); if (["completed", "cancelled"].includes(snapshot.status)) return snapshot; return execute(definition, snapshot); }, async approve(definition, id, step, actor) { const snapshot = await store.get(id); if (!snapshot || snapshot.status !== "waiting-approval" || snapshot.waitingFor !== step) throw new Error(`WRN-WORKFLOW-APPROVAL: '${step}' is not awaiting approval.`); snapshot.waitingFor = `approved:${step}`; snapshot.results[`${step}:approval`] = { actor, approvedAt: Date.now() }; await store.put(snapshot); return execute(definition, snapshot); }, async cancel(id) { const snapshot = await store.get(id); if (!snapshot || ["completed", "cancelled"].includes(snapshot.status)) return false; controllers.get(id)?.abort(); snapshot.status = "cancelled"; snapshot.updatedAt = Date.now(); await store.put(snapshot); return true; }, get: (id) => store.get(id), list: () => store.list(), }; } export function defineDurableWorkflow(definition: WorkflowDefinition): WorkflowDefinition { validate(definition); return definition; }