@mhingston5/agent-workflows

Define agent workflows as data, not code. Write a YAML DAG, validate it, and execute it against any runner — no harness lock-in.

Why

Hardcoded pipeline sequences break when you add branches, gates, or conditional skips. A declarative DAG lets you change pipeline topology by editing YAML, not refactoring dispatch code.

Install

npm install @mhingston5/agent-workflows

Requires a peer dependency on Zod v3 or v4.

30-second example

# workflow.yaml
name: build-and-deploy
on:
  invoke:
    args:
      - name: branch
        type: string
        required: true

nodes:
  - id: test
    deterministic: true
    exec:
      command: "npm test"
    max_retries: 1

  - id: build
    deterministic: true
    depends_on: [test]
    exec:
      command: "npm run build"

  - id: deploy
    deterministic: false
    depends_on: [build]
    when: $args.branch == 'main'
    exec:
      command: "npm run deploy"

import { WorkflowEngine, ShellRunner, parseYaml } from "@mhingston5/agent-workflows";
import { readFileSync } from "fs";

const yaml = readFileSync("workflow.yaml", "utf-8");
const { workflow } = parseYaml(yaml);
if (!workflow) process.exit(1);

const engine = new WorkflowEngine(new ShellRunner());
const result = await engine.execute(workflow, { branch: "main" });

console.log(result.success ? "Deployed" : `Failed: ${result.error}`);

Workflow YAML reference

Top-level

| Field | Type | Required | Description | |-------|------|----------|-------------| | name | string | yes | Workflow identifier | | require_deterministic | boolean | no | Skip non-deterministic nodes when true | | on.invoke.args | Arg[] | no | Input arguments for the workflow | | env.variables | EnvVar[] | no | Workflow-level environment variables | | env.secrets | string[] | no | Secret names available to all nodes |

Node types

Each node has exactly one of these types:

| Type | Field | Description | |------|-------|-------------| | Shell | exec.command | Runs a command via the runner | | Skill | invoke_skill.name | Dispatches to an agent skill via the runner | | Gate | gate.type: approval | Pauses for human approval | | Sub-workflow | uses.workflow | Loads and runs another workflow file |

Node fields

| Field | Type | Default | Description | |-------|------|---------|-------------| | id | string | required | Unique node identifier | | deterministic | boolean | false | Produces reproducible output | | depends_on | string[] | [] | Upstream node IDs | | trigger_rule | enum | all_succeeded | When this node can run | | when | string | — | Conditional expression; node is skipped if falsy | | on_failure | string | — | Node ID to route to on failure | | max_retries | integer | 0 | Retry count (0–3) | | timeout_seconds | integer | — | Per-node timeout | | output.bind | string | — | State key to bind node result to | | artifacts | Artifact[] | — | Named artifacts produced by this node | | env | EnvVar[] | — | Node-level environment variables | | secrets | string[] | — | Node-level secret names |

Trigger rules

• all_succeeded (default) — all dependencies must succeed
• any_completed — at least one dependency finished (succeeded, failed, or skipped)
• always — runs regardless of dependency outcomes

When conditions

Use when to conditionally skip nodes. Supports:

| Operator | Example | |----------|---------| | == | $args.track == 'express' | | != | $state.redirect != true | | > / >= / < / <= | $state.score >= 7 | | && | $a && $b | | \|\| | $a \|\| $b |

Variable references in conditions and invoke_skill.with:

| Syntax | Reads from | |--------|-----------| | $args.FIELD | Workflow input arguments | | $state.FIELD | Mutable workflow state | | $node_id.output | Output of an upstream node | | $node_id.artifact.NAME | Artifact from an upstream node |

Runner contract

Implement AgentWorkflowRunner to connect the engine to your platform:

import type { AgentWorkflowRunner, WorkflowNode, ExecutionContext, NodeResult } from "@mhingston5/agent-workflows";

