From 0d9bd46cfc160a9e13d4fb9d3ee55f7d978dc800 Mon Sep 17 00:00:00 2001 From: Michael Livshits Date: Sun, 31 May 2026 09:43:13 +0300 Subject: [PATCH] Guard workflow results against Promise clone failures --- src/workflow.ts | 70 ++++++++++++++++++++++++++++------ tests/workflow-runtime.test.ts | 46 ++++++++++++++++++++++ 2 files changed, 104 insertions(+), 12 deletions(-) diff --git a/src/workflow.ts b/src/workflow.ts index 2b7bd8d..8d66598 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -72,6 +72,7 @@ export async function runWorkflow( Math.min(options.concurrency ?? Math.max(1, (globalThis.navigator?.hardwareConcurrency ?? 8) - 2), 16), ); const limiter = createLimiter(concurrency); + const pendingAgentRuns = new Set>(); const log = (message: string) => { const text = String(message); @@ -79,10 +80,11 @@ export async function runWorkflow( options.onLog?.(text); }; - const phase = (title: string) => { - state.currentPhase = title; - if (!state.phases.includes(title)) state.phases.push(title); - options.onPhase?.(title); + const phase = (title: unknown) => { + const text = requireString(title, "phase title"); + state.currentPhase = text; + if (!state.phases.includes(text)) state.phases.push(text); + options.onPhase?.(text); }; const budget = Object.freeze({ @@ -95,22 +97,24 @@ export async function runWorkflow( if (options.signal?.aborted) throw new Error("workflow aborted"); }; - const agent = async (prompt: string, agentOptions: AgentOptions = {}) => { + const agent = async (prompt: unknown, agentOptions: unknown = {}) => { throwIfAborted(); if (budget.total !== null && budget.remaining() <= 0) throw new Error("workflow token budget exhausted"); - const assignedPhase = agentOptions.phase ?? state.currentPhase; - const requestedLabel = agentOptions.label?.trim(); - return limiter(async () => { + const taskPrompt = requireString(prompt, "agent prompt"); + const normalizedOptions = normalizeAgentOptions(agentOptions); + const assignedPhase = normalizedOptions.phase ?? state.currentPhase; + const requestedLabel = normalizedOptions.label?.trim(); + const run = limiter(async () => { state.agentCount++; const label = requestedLabel || defaultAgentLabel(assignedPhase, state.agentCount); - options.onAgentStart?.({ label, phase: assignedPhase, prompt }); + options.onAgentStart?.({ label, phase: assignedPhase, prompt: taskPrompt }); try { throwIfAborted(); - const result = await agentRunner.run(prompt, { + const result = await agentRunner.run(taskPrompt, { label, - schema: agentOptions.schema, + schema: normalizedOptions.schema, signal: options.signal, - instructions: buildAgentInstructions(assignedPhase, agentOptions), + instructions: buildAgentInstructions(assignedPhase, normalizedOptions), } as any); throwIfAborted(); state.spent += estimateTokens(result); @@ -123,6 +127,12 @@ export async function runWorkflow( return null; } }); + pendingAgentRuns.add(run); + run.then( + () => pendingAgentRuns.delete(run), + () => pendingAgentRuns.delete(run), + ); + return run; }; const parallel = async (thunks: Array<() => Promise>) => { @@ -202,6 +212,8 @@ export async function runWorkflow( const wrapped = `(async () => {\n${body}\n})()`; const result = await new vm.Script(wrapped, { filename: `${meta.name || "workflow"}.js` }).runInContext(context); + await Promise.allSettled([...pendingAgentRuns]); + assertStructuredCloneable(result, "workflow result"); return { meta, result: result as T, @@ -334,6 +346,40 @@ function createLimiter(limit: number) { }; } +function requireString(value: unknown, name: string): string { + if (typeof value !== "string") throw new TypeError(`${name} must be a string`); + return value; +} + +function optionalString(value: unknown, name: string): string | undefined { + if (value === undefined) return undefined; + return requireString(value, name); +} + +function normalizeAgentOptions(value: unknown): AgentOptions { + if (!value || typeof value !== "object") throw new TypeError("agent options must be an object"); + const options = value as AgentOptions; + return { + ...options, + label: optionalString(options.label, "agent label"), + phase: optionalString(options.phase, "agent phase"), + model: optionalString(options.model, "agent model"), + isolation: options.isolation, + agentType: optionalString(options.agentType, "agent type"), + }; +} + +function assertStructuredCloneable(value: unknown, name: string): void { + try { + structuredClone(value); + } catch (error) { + const detail = error instanceof Error ? ` ${error.message}` : ""; + throw new Error( + `${name} must be structured-cloneable; did you forget to await agent(), parallel(), or pipeline()?${detail}`, + ); + } +} + function defaultAgentLabel(phase: string | undefined, index: number): string { return phase ? `${phase} agent ${index}` : `agent ${index}`; } diff --git a/tests/workflow-runtime.test.ts b/tests/workflow-runtime.test.ts index d8677b9..db7300e 100644 --- a/tests/workflow-runtime.test.ts +++ b/tests/workflow-runtime.test.ts @@ -56,3 +56,49 @@ return { ok: true } assert.deepEqual(result.phases, ["Inspect API", "Inspect UI"]); assert.equal(result.agentCount, 2); }); + +test("runWorkflow rejects unawaited nested agent promises before returning details", async () => { + let ended = 0; + + await assert.rejects( + () => + runWorkflow( + `export const meta = { + name: 'promise_leak', + description: 'Return an unawaited agent promise' +} + +phase('Leak promise') +const scan = agent('scan', { label: 'scan' }) +return { scan } +`, + { + agent: fakeAgent, + onAgentEnd() { + ended++; + }, + }, + ), + /workflow result must be structured-cloneable; did you forget to await agent\(\), parallel\(\), or pipeline\(\)\?.*Promise.*cloned/, + ); + + assert.equal(ended, 1); +}); + +test("runWorkflow rejects non-string runtime phase titles", async () => { + await assert.rejects( + () => + runWorkflow( + `export const meta = { + name: 'bad_phase', + description: 'Use a non-string phase title' +} + +phase(Promise.resolve('Scan')) +return { ok: true } +`, + { agent: fakeAgent }, + ), + /phase title must be a string/, + ); +});