From 775f7f049f4945417c1fab7a95902415149dca04 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Tue, 21 Jul 2026 17:19:50 +0530 Subject: [PATCH] feat(workflow): implement engine API with fakes agent/parallel/pipeline/phase/log/workflow primitives, semaphore, ALS phase, isolation worktree hook, nest depth 1, and executeWorkflow against injectable runProvider for unit tests. Co-Authored-By: Claude --- package.json | 2 +- src/workflow-api.ts | 675 ++++++++++++++++++++++++++++++++++++ src/workflow-engine.test.ts | 402 +++++++++++++++++++++ src/workflow-engine.ts | 206 +++++++++++ 4 files changed, 1284 insertions(+), 1 deletion(-) create mode 100644 src/workflow-api.ts create mode 100644 src/workflow-engine.test.ts create mode 100644 src/workflow-engine.ts diff --git a/package.json b/package.json index bb371809..d5c880d0 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,7 @@ "dev": "node scripts/dev-server.mjs", "postinstall": "node scripts/fix-node-pty-permissions.mjs", "start": "node dist/cli.js serve", - "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts", + "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts && tsx src/workflow-types.test.ts && tsx src/workflow-store.test.ts && tsx src/workflow-script.test.ts && tsx src/workflow-sandbox.test.ts && tsx src/workflow-engine.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], diff --git a/src/workflow-api.ts b/src/workflow-api.ts new file mode 100644 index 00000000..05249df1 --- /dev/null +++ b/src/workflow-api.ts @@ -0,0 +1,675 @@ +import { AsyncLocalStorage } from "node:async_hooks"; +import { createHash } from "node:crypto"; +import type { WorkflowSandboxApi } from "./workflow-sandbox.js"; +import { + WORKFLOW_LIMITS, + WORKFLOW_MAX_ITEMS, + WORKFLOW_MAX_NEST_DEPTH, + buildAgentCacheKeyInput, + createStubBudget, + type AgentIsolationMode, + type AgentOpts, + type WorkflowMeta, +} from "./workflow-types.js"; + +// --------------------------------------------------------------------------- +// Host deps (injected by engine; fakes OK in tests) +// --------------------------------------------------------------------------- + +export interface WorkflowProviderRunInput { + provider: string; + prompt: string; + model?: string; + effort?: string; + workspace: string; + signal?: AbortSignal; + label?: string; + phase?: string; +} + +export interface WorkflowProviderRunResult { + finalResponse: string; + providerSessionId?: string; +} + +export type WorkflowRunProvider = ( + input: WorkflowProviderRunInput, +) => Promise; + +export interface WorkflowWorktreeHandle { + path: string; + /** Called after agent returns or fails. Success+clean may remove; dirty/failure preserves. */ + finalize: (outcome: "success" | "failure") => Promise<{ dirty: boolean; removed: boolean }>; +} + +export type CreateAgentWorktree = (input: { + runId: string; + callIndex: number; + workspaceRoot: string; + baseSha?: string; +}) => Promise; + +export interface WorkflowReplayHit { + value: unknown; + responseText?: string; + structuredJson?: string; + providerSessionId?: string; +} + +export interface WorkflowReplay { + match(callIndex: number, cacheKey: string): WorkflowReplayHit | null; +} + +export interface WorkflowJournal { + appendEvent(input: { + runId: string; + type: + | "phase_started" + | "log" + | "agent_call_started" + | "agent_call_completed" + | "agent_call_failed" + | "agent_call_cached" + | "schema_retry" + | "worktree_created" + | "worktree_finalized"; + phase?: string; + label?: string; + data?: unknown; + }): unknown; + beginAgentCall(input: { + runId: string; + callIndex: number; + cacheKey: string; + provider: string; + model?: string; + effort?: string; + label?: string; + phase?: string; + isolation?: AgentIsolationMode; + worktreePath?: string; + }): unknown; + completeAgentCall(input: { + runId: string; + callIndex: number; + responseText?: string; + structuredJson?: string; + providerSessionId?: string; + dirty?: boolean; + worktreePath?: string; + fromCache?: boolean; + }): unknown; + failAgentCall(input: { + runId: string; + callIndex: number; + error: string; + worktreePath?: string; + dirty?: boolean; + }): unknown; + isCancelRequested(runId: string): boolean; +} + +export interface WorkflowApiDeps { + runId: string; + journal: WorkflowJournal; + meta: WorkflowMeta; + args: unknown; + concurrency: number; + signal: AbortSignal; + workspaceRoot: string; + baseSha?: string; + /** Already-filtered enabled ∩ live provider ids, preference order. */ + enabledProviders: string[]; + runProvider: WorkflowRunProvider; + createWorktree?: CreateAgentWorktree; + replay?: WorkflowReplay; + /** Nested workflow source loader; required for workflow(). */ + resolveNestedSource?: (nameOrRef: string | { scriptPath: string }) => string | Promise; + /** Run a nested script sharing semaphore/callIndex. */ + executeNested?: (input: { + source: string; + args: unknown; + nestDepth: number; + }) => Promise; + nestDepth?: number; +} + +export interface WorkflowApi extends WorkflowSandboxApi { + getCallCount(): number; + getNestDepth(): number; +} + +export class WorkflowEngineError extends Error { + constructor( + readonly kind: + | "cancelled" + | "provider_disabled" + | "provider_unavailable" + | "no_provider" + | "nest_depth" + | "worktree" + | "schema" + | "path" + | "internal", + message: string, + ) { + super(message); + this.name = "WorkflowEngineError"; + } +} + +// --------------------------------------------------------------------------- +// Semaphore +// --------------------------------------------------------------------------- + +export class WorkflowSemaphore { + private active = 0; + private readonly waiters: Array<() => void> = []; + + constructor(readonly limit: number) { + if (!Number.isFinite(limit) || limit < 1) { + throw new Error("WorkflowSemaphore limit must be >= 1"); + } + } + + async acquire(signal?: AbortSignal): Promise { + if (signal?.aborted) throw cancelledError(); + if (this.active < this.limit) { + this.active += 1; + return; + } + await new Promise((resolve, reject) => { + const onAbort = () => { + const idx = this.waiters.indexOf(wake); + if (idx >= 0) this.waiters.splice(idx, 1); + reject(cancelledError()); + }; + const wake = () => { + signal?.removeEventListener("abort", onAbort); + this.active += 1; + resolve(); + }; + this.waiters.push(wake); + signal?.addEventListener("abort", onAbort, { once: true }); + }); + } + + release(): void { + this.active = Math.max(0, this.active - 1); + const next = this.waiters.shift(); + if (next) next(); + } +} + +// --------------------------------------------------------------------------- +// API factory +// --------------------------------------------------------------------------- + +const phaseAls = new AsyncLocalStorage(); + +export function createWorkflowApi(deps: WorkflowApiDeps): WorkflowApi { + const nestDepth = deps.nestDepth ?? 0; + const semaphore = new WorkflowSemaphore(Math.max(1, deps.concurrency)); + let callIndex = 0; + + const agent = async (prompt: unknown, opts: unknown = {}): Promise => { + if (typeof prompt !== "string" || !prompt.trim()) { + throw new WorkflowEngineError("internal", "agent(prompt) requires a non-empty string"); + } + const agentOpts = normalizeAgentOpts(opts); + throwIfCancelled(deps); + + const provider = resolveProvider(agentOpts.provider, deps.meta, deps.enabledProviders); + const phase = agentOpts.phase ?? phaseAls.getStore(); + const isolation: AgentIsolationMode = + agentOpts.isolation === "worktree" ? "worktree" : "shared"; + const index = callIndex; + callIndex += 1; + + const cacheKeyInput = buildAgentCacheKeyInput({ + prompt, + provider, + model: agentOpts.model, + effort: agentOpts.effort, + schema: agentOpts.schema, + isolation, + }); + const cacheKey = hashCacheKey(cacheKeyInput); + + if (deps.replay) { + const hit = deps.replay.match(index, cacheKey); + if (hit) { + deps.journal.beginAgentCall({ + runId: deps.runId, + callIndex: index, + cacheKey, + provider, + model: agentOpts.model, + effort: agentOpts.effort, + label: agentOpts.label, + phase, + isolation, + }); + deps.journal.completeAgentCall({ + runId: deps.runId, + callIndex: index, + responseText: hit.responseText, + structuredJson: hit.structuredJson, + providerSessionId: hit.providerSessionId, + fromCache: true, + }); + deps.journal.appendEvent({ + runId: deps.runId, + type: "agent_call_cached", + phase, + label: agentOpts.label, + data: { callIndex: index, cacheKey, provider }, + }); + return hit.value; + } + } + + await semaphore.acquire(deps.signal); + let worktree: WorkflowWorktreeHandle | null = null; + let worktreePath: string | undefined; + try { + throwIfCancelled(deps); + + if (isolation === "worktree") { + if (!deps.createWorktree) { + throw new WorkflowEngineError( + "worktree", + "isolation: 'worktree' requires createWorktree host support", + ); + } + worktree = await deps.createWorktree({ + runId: deps.runId, + callIndex: index, + workspaceRoot: deps.workspaceRoot, + baseSha: deps.baseSha, + }); + worktreePath = worktree.path; + deps.journal.appendEvent({ + runId: deps.runId, + type: "worktree_created", + phase, + label: agentOpts.label, + data: { callIndex: index, worktreePath, isolation }, + }); + } + + deps.journal.beginAgentCall({ + runId: deps.runId, + callIndex: index, + cacheKey, + provider, + model: agentOpts.model, + effort: agentOpts.effort, + label: agentOpts.label, + phase, + isolation, + worktreePath, + }); + deps.journal.appendEvent({ + runId: deps.runId, + type: "agent_call_started", + phase, + label: agentOpts.label, + data: { + callIndex: index, + cacheKey, + provider, + isolation, + worktreePath, + }, + }); + + const cwd = worktreePath ?? deps.workspaceRoot; + const result = await deps.runProvider({ + provider, + prompt, + model: agentOpts.model, + effort: agentOpts.effort, + workspace: cwd, + signal: deps.signal, + label: agentOpts.label, + phase, + }); + + throwIfCancelled(deps); + + let returnValue: unknown = result.finalResponse; + let structuredJson: string | undefined; + if (agentOpts.schema) { + // Full Ajv enforcement lands in M6; for now extract JSON object when schema set. + const extracted = tryExtractJson(result.finalResponse); + if (extracted === undefined) { + throw new WorkflowEngineError( + "schema", + "agent() schema set but response was not valid JSON", + ); + } + returnValue = extracted; + structuredJson = JSON.stringify(extracted); + } + + let dirty: boolean | undefined; + if (worktree) { + const finalized = await worktree.finalize("success"); + dirty = finalized.dirty; + deps.journal.appendEvent({ + runId: deps.runId, + type: "worktree_finalized", + phase, + label: agentOpts.label, + data: { + callIndex: index, + worktreePath, + dirty: finalized.dirty, + removed: finalized.removed, + }, + }); + worktree = null; + } + + deps.journal.completeAgentCall({ + runId: deps.runId, + callIndex: index, + responseText: truncate(result.finalResponse, WORKFLOW_LIMITS.responseTextBytes), + structuredJson: structuredJson + ? truncate(structuredJson, WORKFLOW_LIMITS.structuredJsonBytes) + : undefined, + providerSessionId: result.providerSessionId, + dirty, + worktreePath, + }); + deps.journal.appendEvent({ + runId: deps.runId, + type: "agent_call_completed", + phase, + label: agentOpts.label, + data: { + callIndex: index, + provider, + isolation, + worktreePath, + dirty, + fromCache: false, + }, + }); + return returnValue; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (worktree) { + try { + const finalized = await worktree.finalize("failure"); + deps.journal.appendEvent({ + runId: deps.runId, + type: "worktree_finalized", + phase, + label: agentOpts.label, + data: { + callIndex: index, + worktreePath, + dirty: finalized.dirty, + removed: finalized.removed, + outcome: "failure", + }, + }); + } catch { + // preserve original error + } + } + deps.journal.failAgentCall({ + runId: deps.runId, + callIndex: index, + error: message, + worktreePath, + }); + deps.journal.appendEvent({ + runId: deps.runId, + type: "agent_call_failed", + phase, + label: agentOpts.label, + data: { callIndex: index, error: message, isolation, worktreePath }, + }); + throw error; + } finally { + semaphore.release(); + } + }; + + const parallel = async (...args: unknown[]): Promise> => { + const thunks = args[0]; + if (!Array.isArray(thunks)) { + throw new WorkflowEngineError("internal", "parallel(thunks) requires an array of functions"); + } + assertMaxItems(thunks.length, "parallel"); + return Promise.all( + thunks.map(async (thunk, index) => { + if (typeof thunk !== "function") { + throw new WorkflowEngineError( + "internal", + `parallel thunks[${index}] must be a function`, + ); + } + try { + return await (thunk as () => Promise)(); + } catch { + return null; + } + }), + ); + }; + + const pipeline = async (...args: unknown[]): Promise> => { + const items = args[0]; + const stages = args.slice(1); + if (!Array.isArray(items)) { + throw new WorkflowEngineError("internal", "pipeline(items, ...stages) requires an items array"); + } + assertMaxItems(items.length, "pipeline"); + for (let i = 0; i < stages.length; i += 1) { + if (typeof stages[i] !== "function") { + throw new WorkflowEngineError("internal", `pipeline stage[${i}] must be a function`); + } + } + return Promise.all( + items.map(async (item, index) => { + let prev: unknown = item; + for (const stage of stages) { + try { + prev = await (stage as (prev: unknown, item: unknown, index: number) => unknown)( + prev, + item, + index, + ); + } catch { + return null; + } + } + return prev; + }), + ); + }; + + const phase = (...args: unknown[]): void => { + const title = args[0]; + if (typeof title !== "string" || !title.trim()) { + throw new WorkflowEngineError("internal", "phase(title) requires a non-empty string"); + } + phaseAls.enterWith(title); + deps.journal.appendEvent({ + runId: deps.runId, + type: "phase_started", + phase: title, + data: { title }, + }); + }; + + const log = (...args: unknown[]): void => { + const message = args.map(String).join(" "); + deps.journal.appendEvent({ + runId: deps.runId, + type: "log", + phase: phaseAls.getStore(), + data: { message: truncate(message, WORKFLOW_LIMITS.eventDataJsonBytes) }, + }); + }; + + 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 childArgs = args[1]; + const source = await deps.resolveNestedSource(nameOrRef); + return deps.executeNested({ + source, + args: childArgs, + nestDepth: nestDepth + 1, + }); + }; + + return { + agent: agent as WorkflowSandboxApi["agent"], + parallel: parallel as WorkflowSandboxApi["parallel"], + pipeline: pipeline as WorkflowSandboxApi["pipeline"], + phase: phase as WorkflowSandboxApi["phase"], + log: log as WorkflowSandboxApi["log"], + args: deps.args, + budget: createStubBudget(), + workflow: workflow as WorkflowSandboxApi["workflow"], + meta: deps.meta, + getCallCount: () => callIndex, + getNestDepth: () => nestDepth, + }; +} + +/** Test helper: read current ALS phase (undefined outside phase). */ +export function getCurrentWorkflowPhase(): string | undefined { + return phaseAls.getStore(); +} + +export function hashCacheKey(input: ReturnType): string { + return createHash("sha256").update(JSON.stringify(input)).digest("hex"); +} + +export function resolveProvider( + optsProvider: string | undefined, + meta: WorkflowMeta, + enabledProviders: string[], +): string { + if (optsProvider) { + if (!enabledProviders.includes(optsProvider)) { + throw new WorkflowEngineError( + "provider_disabled", + `Provider ${optsProvider} is not enabled or not available`, + ); + } + return optsProvider; + } + if (meta.defaultProvider) { + if (!enabledProviders.includes(meta.defaultProvider)) { + throw new WorkflowEngineError( + "provider_unavailable", + `meta.defaultProvider ${meta.defaultProvider} is not enabled or not available`, + ); + } + return meta.defaultProvider; + } + const first = enabledProviders[0]; + if (!first) { + throw new WorkflowEngineError("no_provider", "No agent providers enabled"); + } + return first; +} + +function normalizeAgentOpts(opts: unknown): AgentOpts { + if (opts === undefined || opts === null) return {}; + if (typeof opts !== "object" || Array.isArray(opts)) { + throw new WorkflowEngineError("internal", "agent opts must be an object"); + } + const record = opts as Record; + const out: AgentOpts = {}; + if (typeof record.label === "string") out.label = record.label; + if (typeof record.phase === "string") out.phase = record.phase; + if (record.schema !== undefined) { + if (!record.schema || typeof record.schema !== "object" || Array.isArray(record.schema)) { + throw new WorkflowEngineError("schema", "agent opts.schema must be an object"); + } + out.schema = record.schema as object; + } + if (typeof record.model === "string") out.model = record.model; + if (typeof record.effort === "string") out.effort = record.effort; + if (typeof record.provider === "string") out.provider = record.provider; + if (record.isolation !== undefined) { + if (record.isolation !== "worktree") { + throw new WorkflowEngineError("worktree", 'agent opts.isolation must be "worktree" when set'); + } + out.isolation = "worktree"; + } + if ("writeMode" in record) { + throw new WorkflowEngineError("internal", "writeMode is not supported on agent() (v1)"); + } + return out; +} + +function assertMaxItems(count: number, label: string): void { + if (count > WORKFLOW_MAX_ITEMS) { + throw new WorkflowEngineError( + "internal", + `${label} exceeds max items ${WORKFLOW_MAX_ITEMS} (got ${count})`, + ); + } +} + +function throwIfCancelled(deps: WorkflowApiDeps): void { + if (deps.signal.aborted || deps.journal.isCancelRequested(deps.runId)) { + throw cancelledError(); + } +} + +function cancelledError(): WorkflowEngineError { + return new WorkflowEngineError("cancelled", "Workflow cancelled"); +} + +function truncate(text: string, maxBytes: number): string { + if (Buffer.byteLength(text, "utf8") <= maxBytes) return text; + // rough char truncate for journal safety + let end = Math.min(text.length, maxBytes); + while (end > 0 && Buffer.byteLength(text.slice(0, end), "utf8") > maxBytes) end -= 1; + return `${text.slice(0, end)}…`; +} + +/** Minimal JSON extract for schema path until Ajv module lands. */ +export function tryExtractJson(text: string): unknown | undefined { + const trimmed = text.trim(); + try { + return JSON.parse(trimmed); + } catch { + // strip fenced block + const fence = trimmed.match(/```(?:json)?\s*([\s\S]*?)```/i); + if (fence?.[1]) { + try { + return JSON.parse(fence[1].trim()); + } catch { + // fall through + } + } + const start = trimmed.search(/[{\[]/); + if (start < 0) return undefined; + const slice = trimmed.slice(start); + try { + return JSON.parse(slice); + } catch { + return undefined; + } + } +} diff --git a/src/workflow-engine.test.ts b/src/workflow-engine.test.ts new file mode 100644 index 00000000..ad83aedb --- /dev/null +++ b/src/workflow-engine.test.ts @@ -0,0 +1,402 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm, writeFile, mkdir } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +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"; + +// --------------------------------------------------------------------------- +// Semaphore +// --------------------------------------------------------------------------- +{ + const sem = new WorkflowSemaphore(2); + let concurrent = 0; + let maxConcurrent = 0; + await Promise.all( + Array.from({ length: 6 }, async () => { + await sem.acquire(); + concurrent += 1; + maxConcurrent = Math.max(maxConcurrent, concurrent); + await new Promise((r) => setTimeout(r, 20)); + concurrent -= 1; + sem.release(); + }), + ); + assert.equal(maxConcurrent, 2); +} + +// --------------------------------------------------------------------------- +// parallel → null on throw; barrier +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-engine-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "par", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const order: string[] = []; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "par", description: "d" }, + args: undefined, + concurrency: 4, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async (input) => { + order.push(`start:${input.prompt}`); + await new Promise((r) => setTimeout(r, 10)); + order.push(`end:${input.prompt}`); + if (input.prompt === "fail") throw new Error("boom"); + return { finalResponse: `ok:${input.prompt}` }; + }, + }); + + const results = await api.parallel([ + () => api.agent("a"), + () => api.agent("fail"), + () => api.agent("b"), + ]); + assert.deepEqual(results, ["ok:a", null, "ok:b"]); + assert.equal(api.getCallCount(), 3); + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// pipeline — no barrier across items (item B can finish stage2 before A stage1 ends) +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-pipe-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "pipe", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const events: string[] = []; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "pipe", description: "d" }, + args: undefined, + concurrency: 4, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async () => ({ finalResponse: "x" }), + }); + + const result = await api.pipeline( + ["slow", "fast"], + async (item: unknown) => { + events.push(`s1:${item}:start`); + await new Promise((r) => setTimeout(r, item === "slow" ? 40 : 5)); + events.push(`s1:${item}:end`); + return `${item}-1`; + }, + async (prev: unknown, item: unknown) => { + events.push(`s2:${item}:${prev}`); + return `${prev}-2`; + }, + ); + + assert.deepEqual(result, ["slow-1-2", "fast-1-2"]); + // fast finishes stage1 before slow does + const fastEnd = events.indexOf("s1:fast:end"); + const slowEnd = events.indexOf("s1:slow:end"); + assert.ok(fastEnd >= 0 && slowEnd >= 0 && fastEnd < slowEnd); + // fast may enter stage2 before slow finishes stage1 + const fastS2 = events.indexOf("s2:fast:fast-1"); + assert.ok(fastS2 >= 0 && fastS2 < slowEnd); + + // throw → null for that item + const withNull = await api.pipeline( + [1, 2], + async (n: unknown) => { + if (n === 2) throw new Error("nope"); + return n; + }, + ); + assert.deepEqual(withNull, [1, null]); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// phase ALS — concurrent chains keep separate phases for agent() +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-phase-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "phase", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const seen: Array<{ prompt: string; phase?: string }> = []; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "phase", description: "d" }, + args: undefined, + concurrency: 4, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async (input: WorkflowProviderRunInput) => { + seen.push({ prompt: input.prompt, phase: input.phase }); + await new Promise((r) => setTimeout(r, 15)); + return { finalResponse: "ok" }; + }, + }); + + await api.parallel([ + async () => { + api.phase("A"); + assert.equal(getCurrentWorkflowPhase(), "A"); + return api.agent("from-a"); + }, + async () => { + api.phase("B"); + assert.equal(getCurrentWorkflowPhase(), "B"); + return api.agent("from-b"); + }, + ]); + + const a = seen.find((s) => s.prompt === "from-a"); + const b = seen.find((s) => s.prompt === "from-b"); + assert.equal(a?.phase, "A"); + assert.equal(b?.phase, "B"); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// isolation: worktree uses createWorktree path as cwd +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-iso-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "iso", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const worktrees: string[] = []; + const createWorktree: CreateAgentWorktree = async ({ callIndex }) => { + const path = join(dir, `wt-${callIndex}`); + await mkdir(path, { recursive: true }); + worktrees.push(path); + return { + path, + finalize: async () => ({ dirty: false, removed: true }), + }; + }; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "iso", description: "d" }, + args: undefined, + concurrency: 2, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + createWorktree, + runProvider: async (input) => { + assert.equal(input.workspace, worktrees[0]); + return { finalResponse: "in-wt" }; + }, + }); + + const out = await api.agent("do", { isolation: "worktree", label: "iso" }); + assert.equal(out, "in-wt"); + const calls = store.listAgentCalls(run.id); + assert.equal(calls[0]?.isolation, "worktree"); + assert.equal(calls[0]?.worktreePath, worktrees[0]); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// provider resolve order + no writeMode +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-prov-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "prov", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const used: string[] = []; + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "prov", description: "d", defaultProvider: "claude" }, + args: undefined, + concurrency: 1, + signal: new AbortController().signal, + workspaceRoot: dir, + enabledProviders: ["codex", "claude"], + runProvider: async (input) => { + used.push(input.provider); + return { finalResponse: input.provider }; + }, + }); + assert.equal(await api.agent("x"), "claude"); + assert.equal(await api.agent("y", { provider: "codex" }), "codex"); + await assert.rejects( + async () => api.agent("z", { writeMode: "allowed" } as never), + /writeMode is not supported/, + ); + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// executeWorkflow end-to-end with sandbox + nest depth +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-exec-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "exec", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + + const childPath = join(dir, "child.js"); + await writeFile( + childPath, + ` +export const meta = { name: 'child', description: 'nested' } +return await agent('nested-prompt') +`, + ); + + const prompts: string[] = []; + const { result, callCount } = await executeWorkflow({ + source: ` +export const meta = { name: 'parent', description: 'p' } +const a = await agent('parent-prompt') +const nested = await workflow({ scriptPath: ${JSON.stringify(childPath)} }) +return { a, nested } +`, + runId: run.id, + journal: store, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async (input) => { + prompts.push(input.prompt); + return { finalResponse: `R:${input.prompt}` }; + }, + resolveNestedSource: async (ref) => { + if (typeof ref === "object" && ref.scriptPath) { + const { readFile } = await import("node:fs/promises"); + return readFile(ref.scriptPath, "utf8"); + } + throw new Error("unknown nest ref"); + }, + }); + + assert.deepEqual(result, { + a: "R:parent-prompt", + nested: "R:nested-prompt", + }); + assert.equal(callCount, 2); + assert.deepEqual(prompts, ["parent-prompt", "nested-prompt"]); + + // depth 2 must fail + await assert.rejects( + () => + executeWorkflow({ + source: ` +export const meta = { name: 'deep', description: 'd' } +return await workflow({ scriptPath: ${JSON.stringify(childPath)} }).then(async () => { + // child tries to nest again — child script: + return 1 +}) +`, + runId: run.id, + journal: store, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async () => ({ finalResponse: "x" }), + resolveNestedSource: async () => ` +export const meta = { name: 'mid', description: 'm' } +return await workflow({ scriptPath: 'x' }) +`, + }), + (error: unknown) => + error instanceof WorkflowEngineError && error.kind === "nest_depth", + ); + + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +// --------------------------------------------------------------------------- +// cancel via signal +// --------------------------------------------------------------------------- +{ + const dir = await mkdtemp(join(tmpdir(), "wf-cancel-")); + const store = new WorkflowStore(dir); + const run = store.createRun({ + name: "cancel", + source: "inline", + scriptPath: "inline", + scriptHash: "h", + workspaceRoot: dir, + }); + const ac = new AbortController(); + const api = createWorkflowApi({ + runId: run.id, + journal: store, + meta: { name: "cancel", description: "d" }, + args: undefined, + concurrency: 1, + signal: ac.signal, + workspaceRoot: dir, + enabledProviders: ["codex"], + runProvider: async () => { + ac.abort(); + return { finalResponse: "late" }; + }, + }); + // abort before agent + ac.abort(); + await assert.rejects(async () => api.agent("x"), WorkflowEngineError); + store.close(); + await rm(dir, { recursive: true, force: true }); +} + +void createStubBudget; +console.log("workflow-engine.test.ts: ok"); diff --git a/src/workflow-engine.ts b/src/workflow-engine.ts new file mode 100644 index 00000000..856dc292 --- /dev/null +++ b/src/workflow-engine.ts @@ -0,0 +1,206 @@ +import { availableParallelism } from "node:os"; +import { parseWorkflowScript, type ParsedWorkflowScript } from "./workflow-script.js"; +import { runWorkflowSandbox } from "./workflow-sandbox.js"; +import { + createWorkflowApi, + type CreateAgentWorktree, + type WorkflowApi, + type WorkflowJournal, + type WorkflowReplay, + type WorkflowRunProvider, + WorkflowEngineError, +} from "./workflow-api.js"; +import { + WORKFLOW_HOST_TIMEOUT_MS, + resolveWorkflowConcurrency, + type WorkflowMeta, + type WorkflowErrorKind, +} from "./workflow-types.js"; + +export interface ExecuteWorkflowOptions { + /** Pre-parsed script, or pass `source` instead. */ + parsed?: ParsedWorkflowScript; + source?: string; + filename?: string; + runId: string; + journal: WorkflowJournal & { + appendEvent(input: { + runId: string; + type: string; + phase?: string; + label?: string; + data?: unknown; + }): unknown; + }; + args?: unknown; + concurrency?: number; + signal?: AbortSignal; + workspaceRoot: string; + baseSha?: string; + enabledProviders: string[]; + runProvider: WorkflowRunProvider; + createWorktree?: CreateAgentWorktree; + replay?: WorkflowReplay; + resolveNestedSource?: (nameOrRef: string | { scriptPath: string }) => string | Promise; + nestDepth?: number; + timeoutMs?: number; + /** Optional hooks after API construction (tests). */ + onApi?: (api: WorkflowApi) => void; +} + +export interface ExecuteWorkflowResult { + result: unknown; + meta: WorkflowMeta; + callCount: number; +} + +/** + * Execute one workflow script body (top-level or nested). + * Does not create/claim/complete journal run rows — host/worker owns run lifecycle. + */ +export async function executeWorkflow( + options: ExecuteWorkflowOptions, +): Promise { + const parsed = + options.parsed ?? + parseWorkflowScript(options.source ?? "", { filename: options.filename }); + const nestDepth = options.nestDepth ?? 0; + const signal = options.signal ?? new AbortController().signal; + const concurrency = + options.concurrency ?? + resolveWorkflowConcurrency(parsed.meta.concurrency, availableParallelism()); + + const resolveNestedSource = options.resolveNestedSource; + + // Shared callIndex/semaphore for nested scripts via parent API path. + const api = createWorkflowApi({ + runId: options.runId, + journal: options.journal as WorkflowJournal, + meta: parsed.meta, + args: options.args, + concurrency, + signal, + workspaceRoot: options.workspaceRoot, + baseSha: options.baseSha, + enabledProviders: options.enabledProviders, + runProvider: options.runProvider, + createWorktree: options.createWorktree, + replay: options.replay, + nestDepth, + resolveNestedSource, + executeNested: resolveNestedSource + ? async (input) => + executeNestedOnApi({ + parentOptions: options, + parentApi: api, + source: input.source, + args: input.args, + nestDepth: input.nestDepth, + }) + : undefined, + }); + options.onApi?.(api); + + if (nestDepth === 0) { + options.journal.appendEvent({ + runId: options.runId, + type: "run_started", + data: { + name: parsed.meta.name, + scriptHash: parsed.scriptHash, + concurrency, + }, + }); + } + + try { + const result = await runWorkflowSandbox({ + parsed, + api, + timeoutMs: options.timeoutMs ?? WORKFLOW_HOST_TIMEOUT_MS, + }); + return { + result, + meta: parsed.meta, + callCount: api.getCallCount(), + }; + } catch (error) { + if (error instanceof WorkflowEngineError) { + throw error; + } + throw error; + } +} + +/** + * Nested script execution reusing parent's agent() call counter + semaphore + * by constructing a child API that shares internal state via re-entry. + * + * Implementation: run child sandbox with a new API that has nestDepth+1 but + * delegates agent/parallel/pipeline to the parent API (same callIndex). + */ +async function executeNestedOnApi(input: { + parentOptions: ExecuteWorkflowOptions; + parentApi: WorkflowApi; + source: string; + args: unknown; + nestDepth: number; +}): Promise { + if (input.nestDepth > WORKFLOW_MAX_NEST_DEPTH_LOCAL) { + throw new WorkflowEngineError( + "nest_depth", + `workflow() nesting limited to ${WORKFLOW_MAX_NEST_DEPTH_LOCAL} level`, + ); + } + const parsed = parseWorkflowScript(input.source, { + filename: "workflow:nested", + }); + + // Child surface: reuse parent agent/parallel/pipeline/phase/log/budget/workflow + // so callIndex + semaphore stay shared. Override args + meta for the child body. + const childApi: WorkflowApi = { + agent: input.parentApi.agent, + parallel: input.parentApi.parallel, + pipeline: input.parentApi.pipeline, + phase: input.parentApi.phase, + log: input.parentApi.log, + args: input.args, + budget: input.parentApi.budget, + // Child workflow() must see nestDepth via a wrapper that throws at depth>1. + workflow: async (...args: unknown[]) => { + throw new WorkflowEngineError( + "nest_depth", + `workflow() nesting limited to ${WORKFLOW_MAX_NEST_DEPTH_LOCAL} level`, + ); + }, + meta: parsed.meta, + getCallCount: () => input.parentApi.getCallCount(), + getNestDepth: () => input.nestDepth, + }; + + return runWorkflowSandbox({ + parsed, + api: childApi, + timeoutMs: input.parentOptions.timeoutMs ?? WORKFLOW_HOST_TIMEOUT_MS, + }); +} + +const WORKFLOW_MAX_NEST_DEPTH_LOCAL = 1; + +export function mapEngineErrorKind(error: unknown): WorkflowErrorKind { + if (error instanceof WorkflowEngineError) { + return error.kind; + } + if (error && typeof error === "object" && "name" in error) { + const name = String((error as { name: string }).name); + if (name === "WorkflowScriptError") { + const kind = (error as { kind?: string }).kind; + if (kind === "meta" || kind === "syntax" || kind === "script_too_large") { + return kind; + } + return "syntax"; + } + if (name === "WorkflowDeterminismError") return "determinism"; + } + return "internal"; +}