From 9e15e47802fe7c25682b7db424a18ccdfdd1f1ad Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:45:52 +0000 Subject: [PATCH 1/5] Initial plan From b2183172a1470c258687a22961e906bdcff78200 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:52:22 +0000 Subject: [PATCH 2/5] feat: add bounded dynamic workflows Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com> --- package.json | 1 + skills/rig/rig.ts | 20 +++ src/workflow.test.ts | 152 ++++++++++++++++++++ src/workflow.ts | 332 +++++++++++++++++++++++++++++++++++++++++++ tsconfig.json | 1 + vitest.config.ts | 1 + 6 files changed, 507 insertions(+) create mode 100644 src/workflow.test.ts create mode 100644 src/workflow.ts diff --git a/package.json b/package.json index 8ad3af9..e8e3eb9 100644 --- a/package.json +++ b/package.json @@ -4,6 +4,7 @@ "type": "module", "exports": { ".": "./skills/rig/rig.ts", + "./workflow": "./src/workflow.ts", "./eslint": "./skills/rig/eslint/index.js", "./engines/anthropic": "./skills/rig/engines/anthropic.ts", "./engines/codex": "./skills/rig/engines/codex.ts", diff --git a/skills/rig/rig.ts b/skills/rig/rig.ts index 033ac34..c5d1a5d 100644 --- a/skills/rig/rig.ts +++ b/skills/rig/rig.ts @@ -1957,6 +1957,26 @@ export function agent(spec: AgentSpec): AgentFn { export type AgentDefinitionFactory = typeof agent; +export { + parallel, + pipeline, + runWorkflow, + until, + workflow, + WorkflowLimitError, + type RunWorkflowOptions, + type UntilOptions, + type UntilStep, + type Workflow, + type WorkflowCall, + type WorkflowCallOptions, + type WorkflowContext, + type WorkflowEvent, + type WorkflowLimits, + type WorkflowMeta, + type WorkflowSpec, +} from "../../src/workflow.ts"; + function validate(value: unknown, schema: Schema): ValidationResult { return validateSchema(value, schema, "$", false); } diff --git a/src/workflow.test.ts b/src/workflow.test.ts new file mode 100644 index 0000000..06f9663 --- /dev/null +++ b/src/workflow.test.ts @@ -0,0 +1,152 @@ +import { describe, expect, it, vi } from "vitest"; +import type { AgentFn, CallOptions } from "rig"; +import { + WorkflowLimitError, + parallel, + runWorkflow, + until, + workflow, + type WorkflowEvent, +} from "rig/workflow"; +import { s } from "rig"; + +function fakeAgent( + name: string, + invoke: (input: Input, options?: CallOptions) => Promise | Output, +): AgentFn { + return Object.assign(invoke, { agentName: name }) as unknown as AgentFn; +} + +describe("workflow", () => { + it("passes typed input through a workflow", async () => { + const definition = workflow({ + meta: { name: "typed", description: "typed input", phases: [] }, + input: s.object({ value: s.number }), + body: ({ input }) => input.value * 2, + }); + + await expect(runWorkflow(definition, { args: { value: 4 } })).resolves.toBe(8); + }); + + it("preserves pipeline order and returns null for failed agents", async () => { + const worker = fakeAgent("worker", async (value) => { + await new Promise((resolve) => setTimeout(resolve, 6 - value)); + if (value === 2) throw new Error("failed"); + return value * 10; + }); + const definition = workflow({ + meta: { name: "fan-out", description: "fan out", phases: ["Work"] }, + body: async ({ call, phase, pipeline }) => { + phase("Work"); + return pipeline([1, 2, 3], (value) => call(worker, value, { label: String(value) })); + }, + }); + + await expect(runWorkflow(definition)).resolves.toEqual([10, null, 30]); + }); + + it("shares one concurrency limit across nested fan-out", async () => { + let active = 0; + let peak = 0; + const worker = fakeAgent("worker", async (value) => { + active += 1; + peak = Math.max(peak, active); + await new Promise((resolve) => setTimeout(resolve, 5)); + active -= 1; + return value; + }); + const definition = workflow({ + meta: { name: "nested", description: "nested fan out", phases: [] }, + body: ({ call, pipeline }) => pipeline([1, 2], (outer) => + pipeline([1, 2, 3], (inner) => call(worker, outer * inner))), + }); + + await runWorkflow(definition, { limits: { concurrency: 2 } }); + expect(peak).toBe(2); + }); + + it("fails the run when the total agent cap is exceeded", async () => { + const worker = fakeAgent("worker", (value) => value); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "capped", description: "capped", phases: [] }, + body: ({ call, pipeline }) => pipeline([1, 2, 3], (value) => call(worker, value)), + }); + + await expect(runWorkflow(definition, { + limits: { maxAgents: 2 }, + onEvent: (event) => events.push(event), + })).rejects.toBeInstanceOf(WorkflowLimitError); + expect(events.at(-1)?.type).toBe("run_failed"); + }); + + it("emits phase, log, agent, and completion events", async () => { + const worker = fakeAgent("echo", (value) => value); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "events", description: "events", phases: ["Run"] }, + body: async ({ call, log, phase }) => { + phase("Run"); + log("starting"); + return call(worker, "ok", { label: "echo call" }); + }, + }); + + await expect(runWorkflow(definition, { + onEvent: (event) => events.push(event), + })).resolves.toBe("ok"); + expect(events.map((event) => event.type)).toEqual([ + "run_start", + "phase_start", + "log", + "agent_start", + "agent_done", + "run_done", + ]); + expect(events[3]).toMatchObject({ phase: "Run", label: "echo call" }); + }); + + it("aborts in-flight calls at the wall-clock limit", async () => { + const worker = fakeAgent("slow", (_input, options) => + new Promise((_resolve, reject) => { + options?.signal?.addEventListener("abort", () => reject(options.signal?.reason), { once: true }); + })); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "wall", description: "wall limit", phases: [] }, + body: ({ call }) => call(worker, undefined), + }); + + await expect(runWorkflow(definition, { + limits: { maxWallMs: 5 }, + onEvent: (event) => events.push(event), + })).resolves.toBeNull(); + expect(events.at(-1)).toMatchObject({ type: "run_done", status: "aborted" }); + }); +}); + +describe("workflow primitives", () => { + it("parallel converts thrown tasks to ordered null holes", async () => { + await expect(parallel([ + async () => 1, + async () => { throw new Error("no"); }, + async () => 3, + ])).resolves.toEqual([1, null, 3]); + }); + + it("until stops on completion or repeated progress keys", async () => { + const complete = vi.fn(async (state: number | undefined) => ({ + state: (state ?? 0) + 1, + done: state === 1, + })); + await expect(until({ max: 5 }, complete)).resolves.toBe(2); + expect(complete).toHaveBeenCalledTimes(2); + + const stalled = vi.fn(async (state: number | undefined) => ({ + state: (state ?? 0) + 1, + progressKey: "unchanged", + })); + await expect(until({ max: 10, noProgressRounds: 2 }, stalled)).resolves.toBe(2); + expect(stalled).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/workflow.ts b/src/workflow.ts new file mode 100644 index 0000000..e33e75d --- /dev/null +++ b/src/workflow.ts @@ -0,0 +1,332 @@ +import { availableParallelism } from "node:os"; +import { + agent, + type AgentFn, + type AgentInputValue, + type CallOptions, + type InferSchema, + type PromptBuilder, + type Schema, +} from "../skills/rig/rig.ts"; + +export type WorkflowMeta = { + name: string; + description: string; + phases: readonly string[]; +}; + +export type WorkflowCallOptions = CallOptions & { + label?: string; +}; + +export type WorkflowEvent = + | { type: "run_start"; meta: WorkflowMeta; ts: number } + | { type: "phase_start"; phase: string; ts: number } + | { type: "agent_start"; id: number; agent: string; phase?: string; label?: string; ts: number } + | { type: "agent_done"; id: number; agent: string; phase?: string; label?: string; ms: number; ts: number } + | { type: "agent_failed"; id: number; agent: string; phase?: string; label?: string; error: string; ms: number; ts: number } + | { type: "log"; message: string; phase?: string; ts: number } + | { type: "warning"; message: string; ts: number } + | { type: "run_done"; status: "completed" | "aborted"; ms: number; ts: number } + | { type: "run_failed"; error: string; ms: number; ts: number }; + +export type UntilOptions = { + max: number; + noProgressRounds?: number; +}; + +export type UntilStep = ( + state: S | undefined, + round: number, +) => Promise<{ state: S; done?: boolean; progressKey?: string }> | { state: S; done?: boolean; progressKey?: string }; + +export type WorkflowCall = { + ( + worker: AgentFn, + input: AgentInputValue, + options?: WorkflowCallOptions, + ): Promise; + text(prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise; +}; + +export type WorkflowContext = { + input: Input; + call: WorkflowCall; + pipeline( + items: readonly Item[], + fn: (item: Item, index: number) => Promise | Result, + ): Promise<(Result | null)[]>; + parallel( + thunks: readonly (() => Promise | Result)[], + ): Promise<(Result | null)[]>; + until(options: UntilOptions, step: UntilStep): Promise; + phase(name: string): void; + log(message: string): void; + signal: AbortSignal; +}; + +export type WorkflowSpec = { + meta: WorkflowMeta; + input: Input; + body(context: WorkflowContext>): Promise | Output; +}; + +export type Workflow = { + readonly meta: WorkflowMeta; + readonly inputSchema?: Schema; + readonly body: (context: WorkflowContext) => Promise | Output; +}; + +type WorkflowWithoutInputSpec = { + meta: WorkflowMeta; + body(context: WorkflowContext): Promise | Output; +}; + +export function workflow( + spec: WorkflowSpec, +): Workflow, Output>; +export function workflow(spec: WorkflowWithoutInputSpec): Workflow; +export function workflow( + spec: WorkflowSpec | WorkflowWithoutInputSpec, +): Workflow { + return { + meta: spec.meta, + ...("input" in spec && { inputSchema: spec.input }), + body: spec.body as (context: WorkflowContext) => unknown, + }; +} + +export type WorkflowLimits = { + concurrency?: number; + maxAgents?: number; + maxWallMs?: number; + warnAgents?: number; +}; + +export type RunWorkflowOptions = { + args?: AgentInputValue; + limits?: WorkflowLimits; + onEvent?: (event: WorkflowEvent) => void; + signal?: AbortSignal; +}; + +export class WorkflowLimitError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkflowLimitError"; + } +} + +class Limiter { + readonly #limit: number; + #active = 0; + readonly #waiting: (() => void)[] = []; + + constructor(limit: number) { + this.#limit = limit; + } + + async run(fn: () => Promise): Promise { + if (this.#active >= this.#limit) { + await new Promise((resolve) => this.#waiting.push(resolve)); + } + this.#active += 1; + try { + return await fn(); + } finally { + this.#active -= 1; + this.#waiting.shift()?.(); + } + } +} + +function positiveInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value < 1) { + throw new RangeError(`${name} must be a positive integer.`); + } + return value; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function eventFields(phase: string | undefined, label: string | undefined): { + phase?: string; + label?: string; +} { + return { + ...(phase !== undefined && { phase }), + ...(label !== undefined && { label }), + }; +} + +export async function parallel( + thunks: readonly (() => Promise | Result)[], +): Promise<(Result | null)[]> { + return Promise.all(thunks.map(async (thunk) => { + try { + return await thunk(); + } catch (error) { + if (error instanceof WorkflowLimitError) throw error; + return null; + } + })); +} + +export async function pipeline( + items: readonly Item[], + fn: (item: Item, index: number) => Promise | Result, +): Promise<(Result | null)[]> { + return parallel(items.map((item, index) => () => fn(item, index))); +} + +export async function until(options: UntilOptions, step: UntilStep): Promise { + const max = positiveInteger(options.max, "until max"); + const noProgressRounds = options.noProgressRounds === undefined + ? undefined + : positiveInteger(options.noProgressRounds, "until noProgressRounds"); + let state: S | undefined; + let previousKey: string | undefined; + let sameKeyRounds = 0; + + for (let round = 0; round < max; round += 1) { + const result = await step(state, round); + state = result.state; + if (result.done) return state; + + if (noProgressRounds !== undefined && result.progressKey !== undefined) { + sameKeyRounds = result.progressKey === previousKey ? sameKeyRounds + 1 : 1; + previousKey = result.progressKey; + if (sameKeyRounds >= noProgressRounds) return state; + } + } + + if (state === undefined) { + throw new Error("until did not run."); + } + return state; +} + +function combinedSignal(signal: AbortSignal, callSignal: AbortSignal | undefined): AbortSignal { + return callSignal === undefined ? signal : AbortSignal.any([signal, callSignal]); +} + +export async function runWorkflow( + definition: Workflow, + options: RunWorkflowOptions = {}, +): Promise { + const concurrency = positiveInteger( + options.limits?.concurrency ?? Math.min(16, Math.max(2, availableParallelism())), + "workflow concurrency", + ); + const maxAgents = positiveInteger(options.limits?.maxAgents ?? 1000, "workflow maxAgents"); + const warnAgents = positiveInteger(options.limits?.warnAgents ?? 25, "workflow warnAgents"); + const maxWallMs = options.limits?.maxWallMs; + if (maxWallMs !== undefined) positiveInteger(maxWallMs, "workflow maxWallMs"); + + const started = Date.now(); + const controller = new AbortController(); + const signal = options.signal === undefined + ? controller.signal + : AbortSignal.any([controller.signal, options.signal]); + const timer = maxWallMs === undefined + ? undefined + : setTimeout(() => controller.abort(new Error(`Workflow exceeded maxWallMs (${maxWallMs}).`)), maxWallMs); + timer?.unref(); + + const emit = (event: WorkflowEvent): void => { + try { + options.onEvent?.(event); + } catch { + // Observers must not affect workflow execution. + } + }; + emit({ type: "run_start", meta: definition.meta, ts: started }); + + const limiter = new Limiter(concurrency); + let currentPhase: string | undefined; + let agentCount = 0; + + const call = (async ( + worker: AgentFn, + input: AgentInputValue, + callOptions: WorkflowCallOptions = {}, + ): Promise => { + agentCount += 1; + if (agentCount > maxAgents) { + throw new WorkflowLimitError(`Workflow exceeded maxAgents (${maxAgents}).`); + } + if (agentCount === warnAgents + 1) { + emit({ type: "warning", message: `Workflow scheduled more than ${warnAgents} agents.`, ts: Date.now() }); + } + + const id = agentCount; + const agentName = worker.agentName; + const phase = currentPhase; + const { label, signal: callSignal, ...agentOptions } = callOptions; + const fields = eventFields(phase, label); + const callStarted = Date.now(); + emit({ type: "agent_start", id, agent: agentName, ...fields, ts: callStarted }); + + try { + const output = await limiter.run(() => worker(input, { + ...agentOptions, + signal: combinedSignal(signal, callSignal), + })); + emit({ type: "agent_done", id, agent: agentName, ...fields, ms: Date.now() - callStarted, ts: Date.now() }); + return output; + } catch (error) { + emit({ + type: "agent_failed", + id, + agent: agentName, + ...fields, + error: errorMessage(error), + ms: Date.now() - callStarted, + ts: Date.now(), + }); + return null; + } + }) as WorkflowCall; + + call.text = (prompt, callOptions = {}) => { + const textAgent = agent({ + name: callOptions.label ?? "workflow-text", + instructions: prompt, + }); + return call(textAgent, "", callOptions); + }; + + const context: WorkflowContext = { + input: options.args as Input, + call, + pipeline, + parallel, + until, + phase(name) { + currentPhase = name; + emit({ type: "phase_start", phase: name, ts: Date.now() }); + }, + log(message) { + emit({ type: "log", message, ...(currentPhase !== undefined && { phase: currentPhase }), ts: Date.now() }); + }, + signal, + }; + + try { + const output = await definition.body(context); + emit({ + type: "run_done", + status: signal.aborted ? "aborted" : "completed", + ms: Date.now() - started, + ts: Date.now(), + }); + return output; + } catch (error) { + emit({ type: "run_failed", error: errorMessage(error), ms: Date.now() - started, ts: Date.now() }); + throw error; + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} diff --git a/tsconfig.json b/tsconfig.json index 97ec57c..771e184 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -20,6 +20,7 @@ "noEmit": true, "paths": { "rig": ["./skills/rig/rig.ts"], + "rig/workflow": ["./src/workflow.ts"], "rig/engines/anthropic": ["./skills/rig/engines/anthropic.ts"], "rig/engines/codex": ["./skills/rig/engines/codex.ts"], "rig/engines/gemini": ["./skills/rig/engines/gemini.ts"], diff --git a/vitest.config.ts b/vitest.config.ts index 7d8533c..e24d254 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -5,6 +5,7 @@ export default defineConfig({ resolve: { alias: [ { find: /^rig$/, replacement: resolve(__dirname, "skills/rig/rig.ts") }, + { find: /^rig\/workflow$/, replacement: resolve(__dirname, "src/workflow.ts") }, { find: /^rig\/engines\/anthropic$/, replacement: resolve(__dirname, "skills/rig/engines/anthropic.ts") }, { find: /^rig\/engines\/codex$/, replacement: resolve(__dirname, "skills/rig/engines/codex.ts") }, { find: /^rig\/engines\/gemini$/, replacement: resolve(__dirname, "skills/rig/engines/gemini.ts") }, From bd9dad356d34f43071b374a74e3aba88c1e494f2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:55:39 +0000 Subject: [PATCH 3/5] docs: describe dynamic workflow orchestration Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com> --- README.md | 39 ++++++++ skills/rig/SKILL.md | 1 + skills/rig/references/dynamic-workflows.md | 111 +++++++++++++++++++++ 3 files changed, 151 insertions(+) create mode 100644 skills/rig/references/dynamic-workflows.md diff --git a/README.md b/README.md index f444518..0f394e0 100644 --- a/README.md +++ b/README.md @@ -211,6 +211,45 @@ Use these samples to quickly gauge how well `rig` supports increasingly agentic Per call, you can override `model`, `timeout`, `maxTurns`, and `signal`. +## Dynamic workflows + +Use `workflow()` when ordinary TypeScript should own multi-agent fan-out, branching, +and convergence instead of asking one model to improvise the orchestration: + +```ts +import { agent, s } from "rig"; +import { runWorkflow, workflow } from "rig/workflow"; + +const inspect = agent({ + input: s.string, + output: s.object({ finding: s.string }), + instructions: "Inspect the requested file.", +}); + +const audit = workflow({ + meta: { name: "audit", description: "Inspect files concurrently", phases: ["Audit"] }, + input: s.object({ files: s.array(s.string) }), + body: async ({ input, call, phase, pipeline }) => { + phase("Audit"); + return pipeline(input.files, (file) => call(inspect, file, { label: file })); + }, +}); + +const findings = await runWorkflow(audit, { + args: { files: ["src/a.ts", "src/b.ts"] }, + limits: { concurrency: 8, maxAgents: 100 }, +}); +``` + +`call` returns `null` when an agent fails. `pipeline` and `parallel` preserve input +order and use the run's shared agent concurrency limit. `until` provides a bounded +loop with optional repeated-progress detection. `phase`, `log`, and `onEvent` +expose structured progress without coupling workflows to a UI. Runs default to at +most 1,000 agent calls and warn after 25. + +See [Dynamic workflows](skills/rig/references/dynamic-workflows.md) for limits, +events, failure behavior, and the complete API. + ## Debug logging Set `RIG_DEBUG` to emit lazy JSONL diagnostics to stderr. A category enables itself and its children, `*` enables everything, and a `-` prefix excludes a category: diff --git a/skills/rig/SKILL.md b/skills/rig/SKILL.md index 7019843..915f206 100644 --- a/skills/rig/SKILL.md +++ b/skills/rig/SKILL.md @@ -93,5 +93,6 @@ Read only when the task needs the listed detail: - [Agent API and schemas](references/agent-api.md) — spec fields, schema overloads, tools, and invocation options. - [Prompt intents](references/prompt-intents.md) — complete helper semantics, dynamic inputs, writes, and failure behavior. - [Composition and addons](references/composition.md) — delegation patterns, dynamic sets, repair, steering, and addon lifecycle. +- [Dynamic workflows](references/dynamic-workflows.md) — bounded fan-out, failure semantics, limits, events, and convergence loops. - [Running and engines](references/runtime.md) — markdown/file launch modes, typechecking, Agentic Workflows, and SDK adapters. - [Linting](references/linting.md) — linter usage, autofixes, rules, and rule development. diff --git a/skills/rig/references/dynamic-workflows.md b/skills/rig/references/dynamic-workflows.md new file mode 100644 index 0000000..051cfda --- /dev/null +++ b/skills/rig/references/dynamic-workflows.md @@ -0,0 +1,111 @@ +# Dynamic workflows + +Read this reference when deterministic TypeScript should coordinate multiple +typed agents. + +## Define and run + +Import workflow APIs from `rig/workflow` (they are also exported from `rig`): + +```ts +import { agent, s } from "rig"; +import { runWorkflow, workflow } from "rig/workflow"; + +const review = agent({ + input: s.object({ file: s.string }), + output: s.object({ findings: s.array(s.string) }), + instructions: "Review the requested file.", +}); + +const audit = workflow({ + meta: { + name: "audit", + description: "Review files in parallel", + phases: ["Review"], + }, + input: s.object({ files: s.array(s.string) }), + body: async ({ input, call, phase, pipeline }) => { + phase("Review"); + return pipeline(input.files, (file) => + call(review, { file }, { label: file })); + }, +}); + +const results = await runWorkflow(audit, { + args: { files: ["src/a.ts", "src/b.ts"] }, + limits: { concurrency: 8 }, +}); +``` + +`meta` describes the workflow for tools and progress displays. `input` preserves +Rig schema inference in both `body` and `runWorkflow({ args })`. + +## Context + +| Member | Behavior | +| --- | --- | +| `input` | Typed workflow arguments | +| `call(worker, input, options?)` | Runs a typed agent; returns its output or `null` on agent failure | +| `call.text(prompt, options?)` | Runs a one-off string-output agent | +| `pipeline(items, fn)` | Starts every item immediately; agent calls flow through the shared limiter | +| `parallel(thunks)` | Runs all thunks as a barrier and preserves their order | +| `until(options, step)` | Runs a bounded convergence loop | +| `phase(name)` | Sets the phase attached to subsequent events | +| `log(message)` | Emits a structured log event | +| `signal` | Run cancellation signal for non-agent work | + +Call options support `label` plus the normal per-call `model`, `timeout`, +`maxTurns`, and `signal` overrides. + +`pipeline` and `parallel` turn rejected callbacks into `null` holes. +`WorkflowLimitError` is the exception: exceeding `maxAgents` fails the whole run +so runaway scheduling cannot be hidden as an ordinary worker failure. Exceptions +thrown elsewhere in `body` also fail the run. + +## Limits + +```ts +await runWorkflow(job, { + args, + limits: { + concurrency: 8, + maxAgents: 500, + maxWallMs: 45 * 60_000, + warnAgents: 25, + }, + signal, + onEvent: (event) => process.stderr.write(`${JSON.stringify(event)}\n`), +}); +``` + +- `concurrency` defaults to the machine parallelism clamped between 2 and 16. +- `maxAgents` defaults to 1,000. +- `warnAgents` defaults to 25 and emits one advisory warning when exceeded. +- `maxWallMs` is optional. When reached, it aborts in-flight calls. +- The limiter is shared by all nested primitives and is acquired only by agent + calls, so nested pipelines do not multiply concurrency or deadlock. + +`onEvent` receives `run_start`, `phase_start`, `agent_start`, `agent_done`, +`agent_failed`, `log`, `warning`, `run_done`, and `run_failed`. Event observer +errors never affect the run. + +## Convergence + +Use `until` instead of an open-ended loop: + +```ts +const final = await until( + { max: 8, noProgressRounds: 2 }, + async (previous, round) => { + const state = await checkAndFix(previous, round); + return { + state, + done: state.passed, + progressKey: state.failureSummary, + }; + }, +); +``` + +The loop stops when `done` is true, after `max` rounds, or after +`noProgressRounds` consecutive equal defined progress keys. From 3effd4196ce7d5de8662ad95905928700c5cf732 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 28 Jul 2026 12:59:49 +0000 Subject: [PATCH 4/5] fix: enforce workflow failure boundaries Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com> --- skills/rig/references/dynamic-workflows.md | 12 +++++---- src/workflow.test.ts | 15 +++++++++-- src/workflow.ts | 30 ++++++++++++++++++---- 3 files changed, 45 insertions(+), 12 deletions(-) diff --git a/skills/rig/references/dynamic-workflows.md b/skills/rig/references/dynamic-workflows.md index 051cfda..a5dc1c6 100644 --- a/skills/rig/references/dynamic-workflows.md +++ b/skills/rig/references/dynamic-workflows.md @@ -57,10 +57,12 @@ Rig schema inference in both `body` and `runWorkflow({ args })`. Call options support `label` plus the normal per-call `model`, `timeout`, `maxTurns`, and `signal` overrides. -`pipeline` and `parallel` turn rejected callbacks into `null` holes. -`WorkflowLimitError` is the exception: exceeding `maxAgents` fails the whole run -so runaway scheduling cannot be hidden as an ordinary worker failure. Exceptions -thrown elsewhere in `body` also fail the run. +`parallel` turns rejected thunks into `null` holes. Agent failures passed through +`pipeline` are already `null` because `call` handles them; other pipeline callback +errors fail the run rather than hiding programming bugs. `WorkflowLimitError` is +never converted to `null`: exceeding `maxAgents` fails the whole run so runaway +scheduling cannot be hidden as an ordinary worker failure. Exceptions thrown +elsewhere in `body` also fail the run. ## Limits @@ -81,7 +83,7 @@ await runWorkflow(job, { - `concurrency` defaults to the machine parallelism clamped between 2 and 16. - `maxAgents` defaults to 1,000. - `warnAgents` defaults to 25 and emits one advisory warning when exceeded. -- `maxWallMs` is optional. When reached, it aborts in-flight calls. +- `maxWallMs` is optional. When reached, it aborts in-flight calls and fails the run. - The limiter is shared by all nested primitives and is acquired only by agent calls, so nested pipelines do not multiply concurrency or deadlock. diff --git a/src/workflow.test.ts b/src/workflow.test.ts index 06f9663..46fcadf 100644 --- a/src/workflow.test.ts +++ b/src/workflow.test.ts @@ -80,6 +80,17 @@ describe("workflow", () => { expect(events.at(-1)?.type).toBe("run_failed"); }); + it("does not hide programming errors in pipeline callbacks", async () => { + const definition = workflow({ + meta: { name: "broken", description: "broken callback", phases: [] }, + body: ({ pipeline }) => pipeline([1], () => { + throw new TypeError("callback bug"); + }), + }); + + await expect(runWorkflow(definition)).rejects.toThrow("callback bug"); + }); + it("emits phase, log, agent, and completion events", async () => { const worker = fakeAgent("echo", (value) => value); const events: WorkflowEvent[] = []; @@ -120,8 +131,8 @@ describe("workflow", () => { await expect(runWorkflow(definition, { limits: { maxWallMs: 5 }, onEvent: (event) => events.push(event), - })).resolves.toBeNull(); - expect(events.at(-1)).toMatchObject({ type: "run_done", status: "aborted" }); + })).rejects.toThrow("Workflow exceeded maxWallMs (5)."); + expect(events.at(-1)).toMatchObject({ type: "run_failed" }); }); }); diff --git a/src/workflow.ts b/src/workflow.ts index e33e75d..e919add 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -104,7 +104,7 @@ export type WorkflowLimits = { }; export type RunWorkflowOptions = { - args?: AgentInputValue; + args?: Input; limits?: WorkflowLimits; onEvent?: (event: WorkflowEvent) => void; signal?: AbortSignal; @@ -178,7 +178,7 @@ export async function pipeline( items: readonly Item[], fn: (item: Item, index: number) => Promise | Result, ): Promise<(Result | null)[]> { - return parallel(items.map((item, index) => () => fn(item, index))); + return Promise.all(items.map((item, index) => fn(item, index))); } export async function until(options: UntilOptions, step: UntilStep): Promise { @@ -230,9 +230,19 @@ export async function runWorkflow( const signal = options.signal === undefined ? controller.signal : AbortSignal.any([controller.signal, options.signal]); + let failWallLimit: ((error: WorkflowLimitError) => void) | undefined; + const wallLimit = maxWallMs === undefined + ? undefined + : new Promise((_resolve, reject) => { + failWallLimit = reject; + }); const timer = maxWallMs === undefined ? undefined - : setTimeout(() => controller.abort(new Error(`Workflow exceeded maxWallMs (${maxWallMs}).`)), maxWallMs); + : setTimeout(() => { + const error = new WorkflowLimitError(`Workflow exceeded maxWallMs (${maxWallMs}).`); + controller.abort(error); + failWallLimit?.(error); + }, maxWallMs); timer?.unref(); const emit = (event: WorkflowEvent): void => { @@ -298,12 +308,20 @@ export async function runWorkflow( return call(textAgent, "", callOptions); }; + const runUntil = (untilOptions: UntilOptions, step: UntilStep): Promise => + until(untilOptions, async (state, round) => { + signal.throwIfAborted(); + const result = await step(state, round); + signal.throwIfAborted(); + return result; + }); + const context: WorkflowContext = { input: options.args as Input, call, pipeline, parallel, - until, + until: runUntil, phase(name) { currentPhase = name; emit({ type: "phase_start", phase: name, ts: Date.now() }); @@ -315,7 +333,8 @@ export async function runWorkflow( }; try { - const output = await definition.body(context); + const body = Promise.resolve(definition.body(context)); + const output = await (wallLimit === undefined ? body : Promise.race([body, wallLimit])); emit({ type: "run_done", status: signal.aborted ? "aborted" : "completed", @@ -324,6 +343,7 @@ export async function runWorkflow( }); return output; } catch (error) { + if (!controller.signal.aborted) controller.abort(error); emit({ type: "run_failed", error: errorMessage(error), ms: Date.now() - started, ts: Date.now() }); throw error; } finally { From cedfc6dbe914cf6fbe4db45f041c9a201a1ade80 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 28 Jul 2026 14:15:24 +0000 Subject: [PATCH 5/5] refactor: merge workflow into rig module Co-authored-by: pelikhan <4175913+pelikhan@users.noreply.github.com> --- README.md | 2 +- package.json | 1 - skills/rig/references/dynamic-workflows.md | 4 +- skills/rig/rig.ts | 361 +++++++++++++++++++-- src/workflow.test.ts | 2 +- src/workflow.ts | 352 -------------------- tsconfig.json | 1 - vitest.config.ts | 1 - 8 files changed, 346 insertions(+), 378 deletions(-) delete mode 100644 src/workflow.ts diff --git a/README.md b/README.md index 0f394e0..eb1bd92 100644 --- a/README.md +++ b/README.md @@ -218,7 +218,7 @@ and convergence instead of asking one model to improvise the orchestration: ```ts import { agent, s } from "rig"; -import { runWorkflow, workflow } from "rig/workflow"; +import { runWorkflow, workflow } from "rig"; const inspect = agent({ input: s.string, diff --git a/package.json b/package.json index e8e3eb9..8ad3af9 100644 --- a/package.json +++ b/package.json @@ -4,7 +4,6 @@ "type": "module", "exports": { ".": "./skills/rig/rig.ts", - "./workflow": "./src/workflow.ts", "./eslint": "./skills/rig/eslint/index.js", "./engines/anthropic": "./skills/rig/engines/anthropic.ts", "./engines/codex": "./skills/rig/engines/codex.ts", diff --git a/skills/rig/references/dynamic-workflows.md b/skills/rig/references/dynamic-workflows.md index a5dc1c6..589d2e6 100644 --- a/skills/rig/references/dynamic-workflows.md +++ b/skills/rig/references/dynamic-workflows.md @@ -5,11 +5,11 @@ typed agents. ## Define and run -Import workflow APIs from `rig/workflow` (they are also exported from `rig`): +Import workflow APIs from `rig`: ```ts import { agent, s } from "rig"; -import { runWorkflow, workflow } from "rig/workflow"; +import { runWorkflow, workflow } from "rig"; const review = agent({ input: s.object({ file: s.string }), diff --git a/skills/rig/rig.ts b/skills/rig/rig.ts index c5d1a5d..3d28b9c 100644 --- a/skills/rig/rig.ts +++ b/skills/rig/rig.ts @@ -87,6 +87,7 @@ * INV:p-write-no-path p.write() contributes a write instruction to the prompt; it does NOT return the path; hard-code path in agent output */ import { basename, dirname, isAbsolute, resolve } from "node:path"; +import { availableParallelism } from "node:os"; import { fileURLToPath, pathToFileURL } from "node:url"; import { writeSync } from "node:fs"; import { access, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; @@ -1957,25 +1958,347 @@ export function agent(spec: AgentSpec): AgentFn { export type AgentDefinitionFactory = typeof agent; -export { - parallel, - pipeline, - runWorkflow, - until, - workflow, - WorkflowLimitError, - type RunWorkflowOptions, - type UntilOptions, - type UntilStep, - type Workflow, - type WorkflowCall, - type WorkflowCallOptions, - type WorkflowContext, - type WorkflowEvent, - type WorkflowLimits, - type WorkflowMeta, - type WorkflowSpec, -} from "../../src/workflow.ts"; +export type WorkflowMeta = { + name: string; + description: string; + phases: readonly string[]; +}; + +export type WorkflowCallOptions = CallOptions & { + label?: string; +}; + +export type WorkflowEvent = + | { type: "run_start"; meta: WorkflowMeta; ts: number } + | { type: "phase_start"; phase: string; ts: number } + | { type: "agent_start"; id: number; agent: string; phase?: string; label?: string; ts: number } + | { type: "agent_done"; id: number; agent: string; phase?: string; label?: string; ms: number; ts: number } + | { type: "agent_failed"; id: number; agent: string; phase?: string; label?: string; error: string; ms: number; ts: number } + | { type: "log"; message: string; phase?: string; ts: number } + | { type: "warning"; message: string; ts: number } + | { type: "run_done"; status: "completed" | "aborted"; ms: number; ts: number } + | { type: "run_failed"; error: string; ms: number; ts: number }; + +export type UntilOptions = { + max: number; + noProgressRounds?: number; +}; + +export type UntilStep = ( + state: S | undefined, + round: number, +) => Promise<{ state: S; done?: boolean; progressKey?: string }> | { state: S; done?: boolean; progressKey?: string }; + +export type WorkflowCall = { + ( + worker: AgentFn, + input: AgentInputValue, + options?: WorkflowCallOptions, + ): Promise; + text(prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise; +}; + +export type WorkflowContext = { + input: Input; + call: WorkflowCall; + pipeline( + items: readonly Item[], + fn: (item: Item, index: number) => Promise | Result, + ): Promise<(Result | null)[]>; + parallel( + thunks: readonly (() => Promise | Result)[], + ): Promise<(Result | null)[]>; + until(options: UntilOptions, step: UntilStep): Promise; + phase(name: string): void; + log(message: string): void; + signal: AbortSignal; +}; + +export type WorkflowSpec = { + meta: WorkflowMeta; + input: Input; + body(context: WorkflowContext>): Promise | Output; +}; + +export type Workflow = { + readonly meta: WorkflowMeta; + readonly inputSchema?: Schema; + readonly body: (context: WorkflowContext) => Promise | Output; +}; + +type WorkflowWithoutInputSpec = { + meta: WorkflowMeta; + body(context: WorkflowContext): Promise | Output; +}; + +export function workflow( + spec: WorkflowSpec, +): Workflow, Output>; +export function workflow(spec: WorkflowWithoutInputSpec): Workflow; +export function workflow( + spec: WorkflowSpec | WorkflowWithoutInputSpec, +): Workflow { + return { + meta: spec.meta, + ...("input" in spec && { inputSchema: spec.input }), + body: spec.body as (context: WorkflowContext) => unknown, + }; +} + +export type WorkflowLimits = { + concurrency?: number; + maxAgents?: number; + maxWallMs?: number; + warnAgents?: number; +}; + +export type RunWorkflowOptions = { + args?: Input; + limits?: WorkflowLimits; + onEvent?: (event: WorkflowEvent) => void; + signal?: AbortSignal; +}; + +export class WorkflowLimitError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkflowLimitError"; + } +} + +class WorkflowLimiter { + readonly #limit: number; + #active = 0; + readonly #waiting: (() => void)[] = []; + + constructor(limit: number) { + this.#limit = limit; + } + + async run(fn: () => Promise): Promise { + if (this.#active >= this.#limit) { + await new Promise((resolve) => this.#waiting.push(resolve)); + } + this.#active += 1; + try { + return await fn(); + } finally { + this.#active -= 1; + this.#waiting.shift()?.(); + } + } +} + +function positiveInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value < 1) { + throw new RangeError(`${name} must be a positive integer.`); + } + return value; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function eventFields(phase: string | undefined, label: string | undefined): { + phase?: string; + label?: string; +} { + return { + ...(phase !== undefined && { phase }), + ...(label !== undefined && { label }), + }; +} + +export async function parallel( + thunks: readonly (() => Promise | Result)[], +): Promise<(Result | null)[]> { + return Promise.all(thunks.map(async (thunk) => { + try { + return await thunk(); + } catch (error) { + if (error instanceof WorkflowLimitError) throw error; + return null; + } + })); +} + +export async function pipeline( + items: readonly Item[], + fn: (item: Item, index: number) => Promise | Result, +): Promise<(Result | null)[]> { + return Promise.all(items.map((item, index) => fn(item, index))); +} + +export async function until(options: UntilOptions, step: UntilStep): Promise { + const max = positiveInteger(options.max, "until max"); + const noProgressRounds = options.noProgressRounds === undefined + ? undefined + : positiveInteger(options.noProgressRounds, "until noProgressRounds"); + let state: S | undefined; + let previousKey: string | undefined; + let sameKeyRounds = 0; + + for (let round = 0; round < max; round += 1) { + const result = await step(state, round); + state = result.state; + if (result.done) return state; + + if (noProgressRounds !== undefined && result.progressKey !== undefined) { + sameKeyRounds = result.progressKey === previousKey ? sameKeyRounds + 1 : 1; + previousKey = result.progressKey; + if (sameKeyRounds >= noProgressRounds) return state; + } + } + + if (state === undefined) { + throw new Error("until did not run."); + } + return state; +} + +function combinedSignal(signal: AbortSignal, callSignal: AbortSignal | undefined): AbortSignal { + return callSignal === undefined ? signal : AbortSignal.any([signal, callSignal]); +} + +export async function runWorkflow( + definition: Workflow, + options: RunWorkflowOptions = {}, +): Promise { + const concurrency = positiveInteger( + options.limits?.concurrency ?? Math.min(16, Math.max(2, availableParallelism())), + "workflow concurrency", + ); + const maxAgents = positiveInteger(options.limits?.maxAgents ?? 1000, "workflow maxAgents"); + const warnAgents = positiveInteger(options.limits?.warnAgents ?? 25, "workflow warnAgents"); + const maxWallMs = options.limits?.maxWallMs; + if (maxWallMs !== undefined) positiveInteger(maxWallMs, "workflow maxWallMs"); + + const started = Date.now(); + const controller = new AbortController(); + const signal = options.signal === undefined + ? controller.signal + : AbortSignal.any([controller.signal, options.signal]); + let failWallLimit: ((error: WorkflowLimitError) => void) | undefined; + const wallLimit = maxWallMs === undefined + ? undefined + : new Promise((_resolve, reject) => { + failWallLimit = reject; + }); + const timer = maxWallMs === undefined + ? undefined + : setTimeout(() => { + const error = new WorkflowLimitError(`Workflow exceeded maxWallMs (${maxWallMs}).`); + controller.abort(error); + failWallLimit?.(error); + }, maxWallMs); + timer?.unref(); + + const emit = (event: WorkflowEvent): void => { + try { + options.onEvent?.(event); + } catch { + // Observers must not affect workflow execution. + } + }; + emit({ type: "run_start", meta: definition.meta, ts: started }); + + const limiter = new WorkflowLimiter(concurrency); + let currentPhase: string | undefined; + let agentCount = 0; + + const call = (async ( + worker: AgentFn, + input: AgentInputValue, + callOptions: WorkflowCallOptions = {}, + ): Promise => { + agentCount += 1; + if (agentCount > maxAgents) { + throw new WorkflowLimitError(`Workflow exceeded maxAgents (${maxAgents}).`); + } + if (agentCount === warnAgents + 1) { + emit({ type: "warning", message: `Workflow scheduled more than ${warnAgents} agents.`, ts: Date.now() }); + } + + const id = agentCount; + const agentName = worker.agentName; + const phase = currentPhase; + const { label, signal: callSignal, ...agentOptions } = callOptions; + const fields = eventFields(phase, label); + const callStarted = Date.now(); + emit({ type: "agent_start", id, agent: agentName, ...fields, ts: callStarted }); + + try { + const output = await limiter.run(() => worker(input, { + ...agentOptions, + signal: combinedSignal(signal, callSignal), + })); + emit({ type: "agent_done", id, agent: agentName, ...fields, ms: Date.now() - callStarted, ts: Date.now() }); + return output; + } catch (error) { + emit({ + type: "agent_failed", + id, + agent: agentName, + ...fields, + error: errorMessage(error), + ms: Date.now() - callStarted, + ts: Date.now(), + }); + return null; + } + }) as WorkflowCall; + + call.text = (prompt, callOptions = {}) => { + const textAgent = agent({ + name: callOptions.label ?? "workflow-text", + instructions: prompt, + }); + return call(textAgent, "", callOptions); + }; + + const runUntil = (untilOptions: UntilOptions, step: UntilStep): Promise => + until(untilOptions, async (state, round) => { + signal.throwIfAborted(); + const result = await step(state, round); + signal.throwIfAborted(); + return result; + }); + + const context: WorkflowContext = { + input: options.args as Input, + call, + pipeline, + parallel, + until: runUntil, + phase(name) { + currentPhase = name; + emit({ type: "phase_start", phase: name, ts: Date.now() }); + }, + log(message) { + emit({ type: "log", message, ...(currentPhase !== undefined && { phase: currentPhase }), ts: Date.now() }); + }, + signal, + }; + + try { + const body = Promise.resolve(definition.body(context)); + const output = await (wallLimit === undefined ? body : Promise.race([body, wallLimit])); + emit({ + type: "run_done", + status: signal.aborted ? "aborted" : "completed", + ms: Date.now() - started, + ts: Date.now(), + }); + return output; + } catch (error) { + if (!controller.signal.aborted) controller.abort(error); + emit({ type: "run_failed", error: errorMessage(error), ms: Date.now() - started, ts: Date.now() }); + throw error; + } finally { + if (timer !== undefined) clearTimeout(timer); + } +} function validate(value: unknown, schema: Schema): ValidationResult { return validateSchema(value, schema, "$", false); diff --git a/src/workflow.test.ts b/src/workflow.test.ts index 46fcadf..1446744 100644 --- a/src/workflow.test.ts +++ b/src/workflow.test.ts @@ -7,7 +7,7 @@ import { until, workflow, type WorkflowEvent, -} from "rig/workflow"; +} from "rig"; import { s } from "rig"; function fakeAgent( diff --git a/src/workflow.ts b/src/workflow.ts deleted file mode 100644 index e919add..0000000 --- a/src/workflow.ts +++ /dev/null @@ -1,352 +0,0 @@ -import { availableParallelism } from "node:os"; -import { - agent, - type AgentFn, - type AgentInputValue, - type CallOptions, - type InferSchema, - type PromptBuilder, - type Schema, -} from "../skills/rig/rig.ts"; - -export type WorkflowMeta = { - name: string; - description: string; - phases: readonly string[]; -}; - -export type WorkflowCallOptions = CallOptions & { - label?: string; -}; - -export type WorkflowEvent = - | { type: "run_start"; meta: WorkflowMeta; ts: number } - | { type: "phase_start"; phase: string; ts: number } - | { type: "agent_start"; id: number; agent: string; phase?: string; label?: string; ts: number } - | { type: "agent_done"; id: number; agent: string; phase?: string; label?: string; ms: number; ts: number } - | { type: "agent_failed"; id: number; agent: string; phase?: string; label?: string; error: string; ms: number; ts: number } - | { type: "log"; message: string; phase?: string; ts: number } - | { type: "warning"; message: string; ts: number } - | { type: "run_done"; status: "completed" | "aborted"; ms: number; ts: number } - | { type: "run_failed"; error: string; ms: number; ts: number }; - -export type UntilOptions = { - max: number; - noProgressRounds?: number; -}; - -export type UntilStep = ( - state: S | undefined, - round: number, -) => Promise<{ state: S; done?: boolean; progressKey?: string }> | { state: S; done?: boolean; progressKey?: string }; - -export type WorkflowCall = { - ( - worker: AgentFn, - input: AgentInputValue, - options?: WorkflowCallOptions, - ): Promise; - text(prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise; -}; - -export type WorkflowContext = { - input: Input; - call: WorkflowCall; - pipeline( - items: readonly Item[], - fn: (item: Item, index: number) => Promise | Result, - ): Promise<(Result | null)[]>; - parallel( - thunks: readonly (() => Promise | Result)[], - ): Promise<(Result | null)[]>; - until(options: UntilOptions, step: UntilStep): Promise; - phase(name: string): void; - log(message: string): void; - signal: AbortSignal; -}; - -export type WorkflowSpec = { - meta: WorkflowMeta; - input: Input; - body(context: WorkflowContext>): Promise | Output; -}; - -export type Workflow = { - readonly meta: WorkflowMeta; - readonly inputSchema?: Schema; - readonly body: (context: WorkflowContext) => Promise | Output; -}; - -type WorkflowWithoutInputSpec = { - meta: WorkflowMeta; - body(context: WorkflowContext): Promise | Output; -}; - -export function workflow( - spec: WorkflowSpec, -): Workflow, Output>; -export function workflow(spec: WorkflowWithoutInputSpec): Workflow; -export function workflow( - spec: WorkflowSpec | WorkflowWithoutInputSpec, -): Workflow { - return { - meta: spec.meta, - ...("input" in spec && { inputSchema: spec.input }), - body: spec.body as (context: WorkflowContext) => unknown, - }; -} - -export type WorkflowLimits = { - concurrency?: number; - maxAgents?: number; - maxWallMs?: number; - warnAgents?: number; -}; - -export type RunWorkflowOptions = { - args?: Input; - limits?: WorkflowLimits; - onEvent?: (event: WorkflowEvent) => void; - signal?: AbortSignal; -}; - -export class WorkflowLimitError extends Error { - constructor(message: string) { - super(message); - this.name = "WorkflowLimitError"; - } -} - -class Limiter { - readonly #limit: number; - #active = 0; - readonly #waiting: (() => void)[] = []; - - constructor(limit: number) { - this.#limit = limit; - } - - async run(fn: () => Promise): Promise { - if (this.#active >= this.#limit) { - await new Promise((resolve) => this.#waiting.push(resolve)); - } - this.#active += 1; - try { - return await fn(); - } finally { - this.#active -= 1; - this.#waiting.shift()?.(); - } - } -} - -function positiveInteger(value: number, name: string): number { - if (!Number.isInteger(value) || value < 1) { - throw new RangeError(`${name} must be a positive integer.`); - } - return value; -} - -function errorMessage(error: unknown): string { - return error instanceof Error ? error.message : String(error); -} - -function eventFields(phase: string | undefined, label: string | undefined): { - phase?: string; - label?: string; -} { - return { - ...(phase !== undefined && { phase }), - ...(label !== undefined && { label }), - }; -} - -export async function parallel( - thunks: readonly (() => Promise | Result)[], -): Promise<(Result | null)[]> { - return Promise.all(thunks.map(async (thunk) => { - try { - return await thunk(); - } catch (error) { - if (error instanceof WorkflowLimitError) throw error; - return null; - } - })); -} - -export async function pipeline( - items: readonly Item[], - fn: (item: Item, index: number) => Promise | Result, -): Promise<(Result | null)[]> { - return Promise.all(items.map((item, index) => fn(item, index))); -} - -export async function until(options: UntilOptions, step: UntilStep): Promise { - const max = positiveInteger(options.max, "until max"); - const noProgressRounds = options.noProgressRounds === undefined - ? undefined - : positiveInteger(options.noProgressRounds, "until noProgressRounds"); - let state: S | undefined; - let previousKey: string | undefined; - let sameKeyRounds = 0; - - for (let round = 0; round < max; round += 1) { - const result = await step(state, round); - state = result.state; - if (result.done) return state; - - if (noProgressRounds !== undefined && result.progressKey !== undefined) { - sameKeyRounds = result.progressKey === previousKey ? sameKeyRounds + 1 : 1; - previousKey = result.progressKey; - if (sameKeyRounds >= noProgressRounds) return state; - } - } - - if (state === undefined) { - throw new Error("until did not run."); - } - return state; -} - -function combinedSignal(signal: AbortSignal, callSignal: AbortSignal | undefined): AbortSignal { - return callSignal === undefined ? signal : AbortSignal.any([signal, callSignal]); -} - -export async function runWorkflow( - definition: Workflow, - options: RunWorkflowOptions = {}, -): Promise { - const concurrency = positiveInteger( - options.limits?.concurrency ?? Math.min(16, Math.max(2, availableParallelism())), - "workflow concurrency", - ); - const maxAgents = positiveInteger(options.limits?.maxAgents ?? 1000, "workflow maxAgents"); - const warnAgents = positiveInteger(options.limits?.warnAgents ?? 25, "workflow warnAgents"); - const maxWallMs = options.limits?.maxWallMs; - if (maxWallMs !== undefined) positiveInteger(maxWallMs, "workflow maxWallMs"); - - const started = Date.now(); - const controller = new AbortController(); - const signal = options.signal === undefined - ? controller.signal - : AbortSignal.any([controller.signal, options.signal]); - let failWallLimit: ((error: WorkflowLimitError) => void) | undefined; - const wallLimit = maxWallMs === undefined - ? undefined - : new Promise((_resolve, reject) => { - failWallLimit = reject; - }); - const timer = maxWallMs === undefined - ? undefined - : setTimeout(() => { - const error = new WorkflowLimitError(`Workflow exceeded maxWallMs (${maxWallMs}).`); - controller.abort(error); - failWallLimit?.(error); - }, maxWallMs); - timer?.unref(); - - const emit = (event: WorkflowEvent): void => { - try { - options.onEvent?.(event); - } catch { - // Observers must not affect workflow execution. - } - }; - emit({ type: "run_start", meta: definition.meta, ts: started }); - - const limiter = new Limiter(concurrency); - let currentPhase: string | undefined; - let agentCount = 0; - - const call = (async ( - worker: AgentFn, - input: AgentInputValue, - callOptions: WorkflowCallOptions = {}, - ): Promise => { - agentCount += 1; - if (agentCount > maxAgents) { - throw new WorkflowLimitError(`Workflow exceeded maxAgents (${maxAgents}).`); - } - if (agentCount === warnAgents + 1) { - emit({ type: "warning", message: `Workflow scheduled more than ${warnAgents} agents.`, ts: Date.now() }); - } - - const id = agentCount; - const agentName = worker.agentName; - const phase = currentPhase; - const { label, signal: callSignal, ...agentOptions } = callOptions; - const fields = eventFields(phase, label); - const callStarted = Date.now(); - emit({ type: "agent_start", id, agent: agentName, ...fields, ts: callStarted }); - - try { - const output = await limiter.run(() => worker(input, { - ...agentOptions, - signal: combinedSignal(signal, callSignal), - })); - emit({ type: "agent_done", id, agent: agentName, ...fields, ms: Date.now() - callStarted, ts: Date.now() }); - return output; - } catch (error) { - emit({ - type: "agent_failed", - id, - agent: agentName, - ...fields, - error: errorMessage(error), - ms: Date.now() - callStarted, - ts: Date.now(), - }); - return null; - } - }) as WorkflowCall; - - call.text = (prompt, callOptions = {}) => { - const textAgent = agent({ - name: callOptions.label ?? "workflow-text", - instructions: prompt, - }); - return call(textAgent, "", callOptions); - }; - - const runUntil = (untilOptions: UntilOptions, step: UntilStep): Promise => - until(untilOptions, async (state, round) => { - signal.throwIfAborted(); - const result = await step(state, round); - signal.throwIfAborted(); - return result; - }); - - const context: WorkflowContext = { - input: options.args as Input, - call, - pipeline, - parallel, - until: runUntil, - phase(name) { - currentPhase = name; - emit({ type: "phase_start", phase: name, ts: Date.now() }); - }, - log(message) { - emit({ type: "log", message, ...(currentPhase !== undefined && { phase: currentPhase }), ts: Date.now() }); - }, - signal, - }; - - try { - const body = Promise.resolve(definition.body(context)); - const output = await (wallLimit === undefined ? body : Promise.race([body, wallLimit])); - emit({ - type: "run_done", - status: signal.aborted ? "aborted" : "completed", - ms: Date.now() - started, - ts: Date.now(), - }); - return output; - } catch (error) { - if (!controller.signal.aborted) controller.abort(error); - emit({ type: "run_failed", error: errorMessage(error), ms: Date.now() - started, ts: Date.now() }); - throw error; - } finally { - if (timer !== undefined) clearTimeout(timer); - } -} diff --git a/tsconfig.json b/tsconfig.json index 771e184..97ec57c 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -20,7 +20,6 @@ "noEmit": true, "paths": { "rig": ["./skills/rig/rig.ts"], - "rig/workflow": ["./src/workflow.ts"], "rig/engines/anthropic": ["./skills/rig/engines/anthropic.ts"], "rig/engines/codex": ["./skills/rig/engines/codex.ts"], "rig/engines/gemini": ["./skills/rig/engines/gemini.ts"], diff --git a/vitest.config.ts b/vitest.config.ts index e24d254..7d8533c 100644 --- a/vitest.config.ts +++ b/vitest.config.ts @@ -5,7 +5,6 @@ export default defineConfig({ resolve: { alias: [ { find: /^rig$/, replacement: resolve(__dirname, "skills/rig/rig.ts") }, - { find: /^rig\/workflow$/, replacement: resolve(__dirname, "src/workflow.ts") }, { find: /^rig\/engines\/anthropic$/, replacement: resolve(__dirname, "skills/rig/engines/anthropic.ts") }, { find: /^rig\/engines\/codex$/, replacement: resolve(__dirname, "skills/rig/engines/codex.ts") }, { find: /^rig\/engines\/gemini$/, replacement: resolve(__dirname, "skills/rig/engines/gemini.ts") },