# Workflow authoring reference This document is the authoring reference for pi-workflows definitions. It covers the file format, every node type, edge routing, the step contract the model sees, and how runs behave at runtime. For the on-disk run format, see [run-bundles.md](run-bundles.md). ## Workflow files A workflow is a TypeScript module whose default export is `defineWorkflow(...)`. Files are discovered by suffix (`.workflow.ts`, `.workflow.js`, `.workflow.mts`, `.workflow.mjs`) from these sources, in precedence order: 1. `.pi/workflows/` in the project (highest precedence on name collisions) 2. `~/.pi/agent/workflows/` globally 3. Workflows built into Pi Workflows Pi Workflows includes a built-in `monitor` workflow. A project or global file named `monitor.workflow.ts` replaces it. The workflow's command name is the file stem, so `.pi/workflows/triage.workflow.ts` runs as `/workflow triage`. A direct path also works: `/workflow ./somewhere/x.workflow.ts`. Files are loaded with [jiti](https://github.com/unjs/jiti), so plain TypeScript works without a build step, and `import ... from "@osolmaz/pi-workflows"` resolves to the engine that loaded the file. ```typescript import { agent, compute, defineWorkflow } from "@osolmaz/pi-workflows"; export default defineWorkflow({ name: "example", title: ({ input }) => `example: ${(input as { task?: string }).task}`, presentationPrompt: "Present the final answer clearly and concisely.", startAt: "ask", maxSteps: 50, nodes: { ask: agent({ prompt: ({ input }) => `Answer: ${(input as { task?: string }).task}`, expectedOutput: `{ "answer": "text" }`, }), finish: compute({ run: ({ outputs }) => outputs.ask }), }, edges: [{ from: "ask", to: "finish" }], }); ``` Top-level fields: | Field | Type | Notes | | -------------------- | ---------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `name` | `string` | Required. Used in run ids and the step contract. `answer`, `cancel`, `list`, `pause`, `resume`, and `status` are reserved for `/workflow` subcommands. | | `title` | `string` or function | Optional run title, resolved once at start from `{ input, workflowName }`. Async resolution is bounded (30s) and cancellable. | | `presentationPrompt` | `string` or function | Optional instructions for a normal assistant response after the run. A function receives `{ state, finalOutput, signal }` and may return a prompt or `undefined`. See [Result presentation](#result-presentation). | | `startAt` | `string` | Required. Id of the first node. | | `nodes` | `Record` | Required, non-empty. Node ids must match `[A-Za-z_][A-Za-z0-9_-]*`. | | `edges` | `WorkflowEdge[]` | Required. See routing below. | | `maxSteps` | `number` | Optional loop bound, default 100. The run fails when exceeded. | `defineWorkflow` validates the shape eagerly (node ids, edge shapes, function fields) and validates the graph (unknown targets, duplicate outgoing edges, unreachable nodes) when a run starts. ## Node context Every node callback receives the same context object: ```typescript type WorkflowNodeContext = { input: unknown; // the run input outputs: Record; // accepted output per finished node id results: Record; // full result records, including failures state: WorkflowRunState; // the live run state (read-only by convention) signal: AbortSignal; // aborted on node timeout or run cancellation }; ``` `outputs` only contains nodes that finished with outcome `ok`. When a node runs more than once (a loop), the latest result wins. A failed retry removes the node's earlier output from `outputs`. Long-running compute, action, and checkpoint callbacks should observe `context.signal` (pass it to `fetch`/`spawn`, or check `signal.aborted` between steps). When the node times out or the run is cancelled, the engine stops waiting immediately, but only cooperative callbacks stop doing work. ## Durable runs, parking, and resume Every interactive `/workflow` run is tracked in the project run queue (see [CONTROLLERS.md](CONTROLLERS.md) for the store). The session that starts a run claims it and owns it while it executes; every bundle write proves the claim first (write fencing). Closing the Pi session mid-run no longer cancels the run. The engine **parks**: it stops without a terminal event, releases the claim, and leaves the bundle resumable. When a runner is available again (a reopened Pi session or the standalone host), the run **resumes** at the node it stopped on. Completed nodes replay from the recorded state; only the interrupted node and everything downstream rerun. Resume repairs a torn trace tail, drops trace events the state projection never recorded, and refuses to continue if the workflow source changed since the run started (a forced resume records the mismatch). The standalone host runs without any Pi session: ```bash pi-workflows host --project /path/to/project ``` The host claims parked runs, resumes them, and reconciles durable controllers. Conversation nodes execute in headless `pi --mode rpc` children that load a small bridge extension; the model sees the same `workflow` tool contract as an in-session run. The host is a foreground process: start it in a terminal and stop it with Ctrl-C. A second host for the same project refuses to start, and a host that dies has its orphaned children reaped by the next one. While the host works, any open Pi session stays current: a per-session watermark over the shared run event feed produces catch-up summaries and quiet context updates. ## Node types ### agent Sends a prompt into the current pi conversation and waits for the model to submit output through the `workflow` tool. ```typescript agent({ prompt: ({ outputs }) => `Review this: ${JSON.stringify(outputs.implement)}`, expectedOutput: `{ "verdict": "clean" | "issues_found" }`, validate: (output) => output, // optional; throw to reject the submission timeoutMs: 30 * 60_000, // optional; default 15 minutes statusDetail: "reviewing", // optional; shown in widget and viewer }); ``` The engine appends a step contract to the prompt (see below). When the model calls the tool, the output passes through normalization (a JSON string is parsed tolerantly) and then `validate`. If `validate` throws, the tool call returns an error and the model can retry within the same step. If the agent ends its turn without submitting, the extension nudges it, twice by default, then fails the step. ### compute Runs a TypeScript function inline. Use it for pure data shaping. ```typescript compute({ run: ({ outputs }) => ({ merged: { ...outputs } }) }); ``` ### action Performs a side effect. Two forms exist. The function form runs arbitrary TypeScript: ```typescript action({ run: async ({ input }) => await deployPreview(input) }); ``` The shell form (`shell` is a synonym that requires `exec`) runs a command owned by the runtime, so the workflow author decides exactly what executes, with a timeout and captured output: ```typescript shell({ exec: ({ input }) => ({ command: "git", args: ["status", "--porcelain"], cwd: "/path/to/repo", timeoutMs: 10_000, allowNonZeroExit: false, }), parse: (result) => ({ dirty: result.stdout.trim().length > 0 }), }); ``` Without `parse`, the node output is the full `ShellActionResult` (`stdout`, `stderr`, `exitCode`, `signal`, `durationMs`). A non-zero exit fails the node unless `allowNonZeroExit` is set. Captured stdout and stderr are each capped (default 1,000,000 characters, configurable with `maxOutputChars`) so verbose commands cannot exhaust memory. Both action forms record a receipt (command, exit code, duration) in the step record for auditability, including when the command fails. ### checkpoint Ends the run in a `waiting` state for human review. The checkpoint bundle is terminal, so no process keeps running while the run waits. The human answers with `/workflow answer ` (or plain text), which starts a **continuation run**: a new run with its own bundle and trace, linked to the checkpointed run through `parentRunId`. The continuation receives the answer as its input, carries forward every output the parent produced (including the checkpoint's), and continues routing along the checkpoint's outgoing edge. Outgoing edges from checkpoint nodes are allowed exactly so continuations have somewhere to go; step accounting carries over, so `maxSteps` bounds the whole chain. ```typescript checkpoint({ summary: "human decides how to proceed", run: ({ outputs }) => outputs.reconcile, // optional; default output is { summary } }); ``` ### decision `decision` is sugar over `agent` for constrained choices. It builds the prompt suffix listing the choices, sets `expectedOutput`, and validates that the submitted object carries one of the allowed values in the decision field (default `route`). ```typescript const choices = ["y", "n"] as const; decision({ choices, question: ({ outputs }) => `Same as proposed? ${JSON.stringify(outputs.propose)}`, }); ``` Pair it with `decisionEdge`, which builds the matching `switch` edge and makes a missing case a compile-time error: ```typescript decisionEdge({ from: "compare", choices, cases: { y: "implement", n: "reconcile" } }); ``` ## Edges and routing Each node has at most one outgoing edge. A plain edge is unconditional: ```typescript { from: "a", to: "b" } ``` A `switch` edge routes on a JSON path evaluated against the node's result: ```typescript { from: "review", switch: { on: "$.route", cases: { clean: "done", issues_found: "fix" } } } ``` Path roots: - `$.field` and `$output.field` read from the node's accepted output. - `$result.field` reads from the result record. `$result.outcome` is the main use, with values `ok`, `failed`, `timed_out`, or `cancelled`, which lets a workflow route failures to a recovery node instead of failing the run. A missing case for the resolved value fails the run with a routing error. A node with no outgoing edge (or no matching failure route) ends the run: `completed` on success, `failed`/`timed_out`/`cancelled` otherwise. ## Model workflow control The model sees one `workflow` tool. Its `action` field supports: - `list` for discovered workflow names and sources. - `start` with a workflow name or path and structured input. - `status` for the active run or a supplied run ID. - `pause`, `resume`, and `cancel` for the active run. - `answer` with checkpoint input and an optional run ID. - `submit` for the current workflow step contract. A model-started run is queued until the model's current turn settles. The first workflow prompt then starts a new turn. This keeps the requesting turn outside the workflow's first attempt and prevents an early missing-submission reminder. The normal extension offers all actions. The headless RPC bridge offers only `submit`, so a workflow child cannot recursively control other runs. ### Built-in monitor The built-in `monitor` workflow turns a plain request for repeated checks into one looping workflow run. Its input is: ```json { "task": "Check pull request 123", "everyMinutes": 30, "reportWhen": "Checks fail or the state changes materially", "stopWhen": "The pull request is merged or closed", "maxChecks": 1000 } ``` The first check runs immediately. Each accepted check records a bounded current observation and chooses whether to continue, report, or stop. A report uses a separate agent node so its structured check result is validated before the user sees the message. The next check can read the previous accepted observation. Intervals must be whole minutes from 1 through 1,440. `maxChecks` defaults to 1,000 and cannot exceed 1,000. The workflow also has a finite step limit and bounded observation and report sizes. The interval uses the existing shell action to launch the current Node executable with a timer. This works on every platform supported by Pi. The node and command timeouts are higher than the maximum interval. Cancelling the workflow aborts the timer process immediately. If the owning Pi process or standalone host stops during the wait, normal parking rules abort the shell node and resume later by running that wait again from the beginning. A monitor uses the session's single active workflow slot. It does not provide cron syntax, calendar scheduling, OS notifications, or a background service. ## The step contract Every `agent` prompt ends with a step contract block naming the workflow, the step id, the attempt id, and the expected output shape: ``` --- Workflow step contract (workflow: autoimplement, step: review, attempt: 6f9d…) Complete this step by calling the `workflow` tool exactly once with: {"action": "submit", "step": "review", "attempt": "6f9d…", "output": } Expected output: { "route": "clean" | "issues_found", "reason": "short justification" } The step is complete only after the workflow tool accepts the output. If the tool reports a validation error, correct the output and call it again. ``` The `workflow` tool uses `{ action: "submit", step, attempt, output }` for step results. Submissions are rejected (with a reason the model sees) when no step is pending, the step id is wrong, the attempt id belongs to an earlier attempt of the same node (loops revisit node ids, so each attempt gets a fresh id), or `validate` throws. Acceptance resolves the step and the engine advances; the next agent prompt arrives as a new user message in the same conversation. ## Result presentation Workflow nodes produce structured JSON for routing and persistence. When a person should see a normal prose response after the run, add `presentationPrompt` at the top level: ```typescript export default defineWorkflow({ name: "report", presentationPrompt: ({ state, finalOutput }) => state.status === "waiting" ? `Explain this recommendation and ask the user to decide: ${JSON.stringify(finalOutput)}` : "Summarize the completed result and any remaining limitations.", // ...startAt, nodes, and edges }); ``` After the final run state has been persisted, the Pi extension sends the presentation instructions and bounded final result to the model as a hidden follow-up message. The next visible message is a normal assistant response. Returning `undefined`, returning an empty string, or omitting `presentationPrompt` produces no follow-up. Cancelled runs are never presented. Async prompt builders have 30 seconds to finish and receive an `AbortSignal` that fires on timeout, session shutdown, or when a new workflow or normal user turn starts; stale presentations are discarded. Once a presentation message has been queued, another workflow cannot start until that assistant response settles, so results cannot interleave. Presentation is outside the workflow graph: it cannot route to another node, change the run status, or alter the run bundle. If prompt generation or message delivery fails, the extension reports a warning and leaves the finished run unchanged. Opting in adds one hidden custom message and one assistant response to the normal Pi session; it adds no other persistent data and uses no Pi internals. ## Runtime behavior Runs execute one node at a time. Every transition is persisted to the run bundle before the engine moves on, which is what makes the live viewer possible. Defaults worth knowing: - Node timeout is 15 minutes unless the node sets `timeoutMs`. A timed-out node has outcome `timed_out` and can be routed with `$result.outcome`. - `maxSteps` (workflow-level, default 100) bounds loops built from cycles in the graph. - `/workflow pause` requests a pause: the current step finishes normally, then the run holds at the step boundary (`paused: true` in the run state, `run_paused` in the trace) until `/workflow resume` or `/workflow cancel`. Pausing never interrupts a node mid-flight. - Interrupting a turn (escape) auto-pauses the run: the pending agent step is held without nudges and the engine pauses at the next boundary. Node timeouts keep ticking while held, so a long-abandoned step still times out. `/workflow resume` re-delivers the pending step prompt. - `/workflow cancel` aborts the current node and marks the run `cancelled`. When no run is live but the widget still shows a parked or finished run, the same command clears the widget. - One workflow runs per session at a time. - Agent nudges: if the model ends its turn without submitting the pending step, it gets a reminder, twice by default, then the step fails. ## Workflows started by controllers A controller can start a workflow as a finite child job with `ctx.workflows.ensure()`. The request key is stable across reconciliation passes, and the input fingerprint prevents one key from being reused for different work. ```typescript const run = await ctx.workflows.ensure({ requestKey: `repair:${resource.metadata.generation}`, workflow: "repair-pull-request", input: { repository: resource.spec.repository, number: resource.spec.number }, }); if (run.state !== "succeeded") { return ctx.requeueAfter(5_000, { workflowRun: { requestId: run.requestId, ...(run.runId ? { runId: run.runId } : {}), state: run.state, attempt: run.attempt, }, }); } ``` Child workflow completion queues the parent resource again. A running child left by a stopped host is recorded as a failed run bundle with a `run_interrupted` event. The controller treats that child attempt as interrupted, and the next parent reconciliation starts another immutable attempt. Consequential external mutations should use the controller effect API so uncertain results are observed before retry. See [CONTROLLERS.md](CONTROLLERS.md) for controller definitions and the full recovery contract. ## Using the engine outside pi The engine is pi-agnostic. `WorkflowEngine` takes any `AgentStepExecutor`, so tests (and other hosts) can script agent steps: ```typescript import { WorkflowEngine, type AgentStepExecutor } from "@osolmaz/pi-workflows"; const executor: AgentStepExecutor = { async runAgentStep(request) { const accepted = await request.accept({ answer: "42" }); if (!accepted.ok) throw new Error(accepted.error); return { output: accepted.value }; }, }; const engine = new WorkflowEngine({ executor, outputRoot: "/tmp/runs" }); const { state } = await engine.run(workflow, { task: "..." }); ```