class MyRunner implements AgentWorkflowRunner {
  async execute(node: WorkflowNode, ctx: ExecutionContext): Promise<NodeResult> {
    if (node.invoke_skill) {
      // dispatch to your skill system
      return { success: true, output: { ... } };
    }
    if (node.exec) {
      // run a shell command
      return { success: true, output: "done" };
    }
    if (node.gate) {
      // wait for human approval
      const approved = await askHuman(node.id);
      return { success: approved, output: approved ? "approved" : "rejected" };
    }
    return { success: false, error: `Unknown node type` };
  }
}

ExecutionContext

The runner receives a context with these methods:

| Method / field | Description | |----------------|-------------| | ctx.args | Workflow input arguments | | ctx.env | Merged environment variables | | ctx.secrets | Available secret names | | ctx.readState(key) | Read a value from workflow state | | ctx.writeState(key, value) | Write a value into workflow state | | ctx.log(event) | Emit a structured event |

writeState is useful when a runner needs to flatten complex outputs into state keys that downstream when conditions can reference.

NodeResult

interface NodeResult {
  success: boolean;
  output?: unknown;      // Node output (stored in state if output.bind is set)
  artifacts?: Artifact[]; // Named artifacts
  error?: string;       // Error message on failure
}

Engine options

The WorkflowEngine constructor accepts an optional third argument:

const engine = new WorkflowEngine(runner, subWorkflowLoader, {
  onEvent: (event) => {
    // Receive every engine event (node_start, node_end, node_skip, etc.)
    eventLog.append(event);
  },
});

Event types

| Type | When | |------|------| | node_start | Node begins execution (includes attempt number) | | node_end | Node completes successfully | | node_skip | Node is skipped (includes reason) | | node_retry | Node fails and will be retried | | node_failure | Node fails permanently | | workflow_complete | All nodes finished successfully | | workflow_failure | A node failed and the workflow stops |

Built-in runners

| Runner | Purpose | |--------|---------| | InMemoryRunner | Test runner with per-node handler registration and mock fallbacks | | ShellRunner | Runs exec.command via child_process.execSync |

Validation

CLI

npx workflow-validate workflow.yaml

Programmatic

import { parseYaml } from "@mhingston5/agent-workflows";

const result = parseYaml(yamlString);
if (!result.success) {
  for (const error of result.errors) {
    console.error(`${error.type}: ${error.message}`);
  }
}

Validation checks

• Structural (Zod): field types, required properties, enum values
• Semantic (DAG): cycle detection, unknown dependencies, unresolved output/artifact references, one node type per node

Common patterns

Conditional branch (investigation redirect)

nodes:
  - id: intake
    invoke_skill: { name: intake }
    output: { bind: intake_result }

  - id: investigate
    depends_on: [intake]
    when: $state.redirect_to_investigate == true
    invoke_skill: { name: investigate }

  - id: checkout
    depends_on: [intake]
    when: $state.redirect_to_investigate != true
    invoke_skill: { name: checkout }

The runner uses ctx.writeState("redirect_to_investigate", true) from the intake handler to set the flat state key the when conditions read.

Approval gate

nodes:
  - id: gate-review
    when: $state.requires_review == true
    gate: { type: approval, on_timeout: abort }

  - id: deploy
    depends_on: [build, gate-review]
    trigger_rule: any_completed
    exec: { command: "npm run deploy" }

The gate is skipped when requires_review is falsy. trigger_rule: any_completed lets deploy proceed regardless.

Express track (skip heavy nodes)

nodes:
  - id: design
    when: $args.pipeline_track != 'express'
    invoke_skill: { name: brainstorming }

  - id: implement
    depends_on: [design, checkout]
    trigger_rule: any_completed
    invoke_skill: { name: coordinator }

API

| Export | Description | |--------|-------------| | WorkflowEngine | DAG execution engine | | parseYaml(yaml) | Parse and validate a YAML workflow | | parseJson(json) | Parse and validate a JSON workflow | | parseWorkflow(obj) | Validate an already-parsed object | | InMemoryRunner | Test runner with mock fallbacks | | ShellRunner | Shell command runner |

License

MIT