# pi-workflow

Declarative, routed **multi-agent workflows for [Pi](https://pi.dev)** — 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.

```bash
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):

```yaml
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:

```ts
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