diff --git a/README.md b/README.md
index f444518..eb1bd92 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";
+
+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..589d2e6
--- /dev/null
+++ b/skills/rig/references/dynamic-workflows.md
@@ -0,0 +1,113 @@
+# Dynamic workflows
+
+Read this reference when deterministic TypeScript should coordinate multiple
+typed agents.
+
+## Define and run
+
+Import workflow APIs from `rig`:
+
+```ts
+import { agent, s } from "rig";
+import { runWorkflow, workflow } from "rig";
+
+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.
+
+`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
+
+```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 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.
+
+`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.
diff --git a/skills/rig/rig.ts b/skills/rig/rig.ts
index 033ac34..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,6 +1958,348 @@ export function agent(spec: AgentSpec<any, any>): AgentFn<any, any> {
 
 export type AgentDefinitionFactory = typeof agent;
 
+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<S> = (
+  state: S | undefined,
+  round: number,
+) => Promise<{ state: S; done?: boolean; progressKey?: string }> | { state: S; done?: boolean; progressKey?: string };
+
+export type WorkflowCall = {
+  <Input, Output>(
+    worker: AgentFn<Input, Output>,
+    input: AgentInputValue<Input>,
+    options?: WorkflowCallOptions,
+  ): Promise<Output | null>;
+  text(prompt: string | PromptBuilder, options?: WorkflowCallOptions): Promise<string | null>;
+};
+
+export type WorkflowContext<Input> = {
+  input: Input;
+  call: WorkflowCall;
+  pipeline<Item, Result>(
+    items: readonly Item[],
+    fn: (item: Item, index: number) => Promise<Result> | Result,
+  ): Promise<(Result | null)[]>;
+  parallel<Result>(
+    thunks: readonly (() => Promise<Result> | Result)[],
+  ): Promise<(Result | null)[]>;
+  until<S>(options: UntilOptions, step: UntilStep<S>): Promise<S>;
+  phase(name: string): void;
+  log(message: string): void;
+  signal: AbortSignal;
+};
+
+export type WorkflowSpec<Input extends Schema, Output> = {
+  meta: WorkflowMeta;
+  input: Input;
+  body(context: WorkflowContext<InferSchema<Input>>): Promise<Output> | Output;
+};
+
+export type Workflow<Input, Output> = {
+  readonly meta: WorkflowMeta;
+  readonly inputSchema?: Schema;
+  readonly body: (context: WorkflowContext<Input>) => Promise<Output> | Output;
+};
+
+type WorkflowWithoutInputSpec<Output> = {
+  meta: WorkflowMeta;
+  body(context: WorkflowContext<undefined>): Promise<Output> | Output;
+};
+
+export function workflow<const Input extends Schema, Output>(
+  spec: WorkflowSpec<Input, Output>,
+): Workflow<InferSchema<Input>, Output>;
+export function workflow<Output>(spec: WorkflowWithoutInputSpec<Output>): Workflow<undefined, Output>;
+export function workflow(
+  spec: WorkflowSpec<Schema, unknown> | WorkflowWithoutInputSpec<unknown>,
+): Workflow<unknown, unknown> {
+  return {
+    meta: spec.meta,
+    ...("input" in spec && { inputSchema: spec.input }),
+    body: spec.body as (context: WorkflowContext<unknown>) => unknown,
+  };
+}
+
+export type WorkflowLimits = {
+  concurrency?: number;
+  maxAgents?: number;
+  maxWallMs?: number;
+  warnAgents?: number;
+};
+
+export type RunWorkflowOptions<Input> = {
+  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<T>(fn: () => Promise<T>): Promise<T> {
+    if (this.#active >= this.#limit) {
+      await new Promise<void>((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<Result>(
+  thunks: readonly (() => Promise<Result> | 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<Item, Result>(
+  items: readonly Item[],
+  fn: (item: Item, index: number) => Promise<Result> | Result,
+): Promise<(Result | null)[]> {
+  return Promise.all(items.map((item, index) => fn(item, index)));
+}
+
+export async function until<S>(options: UntilOptions, step: UntilStep<S>): Promise<S> {
+  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<Input, Output>(
+  definition: Workflow<Input, Output>,
+  options: RunWorkflowOptions<Input> = {},
+): Promise<Output> {
+  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<never>((_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 <CallInput, CallOutput>(
+    worker: AgentFn<CallInput, CallOutput>,
+    input: AgentInputValue<CallInput>,
+    callOptions: WorkflowCallOptions = {},
+  ): Promise<CallOutput | null> => {
+    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 = <S>(untilOptions: UntilOptions, step: UntilStep<S>): Promise<S> =>
+    until(untilOptions, async (state, round) => {
+      signal.throwIfAborted();
+      const result = await step(state, round);
+      signal.throwIfAborted();
+      return result;
+    });
+
+  const context: WorkflowContext<Input> = {
+    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
new file mode 100644
index 0000000..1446744
--- /dev/null
+++ b/src/workflow.test.ts
@@ -0,0 +1,163 @@
+import { describe, expect, it, vi } from "vitest";
+import type { AgentFn, CallOptions } from "rig";
+import {
+  WorkflowLimitError,
+  parallel,
+  runWorkflow,
+  until,
+  workflow,
+  type WorkflowEvent,
+} from "rig";
+import { s } from "rig";
+
+function fakeAgent<Input, Output>(
+  name: string,
+  invoke: (input: Input, options?: CallOptions) => Promise<Output> | Output,
+): AgentFn<Input, Output> {
+  return Object.assign(invoke, { agentName: name }) as unknown as AgentFn<Input, Output>;
+}
+
+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<number, number>("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<number, number>("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<number, number>("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("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<string, string>("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<void, string>("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),
+    })).rejects.toThrow("Workflow exceeded maxWallMs (5).");
+    expect(events.at(-1)).toMatchObject({ type: "run_failed" });
+  });
+});
+
+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);
+  });
+});