import { Worker } from "node:worker_threads"; const KEY_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/; const WORKER_SOURCE = String.raw` const { parentPort } = require("node:worker_threads"); const vm = require("node:vm"); const { inspect } = require("node:util"); let nextCallId = 0; const pending = new Map(); const runKeyPattern = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/; function stableRunJson(value) { if (Array.isArray(value)) return "[" + value.map(stableRunJson).join(",") + "]"; if (value && typeof value === "object") return "{" + Object.keys(value).sort().map((key) => JSON.stringify(key) + ":" + stableRunJson(value[key])).join(",") + "}"; return JSON.stringify(value) ?? "undefined"; } function hostCall(method, args) { return new Promise((resolve, reject) => { const callId = ++nextCallId; pending.set(callId, { resolve, reject }); parentPort.postMessage({ type: "call", callId, method, args }); }); } function formatRef(result) { if (!result || typeof result !== "object") throw new Error("runs.ref(result) requires a run result object."); const parts = ["run " + (result.key || "unknown")]; if (result.runId) parts.push("id=" + String(result.runId).slice(0, 8)); return "[" + parts.join("; ") + "]"; } const runFingerprints = new Map(); function validateRunCall(key, params, label, fingerprints) { if (typeof key !== "string" || !runKeyPattern.test(key)) throw new Error(label + " has an invalid key."); if (!params || typeof params !== "object" || Array.isArray(params)) throw new Error(label + " requires a params object."); if (Object.prototype.hasOwnProperty.call(params, "action") || Object.prototype.hasOwnProperty.call(params, "workflowScript") || Object.prototype.hasOwnProperty.call(params, "tasks") || Object.prototype.hasOwnProperty.call(params, "chain") || Object.prototype.hasOwnProperty.call(params, "concurrency") || Object.prototype.hasOwnProperty.call(params, "chainDir")) { const hint = label === "runs.run" ? "; use runs.all(...) and JavaScript control flow for orchestration." : "."; throw new Error(label + " accepts one child via { agent, task } and execution controls only" + hint); } if (params.worktree !== undefined && typeof params.worktree !== "boolean") throw new Error(label + " worktree must be true or false."); assertJsonValue(params, label + " params"); const fingerprint = stableRunJson(params); const existing = fingerprints.get(key); if (existing !== undefined && existing !== fingerprint) throw new Error("Duplicate workflow key '" + key + "' used with incompatible launch params."); fingerprints.set(key, fingerprint); } const runs = Object.freeze({ run(key, params) { validateRunCall(key, params, "runs.run", runFingerprints); return hostCall("run", { key, params }); }, all(items) { if (!Array.isArray(items)) throw new Error("runs.all(items) requires an array."); const fingerprints = new Map(runFingerprints); const calls = []; for (let index = 0; index < items.length; index++) { if (!Object.prototype.hasOwnProperty.call(items, index)) throw new Error("runs.all items must not contain sparse entries."); const item = items[index]; if (!item || typeof item !== "object" || Array.isArray(item)) throw new Error("runs.all item " + index + " must be an object."); const { key, ...params } = item; validateRunCall(key, params, "runs.all item " + index, fingerprints); calls.push({ key, params }); } for (const { key, params } of calls) runFingerprints.set(key, stableRunJson(params)); return Promise.all(calls.map(({ key, params }) => hostCall("run", { key, params, collectFailure: true }))); }, status(keyOrRunId) { return hostCall("status", { keyOrRunId }); }, ref: formatRef, refs(results) { if (!Array.isArray(results)) throw new Error("runs.refs(results) requires an array."); return results.map(formatRef).join("\n"); }, }); let contextObjectPrototype; const capturedConsole = Object.freeze(Object.fromEntries( ["log", "info", "warn", "error"].map((level) => [level, (...args) => { parentPort.postMessage({ type: "console", level, text: args.map((value) => typeof value === "string" ? value : inspect(value, { depth: 4, breakLength: 120 })).join(" ") }); }]), )); function assertJsonValue(value, path = "emit", seen = new Set()) { if (value === null || typeof value === "string" || typeof value === "boolean") return; if (typeof value === "number") { if (!Number.isFinite(value)) throw new Error(path + " must contain only finite JSON numbers."); return; } if (typeof value !== "object") throw new Error(path + " must be a JSON value; received " + typeof value + "."); if (seen.has(value)) throw new Error(path + " must not contain cycles."); seen.add(value); if (Array.isArray(value)) { for (let index = 0; index < value.length; index++) { if (!Object.prototype.hasOwnProperty.call(value, index)) throw new Error(path + " must not contain sparse array entries."); assertJsonValue(value[index], path + "[" + index + "]", seen); } } else { const prototype = Object.getPrototypeOf(value); if (prototype !== null && prototype !== Object.prototype && prototype !== contextObjectPrototype) throw new Error(path + " must contain only plain JSON objects."); if (Object.getOwnPropertySymbols(value).length > 0) throw new Error(path + " must not contain symbol keys."); for (const [key, entry] of Object.entries(value)) assertJsonValue(entry, path + "." + key, seen); } seen.delete(value); } parentPort.on("message", async (message) => { if (message.type === "response") { const entry = pending.get(message.callId); if (!entry) return; pending.delete(message.callId); if (message.ok) entry.resolve(message.value); else entry.reject(new Error(message.error)); return; } if (message.type !== "start") return; try { const sandbox = { runs, emit(value) { assertJsonValue(value); parentPort.postMessage({ type: "emit", value }); }, console: capturedConsole }; const context = vm.createContext(sandbox, { codeGeneration: { strings: false, wasm: false } }); contextObjectPrototype = vm.runInContext("Object.prototype", context); const compiled = new vm.Script("(async () => {\n" + message.script + "\n})()", { filename: "workflow-script.js" }); const value = await compiled.runInContext(context); const persistedValue = value === undefined ? null : value; assertJsonValue(persistedValue, "return"); parentPort.postMessage({ type: "complete", value: persistedValue }); } catch (error) { parentPort.postMessage({ type: "error", error: error && error.stack ? error.stack : String(error) }); } }); `; export interface WorkflowScriptChildResult { key: string; ok: boolean; runId?: string; output: string; error?: string; structuredOutput?: unknown; artifactPaths: string[]; results?: unknown[]; } export interface WorkflowScriptTraceEntry { operation: "run" | "status"; key: string; state: "started" | "completed" | "failed" | "reused"; runId?: string; durationMs?: number; phase?: string; label?: string; error?: string; } export interface WorkflowScriptResult { value: unknown; emits: unknown[]; console: Array<{ level: "log" | "info" | "warn" | "error"; text: string }>; trace: WorkflowScriptTraceEntry[]; children: WorkflowScriptChildResult[]; } export class WorkflowScriptError extends Error { readonly partial: Omit; constructor(message: string, partial: Omit) { super(message); this.name = "WorkflowScriptError"; this.partial = partial; } } export interface RunWorkflowScriptOptions { script: string; timeoutMs: number; signal?: AbortSignal; launch: (key: string, params: Record, signal: AbortSignal) => Promise; status: (keyOrRunId: string, signal: AbortSignal) => Promise; onTrace?: (trace: WorkflowScriptTraceEntry[]) => void; onEmit?: (emits: unknown[]) => void; } function isRecord(value: unknown): value is Record { return !!value && typeof value === "object" && !Array.isArray(value); } function isPlainJsonObject(value: unknown): value is Record { if (!isRecord(value)) return false; const prototype = Object.getPrototypeOf(value); return prototype === null || prototype === Object.prototype; } function omitUndefinedWorkflowValues(value: unknown, seen = new Set()): unknown { if (value === null || typeof value !== "object") return value; if (seen.has(value)) return value; seen.add(value); const normalized = Array.isArray(value) ? value.map((entry) => entry === undefined ? null : omitUndefinedWorkflowValues(entry, seen)) : isPlainJsonObject(value) ? Object.fromEntries(Object.entries(value).flatMap(([key, entry]) => entry === undefined ? [] : [[key, omitUndefinedWorkflowValues(entry, seen)]])) : value; seen.delete(value); return normalized; } export function assertWorkflowJsonValue(value: unknown, path = "value", seen = new Set()): void { if (value === null || typeof value === "string" || typeof value === "boolean") return; if (typeof value === "number") { if (!Number.isFinite(value)) throw new Error(`${path} must contain only finite JSON numbers.`); return; } if (typeof value !== "object") throw new Error(`${path} must be a JSON value; received ${typeof value}.`); if (seen.has(value)) throw new Error(`${path} must not contain cycles.`); seen.add(value); if (Array.isArray(value)) { for (let index = 0; index < value.length; index++) { if (!Object.hasOwn(value, index)) throw new Error(`${path} must not contain sparse array entries.`); assertWorkflowJsonValue(value[index], `${path}[${index}]`, seen); } } else { const prototype = Object.getPrototypeOf(value); if (prototype !== null && prototype !== Object.prototype) throw new Error(`${path} must contain only plain JSON objects.`); if (Object.getOwnPropertySymbols(value).length > 0) throw new Error(`${path} must not contain symbol keys.`); for (const [key, entry] of Object.entries(value)) assertWorkflowJsonValue(entry, `${path}.${key}`, seen); } seen.delete(value); } export function formatWorkflowJsonPreview(value: unknown, maxLength: number): string | undefined { try { assertWorkflowJsonValue(value); const serialized = JSON.stringify(value); return typeof serialized === "string" ? serialized.slice(0, maxLength) : undefined; } catch { return undefined; } } function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (isRecord(value)) return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${stableJson(value[key])}`).join(",")}}`; return JSON.stringify(value) ?? "undefined"; } function validateKey(value: unknown): string { if (typeof value !== "string" || !KEY_PATTERN.test(value)) { throw new Error("runs.run key must be 1-128 characters using letters, numbers, '.', '_' or '-', and start with a letter or number."); } return value; } function workflowStringMetadata(params: Record): Pick { return { ...(typeof params.phase === "string" && params.phase.trim() ? { phase: params.phase.trim() } : {}), ...(typeof params.label === "string" && params.label.trim() ? { label: params.label.trim() } : {}), }; } export async function runWorkflowScript(options: RunWorkflowScriptOptions): Promise { if (!options.script.trim()) throw new Error("workflowScript must not be empty."); if (!Number.isInteger(options.timeoutMs) || options.timeoutMs < 1) throw new Error("workflow script timeout must be a positive integer."); const worker = new Worker(WORKER_SOURCE, { eval: true }); const emits: unknown[] = []; const consoleEntries: WorkflowScriptResult["console"] = []; const trace: WorkflowScriptTraceEntry[] = []; const children = new Map(); const childOrder: string[] = []; const launches = new Map }>(); const childController = new AbortController(); let settled = false; const partial = (): Omit => ({ emits, console: consoleEntries, trace, children: childOrder.flatMap((key) => { const child = children.get(key); return child ? [child] : []; }) }); const traceChanged = () => options.onTrace?.([...trace]); return await new Promise((resolve, reject) => { const finish = (outcome: { value: unknown } | { error: Error }) => { if (settled) return; settled = true; clearTimeout(timer); options.signal?.removeEventListener("abort", onAbort); void worker.terminate(); childController.abort("error" in outcome ? outcome.error : new Error("Workflow script completed; unawaited child launches are aborted.")); if ("error" in outcome) reject(new WorkflowScriptError(outcome.error.message, partial())); else resolve({ value: outcome.value, ...partial() }); }; const onAbort = () => finish({ error: new Error("Workflow script aborted.") }); const timer = setTimeout(() => finish({ error: new Error(`Workflow script timed out after ${options.timeoutMs}ms.`) }), options.timeoutMs); options.signal?.addEventListener("abort", onAbort, { once: true }); if (options.signal?.aborted) return onAbort(); worker.on("error", (error) => finish({ error: new Error(`Workflow worker failed: ${error instanceof Error ? error.message : String(error)}`) })); worker.on("exit", (code) => { if (!settled && code !== 0) finish({ error: new Error(`Workflow worker exited with code ${code}.`) }); }); worker.on("message", (message: Record) => { if (message.type === "emit") { try { assertWorkflowJsonValue(message.value, "emit"); } catch (error) { finish({ error: new Error(`Workflow emit could not be persisted: ${error instanceof Error ? error.message : String(error)}`) }); return; } emits.push(message.value); try { options.onEmit?.([...emits]); } catch (error) { emits.pop(); finish({ error: new Error(`Workflow emit could not be persisted: ${error instanceof Error ? error.message : String(error)}`) }); } return; } if (message.type === "console") { const level = message.level; if ((level === "log" || level === "info" || level === "warn" || level === "error") && typeof message.text === "string") consoleEntries.push({ level, text: message.text }); return; } if (message.type === "complete") { try { assertWorkflowJsonValue(message.value, "return"); } catch (error) { return finish({ error: new Error(`Workflow return could not be persisted: ${error instanceof Error ? error.message : String(error)}`) }); } return finish({ value: message.value }); } if (message.type === "error") return finish({ error: new Error(typeof message.error === "string" ? message.error : "Workflow script failed.") }); if (message.type !== "call" || typeof message.callId !== "number" || typeof message.method !== "string" || !isRecord(message.args)) return; const respond = (promise: Promise) => { void promise.then( (value) => worker.postMessage({ type: "response", callId: message.callId, ok: true, value: omitUndefinedWorkflowValues(value) }), (error: unknown) => worker.postMessage({ type: "response", callId: message.callId, ok: false, error: error instanceof Error ? error.message : String(error) }), ); }; if (message.method === "status") { const keyOrRunId = message.args.keyOrRunId; if (typeof keyOrRunId !== "string" || !keyOrRunId.trim()) return respond(Promise.reject(new Error("runs.status(keyOrRunId) requires a non-empty string."))); const known = children.get(keyOrRunId); const target = known?.runId ?? keyOrRunId; trace.push({ operation: "status", key: keyOrRunId, state: "started", ...(known?.runId ? { runId: known.runId } : {}) }); traceChanged(); respond(options.status(target, childController.signal).then((result) => { trace.push({ operation: "status", key: keyOrRunId, state: result.ok ? "completed" : "failed", ...(result.runId ? { runId: result.runId } : {}), ...(!result.ok ? { error: result.output } : {}) }); traceChanged(); if (!result.ok) throw new Error(`Status '${keyOrRunId}' failed: ${result.output}`); return result; })); return; } if (message.method !== "run") return respond(Promise.reject(new Error(`Unknown runs API method '${message.method}'.`))); let key: string; try { key = validateKey(message.args.key); } catch (error) { return respond(Promise.reject(error)); } const params = message.args.params; if (!isRecord(params)) return respond(Promise.reject(new Error(`runs.run('${key}', params) requires a params object.`))); if (params.action !== undefined) return respond(Promise.reject(new Error(`runs.run('${key}') accepts execution params only; management action is not allowed.`))); if (params.workflowScript !== undefined) return respond(Promise.reject(new Error(`runs.run('${key}') cannot start a nested workflow script.`))); if (params.tasks !== undefined || params.chain !== undefined || params.concurrency !== undefined || params.chainDir !== undefined) { return respond(Promise.reject(new Error(`runs.run('${key}') accepts one child via { agent, task }; use runs.all(...) and JavaScript control flow for orchestration.`))); } if (params.worktree !== undefined && typeof params.worktree !== "boolean") { return respond(Promise.reject(new Error(`runs.run('${key}') worktree must be true or false.`))); } const collectFailure = message.args.collectFailure === true; const deliver = (promise: Promise) => collectFailure ? promise : promise.then((result) => { if (!result.ok) throw new Error(`Run '${key}' failed: ${result.error ?? result.output}`); return result; }); const fingerprint = stableJson(params); const existing = launches.get(key); if (existing) { if (existing.fingerprint !== fingerprint) return respond(Promise.reject(new Error(`Duplicate workflow key '${key}' used with incompatible launch params.`))); trace.push({ operation: "run", key, state: "reused", ...workflowStringMetadata(params) }); traceChanged(); return respond(deliver(existing.promise)); } const startedAt = Date.now(); childOrder.push(key); trace.push({ operation: "run", key, state: "started", ...workflowStringMetadata(params) }); traceChanged(); const promise = Promise.resolve().then(() => options.launch(key, { ...params, async: params.async ?? false }, childController.signal)).then((result) => { const normalized = !result.ok && !result.error ? { ...result, error: result.output } : result; children.set(key, normalized); trace.push({ operation: "run", key, state: normalized.ok ? "completed" : "failed", durationMs: Date.now() - startedAt, ...workflowStringMetadata(params), ...(normalized.runId ? { runId: normalized.runId } : {}), ...(!normalized.ok ? { error: normalized.error ?? normalized.output } : {}) }); traceChanged(); return normalized; }, (error: unknown) => { const text = error instanceof Error ? error.message : String(error); const failure: WorkflowScriptChildResult = { key, ok: false, output: text, error: text, artifactPaths: [] }; children.set(key, failure); trace.push({ operation: "run", key, state: "failed", durationMs: Date.now() - startedAt, ...workflowStringMetadata(params), error: text }); traceChanged(); return failure; }); launches.set(key, { fingerprint, promise }); respond(deliver(promise)); }); worker.postMessage({ type: "start", script: options.script }); }); }