diff --git a/skills/dynamic-workflows/SKILL.md b/skills/dynamic-workflows/SKILL.md index 87322cb8..b5e52363 100644 --- a/skills/dynamic-workflows/SKILL.md +++ b/skills/dynamic-workflows/SKILL.md @@ -64,6 +64,61 @@ return { summary, findings } **No `writeMode`.** Teach read-only vs write in the prompt. Use `isolation: 'worktree'` when parallel mutators would conflict (git required). +### Failure-aware orchestration + +`agent()` and nested `workflow()` throw a stable, provider-independent error: + +```js +{ + name: 'WorkflowScriptRuntimeError', + kind: 'provider_unavailable', + message: 'Claude is unavailable', + retryable: true, +} +``` + +Use normal JavaScript `try/catch` for fallback. Branch on `error.kind`, not on +provider-specific message text: + +```js +let review +try { + review = await agent('Read-only security review', { provider: 'claude' }) +} catch (error) { + if (error.kind !== 'provider_unavailable') throw error + review = await agent('Read-only security review', { provider: 'codex' }) +} +``` + +Default `parallel()` and `pipeline()` behavior remains unchanged: a thrown +branch/item becomes `null`. When failure details must survive fan-out, catch the +error inside each branch: + +```js +async function attempt(task) { + try { + return { ok: true, value: await task() } + } catch (error) { + return { + ok: false, + error: { + kind: error.kind, + message: error.message, + retryable: error.retryable, + }, + } + } +} + +const reviews = await parallel([ + () => attempt(() => agent('Security review', { provider: 'claude' })), + () => attempt(() => agent('Correctness review', { provider: 'codex' })), +]) +``` + +Failed calls remain journaled failures and are retried on resume; catching an +error does not turn that failed agent call into a replayable success. + ### Determinism bans `Date.now()`, `Math.random()`, and `new Date()` without args throw. Pass timestamps via `args` if needed. diff --git a/src/workflow-api.ts b/src/workflow-api.ts index 8ae13c92..251751bd 100644 --- a/src/workflow-api.ts +++ b/src/workflow-api.ts @@ -17,6 +17,11 @@ import { type WorkflowMeta, } from "./workflow-types.js"; import { agentOptsSchema } from "./workflow-contracts.js"; +import { + isWorkflowOperationError, + serializeWorkflowError, + WorkflowScriptRuntimeError, +} from "./workflow-errors.js"; // --------------------------------------------------------------------------- // Host deps (injected by engine; fakes OK in tests) @@ -237,7 +242,7 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { const semaphore = new WorkflowSemaphore(Math.max(1, deps.concurrency)); let callIndex = 0; - const agent = async (prompt: unknown, opts: unknown = {}): Promise => { + const runAgent = async (prompt: unknown, opts: unknown = {}): Promise => { if (typeof prompt !== "string" || !prompt.trim()) { throw new WorkflowEngineError("internal", "agent(prompt) requires a non-empty string"); } @@ -493,11 +498,12 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { } } if (agentCallBegun) { + const scriptError = toWorkflowScriptRuntimeError(error); deps.journal.failAgentCall({ runId: deps.runId, callIndex: index, error: message, - errorKind: error instanceof WorkflowEngineError ? error.kind : "internal", + errorKind: scriptError.kind, worktreePath, }); } @@ -520,6 +526,14 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { } }; + const agent = async (prompt: unknown, opts: unknown = {}): Promise => { + try { + return await runAgent(prompt, opts); + } catch (error) { + throw toWorkflowScriptRuntimeError(error); + } + }; + const parallel = async (...args: unknown[]): Promise> => { const thunks = args[0]; if (!Array.isArray(thunks)) { @@ -599,29 +613,36 @@ export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { }; const workflow = async (...args: unknown[]): Promise => { - if (nestDepth >= WORKFLOW_MAX_NEST_DEPTH) { - throw new WorkflowEngineError( - "nest_depth", - `workflow() nesting limited to ${WORKFLOW_MAX_NEST_DEPTH} level`, - ); - } - if (!deps.resolveNestedSource || !deps.executeNested) { - throw new WorkflowEngineError("internal", "nested workflow() is not configured on this host"); - } - const nameOrRef = args[0] as string | { scriptPath: string }; - const childArgsResult = jsonValueSchema.optional().safeParse(args[1]); - if (!childArgsResult.success) { - throw new WorkflowEngineError( - "internal", - `workflow() args must be JSON-serializable: ${childArgsResult.error.issues[0]?.message ?? "invalid value"}`, - ); + try { + if (nestDepth >= WORKFLOW_MAX_NEST_DEPTH) { + throw new WorkflowEngineError( + "nest_depth", + `workflow() nesting limited to ${WORKFLOW_MAX_NEST_DEPTH} level`, + ); + } + if (!deps.resolveNestedSource || !deps.executeNested) { + throw new WorkflowEngineError( + "internal", + "nested workflow() is not configured on this host", + ); + } + const nameOrRef = args[0] as string | { scriptPath: string }; + const childArgsResult = jsonValueSchema.optional().safeParse(args[1]); + if (!childArgsResult.success) { + throw new WorkflowEngineError( + "internal", + `workflow() args must be JSON-serializable: ${childArgsResult.error.issues[0]?.message ?? "invalid value"}`, + ); + } + const source = await deps.resolveNestedSource(nameOrRef); + return await deps.executeNested({ + source, + args: childArgsResult.data, + nestDepth: nestDepth + 1, + }); + } catch (error) { + throw toWorkflowScriptRuntimeError(error); } - const source = await deps.resolveNestedSource(nameOrRef); - return deps.executeNested({ - source, - args: childArgsResult.data, - nestDepth: nestDepth + 1, - }); }; return { @@ -645,6 +666,43 @@ function formatReplayMiss(miss: WorkflowReplayMiss): string { : miss.reason; } +export function toWorkflowScriptRuntimeError( + error: unknown, +): WorkflowScriptRuntimeError { + if (error instanceof WorkflowScriptRuntimeError) return error; + if (error instanceof WorkflowEngineError) { + return new WorkflowScriptRuntimeError({ + kind: error.kind, + message: error.message, + retryable: + error.kind === "provider_unavailable" || + error.kind === "provider_disabled" || + error.kind === "no_provider", + }); + } + if (isWorkflowOperationError(error)) { + const serialized = serializeWorkflowError(error); + return new WorkflowScriptRuntimeError({ + kind: serialized.kind, + message: serialized.message, + retryable: serialized.retryable, + }); + } + if (error && typeof error === "object" && "name" in error) { + const name = String((error as { name: unknown }).name); + if (name === "AbortError") { + return new WorkflowScriptRuntimeError({ + kind: "cancelled", + message: "Workflow cancelled", + }); + } + } + return new WorkflowScriptRuntimeError({ + kind: "internal", + message: error instanceof Error ? error.message : String(error), + }); +} + /** Test helper: read current ALS phase (undefined outside phase). */ export function getCurrentWorkflowPhase(): string | undefined { return phaseAls.getStore(); diff --git a/src/workflow-contracts.ts b/src/workflow-contracts.ts index 1944b1c5..f15f8c74 100644 --- a/src/workflow-contracts.ts +++ b/src/workflow-contracts.ts @@ -129,6 +129,12 @@ export const workflowErrorKindSchema = z.enum([ ]); export type WorkflowErrorKind = z.infer; +/** Stable provider-independent error shape exposed to workflow scripts. */ +export interface WorkflowScriptThrownError extends Error { + kind: WorkflowErrorKind; + retryable: boolean; +} + export const WORKFLOW_EVENT_TYPES = [ "run_started", "run_completed", diff --git a/src/workflow-engine.test.ts b/src/workflow-engine.test.ts index 5c3438b3..7d563519 100644 --- a/src/workflow-engine.test.ts +++ b/src/workflow-engine.test.ts @@ -6,13 +6,17 @@ import { WorkflowStore } from "./workflow-store.js"; import { executeWorkflow } from "./workflow-engine.js"; import { createWorkflowApi, - WorkflowEngineError, WorkflowSemaphore, getCurrentWorkflowPhase, type WorkflowProviderRunInput, type CreateAgentWorktree, } from "./workflow-api.js"; import { createStubBudget } from "./workflow-types.js"; +import { WorkflowScriptRuntimeError } from "./workflow-errors.js"; +import { + ProviderExecutionError, + ProviderUnavailableError, +} from "./local-agent-errors.js"; // --------------------------------------------------------------------------- // Semaphore @@ -77,6 +81,101 @@ import { createStubBudget } from "./workflow-types.js"; await rm(dir, { recursive: true, force: true }); } +// --------------------------------------------------------------------------- +// Script-visible error contract: try/catch, provider fallback, parallel helper +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-script-errors-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "script-errors", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + + const { result } = await executeWorkflow({ + source: ` +export const meta = { name: 'script-errors', description: 'd' } + +async function attempt(task) { + try { + return { ok: true, value: await task() } + } catch (error) { + return { + ok: false, + error: { + name: error.name, + kind: error.kind, + message: error.message, + retryable: error.retryable, + }, + } + } +} + +let primary +try { + primary = await agent('primary', { provider: 'claude' }) +} catch (error) { + if (error.kind !== 'provider_unavailable') throw error + primary = await agent('fallback', { provider: 'codex' }) +} + +const reviews = await parallel([ + () => attempt(() => agent('retryable-failure', { provider: 'codex' })), + () => attempt(() => agent('success', { provider: 'codex' })), +]) + +return { primary, reviews } +`, + runId: run.id, + journal: store, + workspaceRoot: dir, + enabledProviders: ["claude", "codex"], + runProvider: async (input) => { + if (input.prompt === "primary") { + throw new ProviderUnavailableError("claude", "Claude is unavailable"); + } + if (input.prompt === "retryable-failure") { + throw new ProviderExecutionError({ + provider: "codex", + cause: new Error("temporary provider failure"), + retryable: true, + }); + } + return { finalResponse: `ok:${input.prompt}` }; + }, + }); + + assert.deepEqual(result, { + primary: "ok:fallback", + reviews: [ + { + ok: false, + error: { + name: "WorkflowScriptRuntimeError", + kind: "provider", + message: "codex agent execution failed: temporary provider failure", + retryable: true, + }, + }, + { ok: true, value: "ok:success" }, + ], + }); + + const calls = store.listAgentCalls(run.id); + assert.equal(calls.length, 4); + assert.equal(calls[0]?.status, "failed"); + assert.equal(calls[0]?.errorKind, "provider_unavailable"); + assert.equal(calls[2]?.status, "failed"); + assert.equal(calls[2]?.errorKind, "provider"); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + // --------------------------------------------------------------------------- // pipeline — no barrier across items (item B can finish stage2 before A stage1 ends) // --------------------------------------------------------------------------- @@ -455,7 +554,7 @@ return await workflow({ scriptPath: 'x' }) `, }), (error: unknown) => - error instanceof WorkflowEngineError && error.kind === "nest_depth", + error instanceof WorkflowScriptRuntimeError && error.kind === "nest_depth", ); store.close(); @@ -492,7 +591,7 @@ return await workflow({ scriptPath: 'x' }) }); // abort before agent ac.abort(); - await assert.rejects(async () => api.agent("x"), WorkflowEngineError); + await assert.rejects(async () => api.agent("x"), WorkflowScriptRuntimeError); store.close(); await rm(dir, { recursive: true, force: true }); } diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts index 1611e7ca..74538d1b 100644 --- a/src/workflow-engine.ts +++ b/src/workflow-engine.ts @@ -20,6 +20,7 @@ import { } from "./workflow-types.js"; import { isWorkflowOperationError, + WorkflowScriptRuntimeError, workflowErrorKind, } from "./workflow-errors.js"; @@ -186,6 +187,9 @@ async function executeNestedOnApi(input: { const WORKFLOW_MAX_NEST_DEPTH_LOCAL = 1; export function mapEngineErrorKind(error: unknown): WorkflowErrorKind { + if (error instanceof WorkflowScriptRuntimeError) { + return error.kind; + } if (error instanceof WorkflowEngineError) { return error.kind; } diff --git a/src/workflow-errors.ts b/src/workflow-errors.ts index 4349cc5f..65a5d0e5 100644 --- a/src/workflow-errors.ts +++ b/src/workflow-errors.ts @@ -8,8 +8,29 @@ import { import type { WorkflowErrorKind, WorkflowRunStatus, + WorkflowScriptThrownError, } from "./workflow-types.js"; +/** Public runtime error crossing from the host into workflow JavaScript. */ +export class WorkflowScriptRuntimeError + extends Error + implements WorkflowScriptThrownError +{ + readonly kind: WorkflowErrorKind; + readonly retryable: boolean; + + constructor(input: { + kind: WorkflowErrorKind; + message: string; + retryable?: boolean; + }) { + super(input.message); + this.name = "WorkflowScriptRuntimeError"; + this.kind = input.kind; + this.retryable = input.retryable ?? false; + } +} + export class InvalidWorkflowInputError extends TaggedError( "InvalidWorkflowInputError", )<{ diff --git a/src/workflow-tools.ts b/src/workflow-tools.ts index 5f8c2c6c..5647249a 100644 --- a/src/workflow-tools.ts +++ b/src/workflow-tools.ts @@ -39,6 +39,7 @@ const WORKFLOW_API_CHEATSHEET = ` Workflow scripts (JS only): export const meta = { name, description, phases?, defaultProvider?, concurrency? } agent(prompt, { label?, phase?, schema?, model?, effort?, provider?, isolation?: 'worktree' }) + throws { name: 'WorkflowScriptRuntimeError', kind, message, retryable } parallel(thunks) → Array // barrier; throw → null pipeline(items, ...stages) // no cross-item barrier phase(title); log(msg); args diff --git a/src/workflow-types.ts b/src/workflow-types.ts index ad050e1c..1189e3de 100644 --- a/src/workflow-types.ts +++ b/src/workflow-types.ts @@ -40,6 +40,7 @@ export type { WorkflowPipeline, WorkflowRunSource, WorkflowRunStatus, + WorkflowScriptThrownError, WorkflowTask, } from "./workflow-contracts.js";