diff --git a/README.md b/README.md index 3a62f49..3bf0092 100644 --- a/README.md +++ b/README.md @@ -116,5 +116,7 @@ cat program.ts | node skills/rig/rig.ts --typecheck ## Docs -See [skills/rig/SKILL.md](skills/rig/SKILL.md) for construction rules and -[skills/rig/references/runtime.md](skills/rig/references/runtime.md) for launcher and engine details. +See [skills/rig/SKILL.md](skills/rig/SKILL.md) for construction rules, +[skills/rig/references/runtime.md](skills/rig/references/runtime.md) for launcher and engine details, and +[skills/rig/references/claude-workflow-conversion.md](skills/rig/references/claude-workflow-conversion.md) +for porting Claude Code dynamic workflows to rig. diff --git a/skills/rig/SKILL.md b/skills/rig/SKILL.md index a8f09d0..bd2e51e 100644 --- a/skills/rig/SKILL.md +++ b/skills/rig/SKILL.md @@ -60,6 +60,9 @@ Defaults: `name: "agent"`, `model: "small"`, `maxTurns: 4`, string input/output, | Numeric schema choice | `s.int` for counts/line numbers; `s.number` for measurements and ratios | | Optional versus nullable | `s.optional(shape)` for omission; `s.nullable(shape)` for explicit `null` | | Deterministic TypeScript fan-out | `workflow({ meta, input?, body })` + `export default`; use `call`, `pipeline`, `parallel`, `until` inside `body` | +| One-off prompt inside a workflow | `call.text(prompt)` for a string, `call.json(prompt, schema)` for structured output | +| Reusable workflow step | Define an `agent({ input, output })` and `call(worker, input, { label, phase })` | +| Phase or log from an agent program | Import `phase` / `log` from `rig` and call them at top level; the launcher runs every program inside a workflow | | Custom model-callable operation | `defineTool(name, { description, parameters, handler })` | | Structured-output retries | `maxTurns` on the agent plus `addons: [repair()]` | | Retry with final-turn warning | `addons: [steering(), repair()]` in that order | @@ -103,6 +106,7 @@ 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. +- [Dynamic workflows](references/dynamic-workflows.md) — bounded fan-out, failure semantics, limits, budget, events, and convergence loops. +- [Claude workflow conversion](references/claude-workflow-conversion.md) — mapping Claude Code dynamic-workflow scripts onto rig primitives. - [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/claude-workflow-conversion.md b/skills/rig/references/claude-workflow-conversion.md new file mode 100644 index 0000000..7b0619e --- /dev/null +++ b/skills/rig/references/claude-workflow-conversion.md @@ -0,0 +1,155 @@ +# Converting Claude dynamic workflows to rig + +Read this reference when porting a Claude Code dynamic workflow script +(`.claude/workflows/*.workflow.js`) to a rig `workflow()` program, or when +comparing the two API surfaces. + +A dynamic workflow is a sandboxed JavaScript module that exports a literal +`meta`, reads `args`, and orchestrates subagents with injected globals +(`agent`, `parallel`, `pipeline`, `phase`, `log`, `budget`). rig covers the same +shape with a typed `workflow({ meta, input, body })` whose `body` receives those +primitives from its context instead of from globals. + +## Primitive mapping + +| Dynamic workflow | rig | Notes | +| --- | --- | --- | +| `export const meta = { name, description, phases, whenToUse }` | `workflow({ meta: { name, description, phases, whenToUse } })` | `phases` accepts `"Title"` or `{ title, detail }`; `meta` may reference variables | +| `args` (JSON string or object) | `input` schema + `context.input` | Parsed and validated by the launcher; no defensive `JSON.parse` | +| `await agent(prompt)` | `await call.text(prompt, options?)` | Returns `string \| null` | +| `await agent(prompt, { schema })` | `await call.json(prompt, schema, options?)` | `schema` is `s.object({ ... })`; result is typed and validated | +| Reused prompt + schema pair | `agent({ input, output, instructions })` then `call(worker, input, options?)` | Preferred for anything invoked more than once | +| `parallel(thunks)` | `parallel(thunks)` | Same barrier semantics; failures become `null` holes | +| `pipeline(items, ...stages)` | `pipeline(items, ...stages)` | Stages receive `(previous, item, index)`; the first stage's `previous` is the item | +| `phase(title)` | `phase(title)` | Same | +| `{ phase: "Verify" }` on a call | `{ phase: "Verify" }` in call options | Overrides the ambient phase for that call only | +| `{ label: "verify:x" }` | `{ label: "verify:x" }` | Appears in `agent_start`/`agent_done` events | +| `{ model: "sonnet" }` | `{ model: "sonnet" }` | rig passes the id straight to the engine | +| `{ effort: "high" }` | — | Not modeled; encode importance structurally or in the model id | +| `{ agentType: "Explore" }` | Prompt wording plus a narrow `tools` list | rig has no built-in read-only agent type | +| `{ timeoutMs }` / `{ retries }` | `{ timeout }` on the call; `maxTurns` + `repair()` on the agent | rig retries are turn-based, not process-based | +| `log(message)` | `log(message)` | Same | +| `budget.total / spent() / remaining()` | `budget.total / spent() / remaining()` | rig meters **agent calls** (`limits.maxAgents`), not tokens | +| `workflow(ref, args)` | `call.workflow(child, args, options?)` | Shares the limiter, budget, phase, and event stream | +| top-level `return value` | `return value` from `body` | Same | +| `Workflow({ scriptPath, args })` from a session | `cat args.json \| node skills/rig/rig.ts program.ts` | See [runtime](runtime.md) | + +Globals such as `phase` and `log` also exist as module-level imports from `rig`. +The launcher runs every program — including one whose root export is an `agent` — +inside a workflow run, so a partially ported script can call `phase()` and `log()` +at top level before the orchestration itself moves into `workflow({ body })`. + +## Schema conversion + +Dynamic workflows pass OpenAI-strict JSON Schema literals. rig schemas are +`s.*` values that compile to the same JSON Schema, so conversion is mechanical: + +| JSON Schema | rig | +| --- | --- | +| `{ type: "string" }` | `s.string` | +| `{ type: "string", description: "d" }` | `s.string("d")` | +| `{ type: "integer" }` | `s.int` | +| `{ type: "boolean" }` | `s.boolean` | +| `{ type: "string", enum: [...] }` | `s.enum("a", "b")` | +| `{ type: "array", items: X }` | `s.array(X)` | +| `{ type: "object", properties, required, additionalProperties: false }` | `s.object({ ... })` | +| `{ anyOf: [X, { type: "null" }] }` | `s.nullable(X)` | +| omitted from `required` | `s.optional(X)` | + +`additionalProperties: false` and a full `required` list are implicit in +`s.object`, so drop them. Use `s.path` for file paths and `s.url` for URIs. + +## Worked conversion + +Original dynamic workflow: + +```js +export const meta = { + name: 'audit', + description: 'Find and verify issues', + phases: [{ title: 'Find' }, { title: 'Verify' }], +} + +const FINDINGS = { + type: 'object', additionalProperties: false, + required: ['findings'], + properties: { + findings: { type: 'array', items: { + type: 'object', additionalProperties: false, + required: ['title', 'file'], + properties: { title: { type: 'string' }, file: { type: 'string' } }, + } }, + }, +} + +phase('Find') +const found = await parallel(args.areas.map((area) => () => + agent(`Audit ${area}. Report findings.`, { label: area, schema: FINDINGS }))) + +phase('Verify') +const verified = await pipeline( + found.filter(Boolean).flatMap((r) => r.findings), + (finding) => agent(`Verify: ${finding.title} in ${finding.file}.`, { + phase: 'Verify', schema: VERDICT, + }), +) +return verified.filter(Boolean).filter((v) => v.real) +``` + +Ported to rig: + +```ts +import { s, workflow } from "rig"; + +const finding = s.object({ title: s.string, file: s.path }); + +// Workflow role: audit areas in parallel, then verify each finding. +const audit = workflow({ + meta: { + name: "audit", + description: "Find and verify issues", + phases: [{ title: "Find" }, { title: "Verify", detail: "one verifier per finding" }], + }, + input: s.object({ areas: s.array(s.string) }), + body: async ({ call, input, parallel, phase, pipeline }) => { + phase("Find"); + const found = await parallel(input.areas.map((area) => () => + call.json(`Audit ${area}. Report findings.`, s.object({ findings: s.array(finding) }), { label: area }))); + + phase("Verify"); + const verified = await pipeline( + found.flatMap((result) => result?.findings ?? []), + (f: { title: string; file: string }) => + call.json(`Verify: ${f.title} in ${f.file}.`, s.object({ real: s.boolean }), { phase: "Verify" }), + ); + return verified.filter((v) => v?.real); + }, +}); + +export default audit; +``` + +## Behavior differences to keep in mind + +- **Failure holes.** `call` returns `null` when an agent fails, and `parallel` + turns rejected thunks into `null`. A rig `pipeline` stage that throws fails the + whole run instead of dropping that item to `null`, so programming bugs stay + visible; wrap a stage in `try`/`catch` when you want the Claude behavior. +- **Budget units.** rig counts agent calls, not tokens, so guard loops with + `budget.remaining() > n` where `n` is a call count. +- **Nesting depth.** `call.workflow` has no one-level restriction, but it shares + the parent's `maxAgents` and concurrency, so a nested run cannot escape the + parent's limits. +- **No sandbox restrictions.** rig programs are normal TypeScript modules: + `Date.now()`, imports, and Node built-ins are allowed, and the launcher owns + isolation instead of the runtime. +- **No resume journal, worktree isolation, or human checkpoints.** These are + runtime features of Claude Code, not API surface; a rig workflow that needs a + human decision should return a structured `needs_human` result and be re-run + with the answer in its input. + +## Related references + +- [Dynamic workflows](dynamic-workflows.md) — full rig workflow API. +- [Agent API and schemas](agent-api.md) — `s.*` helpers and call options. +- [Running and engines](runtime.md) — launching a workflow program. diff --git a/skills/rig/references/dynamic-workflows.md b/skills/rig/references/dynamic-workflows.md index facf30e..309649f 100644 --- a/skills/rig/references/dynamic-workflows.md +++ b/skills/rig/references/dynamic-workflows.md @@ -37,8 +37,10 @@ const results = await runWorkflow(audit, { }); ``` -`meta` describes the workflow for tools and progress displays. `input` preserves -Rig schema inference in both `body` and `runWorkflow({ args })`. +`meta` describes the workflow for tools and progress displays: `name`, +`description`, optional `phases` (each `"Title"` or `{ title, detail }`), and +optional `whenToUse`. `input` preserves Rig schema inference in both `body` and +`runWorkflow({ args })`. ## Workflow as default export @@ -75,6 +77,28 @@ export default linter; - Omit `input` for a no-input program (inline mode) or add `input: s.object({ ... })` for file-mode programs that read stdin JSON. - Use `agent()` with `agents:` when an LLM should improvise the coordination order. Use `workflow()` when TypeScript owns the orchestration (fan-out, branching, convergence). +## Top-level constructs in agent programs + +The launcher runs every program inside a workflow run, including programs whose +root export is an `agent`, a string, or a prompt builder. Module evaluation +happens inside that run, so `phase()` and `log()` imported from `rig` work at the +program's top level without declaring a `workflow()`: + +```ts +import { agent, log, phase } from "rig"; + +phase("Review"); +log("reviewing the staged diff"); + +export default agent({ instructions: "Review the staged diff." }); +``` + +`currentWorkflow()` returns the active run context (`call`, `budget`, `signal`, +`phase`, `log`) or `undefined` outside a run; `phase()` and `log()` are no-ops +outside a run. A `workflow()` default export is nested into the same run, so it +shares the launcher's limiter, budget, and event stream instead of starting a +second run. + ## Context | Member | Behavior | @@ -82,15 +106,33 @@ export default linter; | `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 | +| `call.json(prompt, schema, options?)` | Runs a one-off agent constrained to `schema`; returns typed output or `null` | +| `call.workflow(child, args?, options?)` | Runs another workflow inline on the same limiter, budget, and event stream | +| `pipeline(items, ...stages)` | Streams each item through every stage independently; 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 | +| `budget` | Agent-call meter: `total`, `spent()`, `remaining()` | | `signal` | Run cancellation signal for non-agent work | -Call options support `label` plus the normal per-call `model`, `timeout`, -`maxTurns`, and `signal` overrides. +Call options support `label` and `phase` (a per-call phase override) plus the +normal per-call `model`, `timeout`, `maxTurns`, and `signal` overrides. + +Each `pipeline` stage receives `(previous, item, index)`. The first stage's +`previous` is the item itself, so a single-stage pipeline is just +`pipeline(items, (item) => ...)`. Stages run per item with no barrier between +them, so one item can be in stage 3 while another is still in stage 1. + +`budget` is denominated in agent calls, not tokens: `budget.total` is the +effective `limits.maxAgents`, `spent()` counts started calls, and `remaining()` +is what is left before the run fails. Use it to scale depth: +`while (budget.remaining() > 10) { ... }`. + +`call.workflow` runs a child `workflow()` inline. It shares the parent's +concurrency limiter, agent budget, cancellation signal, and `onEvent` stream, and +brackets the child with `log` events. Restore a phase after the nested run if the +child called `phase()`. `parallel` turns rejected thunks into `null` holes. Agent failures passed through `pipeline` are already `null` because `call` handles them; other pipeline callback @@ -146,3 +188,10 @@ const final = await until( The loop stops when `done` is true, after `max` rounds, or after `noProgressRounds` consecutive equal defined progress keys. + +## Porting from Claude dynamic workflows + +The rig primitives mirror the Claude Code dynamic-workflow globals (`meta`, +`args`, `agent`, `parallel`, `pipeline`, `phase`, `log`, `budget`, nested +`workflow`). See [Converting Claude dynamic workflows to rig](claude-workflow-conversion.md) +for the full mapping, schema translation table, and behavior differences. diff --git a/skills/rig/references/runtime.md b/skills/rig/references/runtime.md index 732f5e2..edf3915 100644 --- a/skills/rig/references/runtime.md +++ b/skills/rig/references/runtime.md @@ -39,6 +39,12 @@ Stdin coercion follows the root schema: The launcher writes string results, or the string `text` field of an object result, directly to stdout. It JSON-serializes other results. +Both modes evaluate the program and run its root inside a workflow run, so +top-level `phase()` and `log()` work in any program and `currentWorkflow()` is +defined from module scope. A `workflow` default export nests into that run +instead of starting a second one. Run events are emitted under the +`workflow:event` debug category. + Add `--server` in either mode to start the Copilot server over stdio and force the Copilot engine. Without it, `copilotEngine()` connects over HTTP using `COPILOT_SDK_URI`, then `localhost:7777`. Use `--help`, `-h`, `help`, `/help`, or `/?` to print launcher usage. diff --git a/skills/rig/rig.ts b/skills/rig/rig.ts index 0199e65..58cca3a 100644 --- a/skills/rig/rig.ts +++ b/skills/rig/rig.ts @@ -98,6 +98,7 @@ import { writeSync } from "node:fs"; import { access, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { execFile } from "node:child_process"; import { promisify } from "node:util"; +import { AsyncLocalStorage } from "node:async_hooks"; import { CopilotClient, RuntimeConnection, approveAll } from "@github/copilot-sdk"; import type { CopilotClientOptions } from "@github/copilot-sdk"; @@ -1462,7 +1463,9 @@ export async function launchRigProgram(programPath: string, options: LaunchOptio debugLauncherProgram({ mode: "import", program: resolvedPath, cwd, server: options.startServer === true }); configureAgent(defaultAgentFactory({ cwd, ...(options.startServer ? { startServer: true } : {}) })); - await import(pathToFileURL(resolvedPath).href); + await runInRootWorkflow("launcher-program", async () => { + await import(pathToFileURL(resolvedPath).href); + }); } async function readStdin(stream: NodeJS.ReadableStream): Promise { @@ -1514,7 +1517,12 @@ function asRootProgram(value: unknown, name: string): AgentFn | undefined { const hasInput = "inputSchema" in w; const inputSchema: Schema = hasInput ? (w.inputSchema as Schema) : defaultStringSchema; const fn = Object.assign( - async (input: unknown) => runWorkflow(w, hasInput ? { args: input } : {}), + async (input: unknown) => { + const args = hasInput ? input : undefined; + const ambient = currentWorkflow(); + // Nest into the launcher's root run so limits, budget, and events are shared. + return ambient ? ambient.call.workflow(w, args) : runWorkflow(w, { args }); + }, { inputSchema, outputSchema: defaultStringSchema as Schema, @@ -1529,6 +1537,19 @@ function asRootProgram(value: unknown, name: string): AgentFn | undefined { return undefined; } +/** + * Runs a launcher program inside a workflow run so a program that exports an + * agent (or a plain prompt) can still use top-level workflow constructs such as + * `phase()` and `log()`. Module evaluation happens inside `body`, so top-level + * program statements observe the ambient run too. + */ +async function runInRootWorkflow(name: string, body: () => Promise): Promise { + return runWorkflow( + { meta: { name, description: "Rig program root" }, body }, + { onEvent: (event) => debugWorkflowEvent(event) }, + ); +} + function noInputInvocation(agentFn: AgentFn): unknown | undefined { const schema = agentFn.inputSchema; if ("type" in schema && schema.type === "string") { @@ -1731,16 +1752,19 @@ async function runRootAgentFromStdin( } configureAgent(defaultAgentFactory({ cwd, ...(options.startServer ? { startServer: true } : {}) })); - const mod = await import(pathToFileURL(resolvedPath).href); - const rootAgent = asRootProgram(mod.default, "launcher-root"); - if (!rootAgent) { - throw new Error("Expected program to export a root value (agent, workflow, string, or prompt builder) as default export."); - } - debugLauncherProgram({ mode: "file", program: resolvedPath, root: rootAgent.agentName, promptLength: prompt.length }); - - const result = await rootAgent(coerceStdinInput(rootAgent, prompt)); + let rootName = "launcher-root"; + const result = await runInRootWorkflow("launcher-root", async () => { + const mod = await import(pathToFileURL(resolvedPath).href); + const rootAgent = asRootProgram(mod.default, "launcher-root"); + if (!rootAgent) { + throw new Error("Expected program to export a root value (agent, workflow, string, or prompt builder) as default export."); + } + rootName = rootAgent.agentName; + debugLauncherProgram({ mode: "file", program: resolvedPath, root: rootAgent.agentName, promptLength: prompt.length }); + return rootAgent(coerceStdinInput(rootAgent, prompt)); + }); const rendered = renderStdout(result); - debugLauncherResult({ mode: "file", root: rootAgent.agentName, bytes: Buffer.byteLength(rendered) }); + debugLauncherResult({ mode: "file", root: rootName, bytes: Buffer.byteLength(rendered) }); io.stdout.write(rendered); } @@ -1768,19 +1792,23 @@ async function runProgramCodeFromStdin( return; } configureAgent(defaultAgentFactory({ cwd, ...(options.startServer ? { startServer: true } : {}) })); - const mod = await import(pathToFileURL(tempProgramPath).href); - const rootAgent = asRootProgram(mod.default, "launcher-inline-root"); - if (!rootAgent) { - throw new Error("Expected program to export a root value (agent, workflow, string, or prompt builder) as default export."); - } - debugLauncherProgram({ mode: "stdin", program: tempProgramPath, root: rootAgent.agentName, codeLength: programCode.length }); - const input = noInputInvocation(rootAgent); - if (input === undefined) { - throw new Error("Expected stdin program root agent to have no input (omit input or use input: s.object({}))."); - } - const result = await rootAgent(input); + let rootName = "launcher-inline-root"; + const result = await runInRootWorkflow("launcher-inline-root", async () => { + const mod = await import(pathToFileURL(tempProgramPath).href); + const rootAgent = asRootProgram(mod.default, "launcher-inline-root"); + if (!rootAgent) { + throw new Error("Expected program to export a root value (agent, workflow, string, or prompt builder) as default export."); + } + rootName = rootAgent.agentName; + debugLauncherProgram({ mode: "stdin", program: tempProgramPath, root: rootAgent.agentName, codeLength: programCode.length }); + const input = noInputInvocation(rootAgent); + if (input === undefined) { + throw new Error("Expected stdin program root agent to have no input (omit input or use input: s.object({}))."); + } + return rootAgent(input); + }); const rendered = renderStdout(result); - debugLauncherResult({ mode: "stdin", root: rootAgent.agentName, bytes: Buffer.byteLength(rendered) }); + debugLauncherResult({ mode: "stdin", root: rootName, bytes: Buffer.byteLength(rendered) }); io.stdout.write(rendered); } finally { await rm(tempDir, { recursive: true, force: true }); @@ -2011,14 +2039,36 @@ export function agent(spec: AgentSpec): AgentFn { export type AgentDefinitionFactory = typeof agent; +/** One declared phase of a workflow. A bare string is shorthand for `{ title }`. */ +export type WorkflowPhase = { + title: string; + detail?: string; +}; + export type WorkflowMeta = { name: string; description: string; - phases: readonly string[]; + /** Declared phases, in order. Each entry is a title or `{ title, detail }`. */ + phases?: readonly (string | WorkflowPhase)[]; + /** Guidance describing when this workflow should be selected. */ + whenToUse?: string; }; export type WorkflowCallOptions = CallOptions & { + /** Display label for this call in events and progress output. */ label?: string; + /** Phase recorded for this call, overriding the ambient `phase()`. */ + phase?: string; +}; + +/** Token-free budget meter, denominated in agent calls. */ +export type WorkflowBudget = { + /** Total agent calls allowed in this run (`limits.maxAgents`). */ + readonly total: number; + /** Agent calls started so far. */ + spent(): number; + /** Agent calls still available before the run fails. */ + remaining(): number; }; export type WorkflowEvent = @@ -2048,22 +2098,36 @@ export type WorkflowCall = { input: AgentInputValue, options?: WorkflowCallOptions, ): Promise; + /** Runs a one-off string-output agent built from `prompt`. */ text(prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise; + /** Runs a one-off agent built from `prompt` and constrained to `output`. */ + json( + prompt: string | PromptBuilder, + output: Output, + options?: WorkflowCallOptions, + ): Promise | null>; + /** Runs another workflow inline, sharing this run's limiter, budget, and events. */ + workflow( + child: Workflow, + args?: Input, + options?: WorkflowNestedOptions, + ): Promise; +}; + +export type WorkflowNestedOptions = { + /** Display label for the nested run in log events. */ + label?: string; }; 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)[]>; + pipeline: typeof pipeline; + parallel: typeof parallel; until(options: UntilOptions, step: UntilStep): Promise; phase(name: string): void; log(message: string): void; + budget: WorkflowBudget; signal: AbortSignal; }; @@ -2098,6 +2162,27 @@ export function workflow( }; } +const workflowStore = new AsyncLocalStorage>(); + +/** + * Returns the context of the innermost active workflow run, or `undefined` + * outside a run. Launcher programs always run inside a workflow, so a rig + * program can reach `call`, `budget`, and `signal` from module scope. + */ +export function currentWorkflow(): WorkflowContext | undefined { + return workflowStore.getStore(); +} + +/** Sets the ambient phase of the active workflow run. No-op outside a run. */ +export function phase(name: string): void { + currentWorkflow()?.phase(name); +} + +/** Emits a structured log event on the active workflow run. No-op outside a run. */ +export function log(message: string): void { + currentWorkflow()?.log(message); +} + export type WorkflowLimits = { concurrency?: number; maxAgents?: number; @@ -2176,11 +2261,52 @@ export async function parallel( })); } -export async function pipeline( +/** + * One pipeline stage. The first stage receives the item itself as `previous`, + * matching the `(previous, item, index)` shape of dynamic-workflow stages. + */ +export type PipelineStage = ( + previous: Previous, + item: Item, + index: number, +) => Promise | Next; + +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))); + stage1: PipelineStage, +): Promise<(R1 | null)[]>; +export async function pipeline( + items: readonly Item[], + stage1: PipelineStage, + stage2: PipelineStage, +): Promise<(R2 | null)[]>; +export async function pipeline( + items: readonly Item[], + stage1: PipelineStage, + stage2: PipelineStage, + stage3: PipelineStage, +): Promise<(R3 | null)[]>; +export async function pipeline( + items: readonly Item[], + stage1: PipelineStage, + stage2: PipelineStage, + stage3: PipelineStage, + stage4: PipelineStage, +): Promise<(R4 | null)[]>; +export async function pipeline( + items: readonly Item[], + ...stages: PipelineStage[] +): Promise<(unknown | null)[]> { + if (stages.length === 0) { + throw new Error("pipeline requires at least one stage."); + } + return Promise.all(items.map(async (item, index) => { + let value: unknown = item; + for (const stage of stages) { + value = await stage(value, item, index); + } + return value; + })); } export async function until(options: UntilOptions, step: UntilStep): Promise { @@ -2276,8 +2402,8 @@ export async function runWorkflow( const id = agentCount; const agentName = worker.agentName; - const phase = currentPhase; - const { label, signal: callSignal, ...agentOptions } = callOptions; + const { label, phase: phaseOverride, signal: callSignal, ...agentOptions } = callOptions; + const phase = phaseOverride ?? currentPhase; const fields = eventFields(phase, label); const callStarted = Date.now(); emit({ type: "agent_start", id, agent: agentName, ...fields, ts: callStarted }); @@ -2311,6 +2437,25 @@ export async function runWorkflow( return call(textAgent, "", callOptions); }; + call.json = (( + prompt: string | PromptBuilder, + output: Output, + callOptions: WorkflowCallOptions = {}, + ) => { + const jsonAgent = agent({ + name: callOptions.label ?? "workflow-json", + instructions: prompt, + output, + }); + return call(jsonAgent, "", callOptions); + }) as WorkflowCall["json"]; + + const budget: WorkflowBudget = { + total: maxAgents, + spent: () => agentCount, + remaining: () => Math.max(0, maxAgents - agentCount), + }; + const runUntil = (untilOptions: UntilOptions, step: UntilStep): Promise => until(untilOptions, async (state, round) => { signal.throwIfAborted(); @@ -2319,24 +2464,53 @@ export async function runWorkflow( return result; }); - const context: WorkflowContext = { - input: options.args as Input, + const setPhase = (name: string): void => { + currentPhase = name; + emit({ type: "phase_start", phase: name, ts: Date.now() }); + }; + + const writeLog = (message: string): void => { + emit({ type: "log", message, ...(currentPhase !== undefined && { phase: currentPhase }), ts: Date.now() }); + }; + + const makeContext = (input: ContextInput): WorkflowContext => ({ + 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() }); - }, + phase: setPhase, + log: writeLog, + budget, signal, - }; + }); + + call.workflow = (async ( + child: Workflow, + args?: ChildInput, + nestedOptions: WorkflowNestedOptions = {}, + ): Promise => { + const name = nestedOptions.label ?? child.meta.name; + const outerPhase = currentPhase; + const childContext = makeContext(args as ChildInput); + writeLog(`workflow ${name} started`); + try { + return await workflowStore.run( + childContext as WorkflowContext, + () => child.body(childContext), + ); + } finally { + currentPhase = outerPhase; + writeLog(`workflow ${name} finished`); + } + }) as WorkflowCall["workflow"]; + + const context = makeContext(options.args as Input); try { - const body = Promise.resolve(definition.body(context)); + const body = Promise.resolve( + workflowStore.run(context as WorkflowContext, () => definition.body(context)), + ); const output = await (wallLimit === undefined ? body : Promise.race([body, wallLimit])); emit({ type: "run_done", diff --git a/skills/rig/samples/310-workflow-audit-verify.md b/skills/rig/samples/310-workflow-audit-verify.md new file mode 100644 index 0000000..0f31530 --- /dev/null +++ b/skills/rig/samples/310-workflow-audit-verify.md @@ -0,0 +1,37 @@ +# 310 - Audit and Verify (Dynamic Workflow Port) + +Mirrors a Claude Code dynamic workflow: fan out finders, then stream each finding +through a verifier stage. See +[claude-workflow-conversion.md](../references/claude-workflow-conversion.md). + +```rig +import { s, workflow } from "rig"; + +const finding = s.object({ title: s.string, file: s.path }); + +// Workflow role: audit source areas in parallel, then verify each finding. +const audit = workflow({ + meta: { + name: "audit", + description: "Find and verify repository issues", + phases: [{ title: "Find" }, { title: "Verify", detail: "one verifier per finding" }], + whenToUse: "Auditing several areas that each need independent verification.", + }, + body: async ({ call, parallel, phase, pipeline }) => { + phase("Find"); + const areas = ["skills", "src", "scripts"]; + const found = await parallel(areas.map((area) => () => + call.json(`Audit ${area}/ for risky patterns.`, s.object({ findings: s.array(finding) }), { label: area }))); + + phase("Verify"); + const verdicts = await pipeline( + found.flatMap((result) => result?.findings ?? []), + (item: { title: string; file: string }) => + call.json(`Verify "${item.title}" in ${item.file}.`, s.object({ real: s.boolean }), { phase: "Verify" }), + ); + return verdicts.filter((verdict) => verdict?.real).length; + }, +}); + +export default audit; +``` diff --git a/src/launcher.stdin-ambient.fixture.ts b/src/launcher.stdin-ambient.fixture.ts new file mode 100644 index 0000000..5138b3e --- /dev/null +++ b/src/launcher.stdin-ambient.fixture.ts @@ -0,0 +1,12 @@ +import { agent, currentWorkflow, log, phase } from "rig"; + +phase("Prepare"); +log("program loaded"); + +(globalThis as { __launcherAmbientRun?: boolean }).__launcherAmbientRun = currentWorkflow() !== undefined; + +const root = agent({ + name: "launcher-stdin-ambient-root", +}); + +export default root; diff --git a/src/launcher.test.ts b/src/launcher.test.ts index 29e1116..d8f087f 100644 --- a/src/launcher.test.ts +++ b/src/launcher.test.ts @@ -114,6 +114,24 @@ it("supports stdin mode for string input/output root agents", async () => { expect(output.join("")).toBe("done"); }); +it("wraps an agent program in a workflow so top-level constructs work", async () => { + const fixturePath = resolve(dirname(fileURLToPath(import.meta.url)), "./launcher.stdin-ambient.fixture.ts"); + const stdin = Readable.from(["Review this patch"]); + const output: string[] = []; + const stdout = new Writable({ + write(chunk, _encoding, callback) { + output.push(chunk.toString()); + callback(); + }, + }); + + mocks.setSendAndWaitImpl(async () => JSON.stringify("done")); + await runLauncherCli([fixturePath], {}, { stdin, stdout }); + + expect((globalThis as { __launcherAmbientRun?: boolean }).__launcherAmbientRun).toBe(true); + expect(output.join("")).toBe("done"); +}); + it("supports stdin mode when root default export is a string", async () => { const fixturePath = resolve(dirname(fileURLToPath(import.meta.url)), "./launcher.stdin-default-string.fixture.ts"); const stdin = Readable.from(["Review this patch"]); diff --git a/src/workflow.test.ts b/src/workflow.test.ts index 1446744..eaa7524 100644 --- a/src/workflow.test.ts +++ b/src/workflow.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it, vi } from "vitest"; import type { AgentFn, CallOptions } from "rig"; import { + configureAgent, + currentWorkflow, + log, + phase, WorkflowLimitError, parallel, runWorkflow, @@ -161,3 +165,171 @@ describe("workflow primitives", () => { expect(stalled).toHaveBeenCalledTimes(2); }); }); + +describe("dynamic-workflow parity", () => { + it("threads pipeline stages per item and keeps the original item", async () => { + const seen: [number, number, number][] = []; + const definition = workflow({ + meta: { + name: "stages", + description: "multi-stage pipeline", + phases: [{ title: "Work", detail: "two stages" }], + whenToUse: "when each item needs several steps", + }, + body: ({ pipeline }) => pipeline( + [1, 2, 3], + (previous: number, item: number, index: number) => { + seen.push([previous, item, index]); + return previous * 10; + }, + (previous: number, item: number) => previous + item, + ), + }); + + await expect(runWorkflow(definition)).resolves.toEqual([11, 22, 33]); + expect(seen).toEqual([[1, 1, 0], [2, 2, 1], [3, 3, 2]]); + }); + + it("records a per-call phase override", async () => { + const worker = fakeAgent("echo", (value) => value); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "phased", description: "phase override" }, + body: async ({ call, phase }) => { + phase("Ambient"); + return call(worker, "ok", { phase: "Verify", label: "check" }); + }, + }); + + await runWorkflow(definition, { onEvent: (event) => events.push(event) }); + expect(events.find((event) => event.type === "agent_start")).toMatchObject({ + phase: "Verify", + label: "check", + }); + }); + + it("meters the agent budget", async () => { + const worker = fakeAgent("worker", (value) => value); + const definition = workflow({ + meta: { name: "budgeted", description: "budget meter" }, + body: async ({ budget, call }) => { + const before = budget.remaining(); + await call(worker, 1); + return { total: budget.total, before, spent: budget.spent(), after: budget.remaining() }; + }, + }); + + await expect(runWorkflow(definition, { limits: { maxAgents: 4 } })).resolves.toEqual({ + total: 4, + before: 4, + spent: 1, + after: 3, + }); + }); + + it("runs a nested workflow on the shared limiter, budget, and events", 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 child = workflow({ + meta: { name: "child", description: "child run" }, + body: ({ call, parallel: runParallel }) => + runParallel([() => call(worker, 1), () => call(worker, 2)]), + }); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "parent", description: "parent run" }, + body: async ({ budget, call }) => { + const nested = await call.workflow(child); + return { nested, spent: budget.spent() }; + }, + }); + + await expect(runWorkflow(definition, { + limits: { concurrency: 1 }, + onEvent: (event) => events.push(event), + })).resolves.toEqual({ nested: [1, 2], spent: 2 }); + expect(peak).toBe(1); + expect(events.filter((event) => event.type === "log").map((event) => event.message)).toEqual([ + "workflow child started", + "workflow child finished", + ]); + }); +}); + +describe("workflow one-off agents", () => { + it("runs call.text and call.json against the configured engine", async () => { + const prompts: string[] = []; + configureAgent(() => ({ + ask: async (prompt: string) => { + prompts.push(prompt); + return prompt.includes("count") ? '{"answer":"pong","count":1}' : '"pong"'; + }, + close: async () => {}, + })); + + const definition = workflow({ + meta: { name: "one-off", description: "text and json" }, + body: async ({ call }) => ({ + text: await call.text("Reply with one word: pong."), + json: await call.json( + "Reply with the answer and a count.", + s.object({ answer: s.string, count: s.int }), + { label: "answerer" }, + ), + }), + }); + + await expect(runWorkflow(definition)).resolves.toEqual({ + text: "pong", + json: { answer: "pong", count: 1 }, + }); + expect(prompts).toHaveLength(2); + }); +}); + +describe("ambient workflow context", () => { + it("routes top-level phase and log to the active run", async () => { + const worker = fakeAgent("worker", (value) => value); + const events: WorkflowEvent[] = []; + const definition = workflow({ + meta: { name: "ambient", description: "ambient helpers" }, + body: async ({ call }) => { + phase("Work"); + log("started"); + return call(worker, 1); + }, + }); + + await expect(runWorkflow(definition, { onEvent: (event) => events.push(event) })).resolves.toBe(1); + expect(events.find((event) => event.type === "phase_start")).toMatchObject({ phase: "Work" }); + expect(events.find((event) => event.type === "log")).toMatchObject({ message: "started", phase: "Work" }); + expect(events.find((event) => event.type === "agent_start")).toMatchObject({ phase: "Work" }); + }); + + it("exposes the run context through currentWorkflow and clears it afterwards", async () => { + const definition = workflow({ + meta: { name: "context", description: "context lookup" }, + body: async () => { + await Promise.resolve(); + return currentWorkflow()?.budget.total; + }, + }); + + await expect(runWorkflow(definition, { limits: { maxAgents: 7 } })).resolves.toBe(7); + expect(currentWorkflow()).toBeUndefined(); + }); + + it("ignores top-level phase and log outside a run", () => { + expect(() => { + phase("Work"); + log("no run"); + }).not.toThrow(); + }); +});