pi-workflow

Declarative, routed multi-agent workflows for Pi — a Microsoft-agentic-workflow-style YAML grammar run by a decoupled, fail-closed state machine with a sandboxed template/expression engine.

Install it as a Pi extension and drive workflows with /workflow; or import the engine (schema + runner + expr + store) into any host and wire your own seams.

pi package add pi-workflow      # or: npm i pi-workflow

Feed it YAML

A workflow is a flat graph of typed steps joined by conditional routes. Author it as YAML (ergonomic) or JSON (what an editor writes):

workflow:
  name: triage-issue
  entry_point: classify
  default_model: openrouter/openai/gpt-4o-mini

steps:
  - name: classify
    type: agent
    prompt: |
      Classify this issue and return JSON {"severity": "...", "area": "..."}:
      {{ workflow.input.body }}
    output: { severity: { type: string }, area: { type: string } }
    routes:
      - to: escalate
        when: classify.output.severity == 'critical'
      - to: label            # unconditional fallthrough (must be last)

  - name: escalate
    type: human_gate
    prompt: "Critical issue in {{ classify.output.area }} — page on-call?"
    options:
      - { name: page, description: "Page on-call" }
      - { name: skip, description: "Just label it" }
    routes:
      - to: label

  - name: label
    type: agent
    prompt: "Suggest 3 GitHub labels for a {{ classify.output.area }} issue."
    routes:
      - to: $end

Then, inside Pi:

/workflow run ./triage-issue.yaml {"body": "app crashes on launch"}
/workflow list                 # saved workflows in ~/.pi/workflows
/workflow validate ./wf.yaml   # parse + graph-check without running
/workflow run triage-issue     # run a saved one by name

Step types

type	what it does
agent	one headless LLM turn under a safe-tools ceiling; best-effort JSON → output
set	pure context transform (value / values), no LLM
wait	sleep (30s / 5m / 1h)
human_gate	pause and ask the user to choose (surfaces in the Pi approval UI)
script	run a fixed command + args (⚠️ direct execution — gated, see below)
workflow	run another local workflow as a black-box sub-step
terminate	end the run success / failed

Fan-out groups run concurrently, bounded by a cap:

• parallel — a static set of already-defined agent steps.
• for_each — one agent per item of an array (source), with max_concurrent, failure_mode, and optional key_by.

Routes are evaluated top-to-bottom; the first when that is truthy wins, and a when-less route is the fallthrough. Targets are another step, $end (success), or $fail (failed).

Templates & conditions

{{ ... }} in prompts/values and when: conditions are evaluated by a confined, hand-written interpreter (expr.ts) — no eval, no Function, prototype-safe path lookups, DoS-bounded. An expression can only read the run context and apply a fixed set of Jinja-style | filters (default, length, upper, join, int, round, keys, …). It can never reach a function, a global, or the prototype chain — which is what makes running a workflow you imported from someone else safe.

The run context exposes workflow.input.* and, per executed step, <name>.output.* (plus <name>.choice for a gate).

Security posture (fail-closed by design)

• Strict schema. Every object is strictObject and steps are a discriminatedUnion — an unknown key or step type is a hard reject, never a permissive default.
• script steps bypass the agent gate, so the runner treats them as direct code execution: unattended, a script never runs — the run pauses and records a needs-approval notice (deferred). Attended, it runs only after an explicit approval.
• Sub-workflow refs are local-relative only. Registry / URL / GitHub / absolute / ~ / .. refs are rejected by the schema, and the loader re-confines the resolved path to the workflows dir — a remote ref would be an unsigned code-fetch (= RCE).
• Graph holes fail closed. A dangling route, an unresolvable entry_point, or a cyclic graph (past max_iterations) ends the run failed rather than guessing.

Use the engine directly

The runner imports no UI, no network, no Pi — every effect is injected, so you can host it anywhere:

import { runWorkflow } from "pi-workflow/runner";
import { parseWorkflowText } from "pi-workflow/yaml";

const { workflow } = parseWorkflowText(yamlText);
const result = await runWorkflow(workflow, { body: "…" }, {
  runAgent: async (spec) => ({ text: "…", output: {}, status: "ok" }),
  runScript: async (step, cwd) => ({ output: {}, status: "ok", exitCode: 0 }),
  askGate: async (step, prompt) => "page",
  attended: () => true,
  deferForApproval: async (reason) => {},
  sleep: (ms) => new Promise((r) => setTimeout(r, ms)),
  log: (m) => console.log(m),
  loadSubWorkflow: (ref) => null,   // optional
});

For a Pi host, makePiRunnerDeps(ctx) (from pi-workflow/pi-runner) wires all of the above to the extension command context automatically.

Configuration

makePiWorkflowExtension(opts) options: commandName (default workflow), workflowsDir (default ~/.pi/workflows, or $PI_WORKFLOWS_DIR), defaultModel, concurrency (default 4), quiet, onEvent.

License

MIT