Pipo — Specification

Pipo is a simple, agent-first pipeline manager for continuous data. You describe a pipeline in a .pipo file. Pipo keeps it running and turns every incoming event into a delivered, verified result.

• Status: draft v1
• Supersedes: docs/idea.txt

1. What Pipo is#

Pipo runs continuous pipelines. A pipeline has no "done" state, only delivery. Events come in from outside (webhooks, API calls, sensors, schedules, file changes), from inside (CLI, UI, an agent) or from another pipeline (§3.14). They pass through a small graph of nodes and are written to a destination: a database, a file or an API. An event counts as finished when Pipo can show it was delivered.

Pipo has four parts:

Why Pipo (rather than n8n and similar tools)#

• Simple. It's built for personal and small-team pipelines, and developer experience comes first. One file describes one pipeline, the file is the source of truth, and there's nothing hidden in a database.
• Agent-first. An agent can write pipelines, run them, step in while they're running and step out again, all through the same documented interfaces a human uses. It needs no workarounds.
• Resilient. Every pipeline is an isolated process with its own journal. A crashing pipeline, or even a crashing engine, does not take the others down.
• Delivery-oriented. Success is defined by verified delivery at the destination, not by "the code ran".

Non-goals#

• Not a human-approval or BPM workflow tool: pipelines have no approval steps. (A human does approve the pipelines agents write, §9.4.)
• Not multi-tenant or hosted. It's a local or self-hosted tool.
• Agents don't change engine code at runtime. They change .pipo files through a controlled protocol (§9).
• No exactly-once delivery. Pipo guarantees at-least-once delivery, plus idempotency keys (§7.3).

2. Core concepts#

2.1 Packet lifecycle#

received ──► rejected                      (input validation failed; terminal)
    │
    ▼
accepted ──► processing(node…) ──► writing ──► verifying ──► delivered   (terminal ✓)
                 │                    │            │
                 ├──► filtered        │            │                      (terminal, counted as success)
                 └────────────────────┴────────────┴──► dead_lettered     (terminal ✗)

A packet is written to the journal before its input is acknowledged. After a crash, the packet resumes from its last committed node.

A packet (or a fan-out copy) handed to the agent (then: agent, agent.on_stall: handle) is escalated: it stops where it was and waits, across restarts, until it is resolved: retried from that step, dead-lettered or dropped (§9.2, D50). It counts as pending.

A settled packet (delivered or filtered) can be rerun from any node it passed: it goes back in flight at that node with the data it had there, and that node and everything after it run again, down to the output, whose write is idempotent on the packet's key (pipo rerun, D79). The steps before it never run again.

2.2 Pipeline lifecycle#

stopped ─► starting ─► active ◄─► paused
                         │
                         ├─► draining ─► completed   (lifetime reached, or stop requested)
                         ├─► jammed                  (stall detected; still running, flagged)
                         └─► failed                  (an error policy chose `halt`)

• Pause stops processing but keeps accepting packets into the journal, up to buffer.max. Beyond that limit, HTTP inputs answer 503 Retry-After. On resume, the backlog is processed. Every pause records why it happened: manual, agent, error, stall or budget.
• Drain stops intake, lets in-flight packets reach a terminal state (bounded by drain_timeout) and then stops.

3. The .pipo file#

A .pipo file is YAML with a published JSON Schema. Plain YAML tooling works out of the box (§10.1), agents write it reliably, and the Phase 2 visual editor can load and save it without losing anything.

3.1 Top-level keys#

pipo: 1                 # required — spec version
name: people-intake     # required — unique per engine, [a-z0-9-]
description: ...        # optional
fn: ./people.fn.ts      # optional — module of user functions (§3.6)
secrets: {...}          # optional — named secret references (§3.7)
lifetime: {...}         # optional — when the pipeline ends (§3.8)
concurrency: 4          # optional — packets processed in parallel (default 4; 1 = FIFO)
buffer: { max: 10000 }  # optional — max accepted-but-unprocessed packets
errors: {...}           # optional — default error policy (§3.9)
input: {...}            # one input, or:
inputs: {...}           #   several, as a map of input name → input (§3.3.1); exactly one of the two is required
nodes: {...}            # optional — map of node id → node
output: {...}           # required
delivered: {...}        # optional — delivery verification (§3.10); default `check: ack`
agent: {...}            # optional — what agents may do with this pipeline (§9.3)
agent_budget: {...}     # optional — cost caps for agent nodes (§3.11)
retention: {...}        # optional — how long journal data is kept (§3.12)

The same pattern repeats everywhere: verb: connector, then with: for that connector's settings. input.via: http · nodes.x.tap: log · output.to: sqlite · delivered.check: record_exists. Each connector publishes the JSON Schema for its with: block, so validation and autocomplete adapt to the connector you choose.

3.2 Expressions and templates#

Expressions appear in validate, filter, route, loop.until and lifetime.until. They use a safe subset of JavaScript expression syntax. jsep turns the text into a syntax tree. Pipo then evaluates that tree with its own evaluator, which accepts only the node types it allows. JavaScript's eval and Function are never used. jsep sits behind an internal parse() interface and is covered by a conformance test suite, so the parser can be replaced later without changing how expressions behave.

• Allowed: literals (string, number, boolean, null, array, object); member and index access (data.a.b, data.items[0], optional chaining ?.); the operators + - * / %, == != === !== < <= > >=, && || ! and ??; in; the ternary a ? b : c; and calls to the helper functions listed below.
• Not allowed: assignment, ++/--, statements, arrow functions, new, this, method calls on values (write len(x), not x.length()), and any global other than the context variables.
• Semantics:
  • Member access on a missing or null value gives undefined. It never throws, so data.a.b is safe even when a is missing.
  • == and === both compare strictly and deeply, with no type coercion: 36 == "36" is false and [1] == [1] is true.
  • x in y tests membership: an array containing x, a string containing the substring x, or an object having the key x.
  • Only own properties are visible. Prototype members such as toString are undefined.
• Limits: the source length (2,000 characters), the syntax-tree depth (32) and the number of nodes (256) are capped. Expressions can't loop, so evaluation time is bounded by their size. matches() patterns are capped at 200 characters. matches() runs on the runner's regular-expression engine (Rust's regex crate), which has no lookaround or backreferences: such a pattern is an error ("Invalid regular expression…"), as is any pattern that fails to parse; pipo check and pipo compile don't run patterns.
• Escaping: in templates, $${ writes a literal ${.
• pipo check parses every expression at validation time and reports unknown identifiers and helper functions with their exact position in the file.

Templates are strings containing ${expr}. When a whole value is a single ${expr}, the result keeps its type (number, object and so on) instead of being turned into a string.

Helper functions: exists(x), len(x), size(x) (bytes), type(x), lower(s), upper(s), trim(s), matches(s, re), default(x, y), json(x), now() (epoch ms), iso(ms), duration("5m").

Context variables:

3.3 input#

Every pipeline has at least one input. An input defines how packets arrive, how they're parsed and what makes them valid. Most pipelines have one, written as input:; several are declared under inputs: (§3.3.1).

input:
  via: http                 # http | schedule | watch | push | system | telegram | pipeline | poll | tail | command | crawl
  with: {...}               # connector settings
  format: json              # json | text | csv | form | bytes
  schema: ./person.schema.json   # optional JSON Schema check, run before `validate`
  validate:                 # every rule must be true
    - exists(data)
    - data.age > 30
  on_invalid:               # deterministic failure: never retried
    respond: 422            # http only
    message: "Rejected: rule '${error.rule}' failed"

Each input's manifest carries a sample: what one packet's data looks like (watch, system, telegram, tail, command, crawl; schedule sends its payload; poll sends what the URL answers). http and push have none, since the sender decides, unless schema names a file (D70).

By default, http answers 202 {"packet_id": "..."} once the packet is in the journal. With respond: delivered, it waits for delivery up to timeout and returns the final state.

3.3.1 Several inputs#

inputs: is a map from input name to input, each written exactly like input:. input: is shorthand for an inputs: map with one input named input, so a file with one input never changes shape and from: input keeps working.

inputs:
  webhook:
    via: http
    with: { path: /people }
    schema: ./person.schema.json
  nightly:
    via: schedule
    with: { cron: "0 3 * * *", payload: { backfill: true } }
nodes:
  fetch:
    from: nightly
    transform: http
    with: { url: "https://example.com/people" }
  clean:
    from: [webhook, fetch]     # fan-in: each packet flows through on its own
    transform: fn.clean
output:
  from: clean
  to: sqlite
  with: { path: ./people.db, table: people }

• Names follow node ids ([A-Za-z_][A-Za-z0-9_-]*) and share their namespace: an input can't have the name of a node, nor be called output (P061). Nodes and the output take packets from an input by its name.
• Each input is its own edge. It has its own format, schema, validate and on_invalid, its own respond (http), and its own durable state in the journal (the watch snapshot, the telegram update offset), so adding a second input never resets the first. meta.input names the input a packet came from, so a route can tell them apart; to give packets from different inputs one shape, put a transform after each before they meet.
• Shared by the pipeline. One journal, one buffer.max, one concurrency, one lifetime (max_packets counts packets from every input) and one pause: while the pipeline is paused, every input still journals what it receives. When the buffer is full, each input pushes back in its own way: http answers 503, telegram and watch wait and try again, a schedule tick or a system sample is skipped, push and pipeline answer unavailable.
• http inputs share the runner's one listener (/in/<pipeline><path>), so two of them need different paths or methods, and listen, if set on more than one, must be the same port. Two telegram inputs of one pipeline can't poll the same bot. Either clash is P062.
• push (pipo push, MCP, the dashboard) names the input with input; it may be left out when there is one input, or exactly one via: push input.
• A pipeline has at most 16 inputs.

3.3.2 Active inputs: poll, tail, command, crawl#

Most inputs wait for something to arrive. These four go and get data themselves while the runner runs, and they remember where they got to in the journal (like watch and telegram), so a restart neither repeats nor loses what they already handed over (D87).

Exactly once, through the cursor. Each packet commits the input's position (the keys seen, the cursor, the byte offset) in the packet's own journal transaction. After a crash, an active input starts again from the last committed position: whatever wasn't journaled is fetched again, and whatever was is not. command is the exception: a program's output has no position Pipo can return to (see below).

Backpressure. When the runner can't take a packet (buffer full, draining), poll and crawl stop the current fetch or crawl and leave the rest for the next one, tail waits and reads the same line again, and command stops reading the program's output, so the program blocks on its next write.

poll#

input:
  via: poll
  with:
    url: https://api.github.com/repos/acme/app/releases
    every: 10m                     # or cron: "*/10 * * * *" (exactly one, P038)
    headers: { Authorization: "Bearer ${secrets.github}" }
    items: data                    # the list in the response; each element is one packet
    key: item.id                   # what identifies an element
    emit: new                      # new | changed | all

• When. every fetches at start and then every interval; the time of the last fetch is kept, so a restart waits out the rest of the interval rather than fetching again. cron fetches at each tick (UTC). Ticks missed while a fetch runs or the runner is down are skipped. An answer of 429 or 503 with Retry-After postpones the next fetch until then.
• The request. url (required, http or https), method (default GET), headers, body, timeout (default 30s) and success (statuses that count as success; default 200-299; 304 Not Modified is always "nothing new"). url, headers and body are templates rendered for each request, with env, secrets and cursor (below). The input sends If-None-Match and If-Modified-Since when the last answer had an ETag or Last-Modified, and User-Agent: pipo/<version> unless headers sets one.
• parse decides what data is: json, text, feed (RSS 2.0, RSS 1.0, Atom or JSON Feed: {title, link, entries}, the entries normalized as below), html (a page, below) or auto (default: by Content-Type; a JSON Feed or an RSS/Atom document is a feed). A body that doesn't parse as asked is an error for that fetch, logged with a hint.
• items is an expression over data, response ({status, url, headers}) and cursor that yields the list to split into packets. Each element becomes one packet. Without items, the whole answer is one element (with parse: feed, the entries are). A value that isn't a list is one element; null or undefined is none.
• key is an expression over item (the element) and data that names an element. The default is item.id for a feed, the url as written for a whole answer (the template, so a secret or the cursor in it never becomes the key; a later page's own URL), and a hash of the element's content otherwise. meta.source is the key, and meta.trigger is poll.
• emit: new sends an element whose key hasn't been seen before; changed also sends one whose content differs from the last time its key was seen; all sends every element of every fetch (no dedupe; after a crash during a fetch, its elements may be sent again). The default is changed for a whole answer and new otherwise, so a plain url + every emits the answer each time it changes.
• backfill (default true): on the first fetch, emit what is already there. false records it as seen and emits only what appears later (a notifier, not an import).
• remember (default 10000) caps the keys kept. Every fetch refreshes the keys it sees, so only keys that stopped appearing are forgotten, oldest first (keys last seen in the same fetch go in key order); if one comes back, it counts as new.
• When the runner can't take a packet, the fetch stops there; what it didn't hand over counts as unseen and is fetched again 5 s later (every) or at the next tick (cron). A fetch that fails (unreachable, a status outside success, a body that doesn't parse) is logged with a hint and tried again at the next fetch.
• cursor is an expression over data, items, response and cursor (the previous value) evaluated after each page. Its last value that isn't null or undefined is saved once the fetch's packets are journaled, and is ${cursor} in url, headers and body of the next fetch (null before the first: write ${cursor ?? '1970-01-01T00:00:00Z'}). Use it for "since" parameters; items is the page's list of elements.
• next is an expression over data and response that yields the next page's URL (relative URLs resolve against the current one). The input follows it within the same fetch, up to max_pages pages (default 10), and stops at null, undefined, "" or a URL it already fetched this time.
• parse: html gives data = {url, status, title, text, links, html}: text is the visible text (scripts and styles removed, whitespace collapsed), links the page's absolute http(s) links. select (a CSS selector, html only, P065) narrows text and html to the matching elements, joined with a blank line; without select, html is null. url + parse: html + select + the default emit: changed watches one part of a page for changes.
• Feed entries are {id, title, link, published, updated, summary, content, author, feed: {title, link}} (ISO 8601 dates; missing fields null; id falls back to link, then to a hash of the entry).
• Bodies larger than 10 MiB are refused for that fetch.

tail#

input:
  via: tail
  with: { path: /var/log/nginx/access.log, parse: text }

• Follows one file (path, relative to the pipeline file) and sends each complete line (ending in \n) as a packet; a final line without \n waits until it has one. parse: text (default) gives data = {line, path}; parse: json gives the line parsed as JSON (a line that isn't JSON is logged and skipped). meta.trigger is tail, meta.source is <path>:<byte offset of the line>.
• from: where a first start begins, end (default: only lines written from now on) or start (the whole file). Later starts resume after the last line journaled.
• Rotation and truncation. The file's identity (device and inode) is saved with the offset. When the path names a new file (rotated by rename), the old one is read to its end first, then the new one from its start; when the file shrinks below the offset (truncated), it is read again from the start. A file rotated while the runner was down is read from the start; the end of the old one is lost.
• The file is checked every 250 ms. A missing file is waited for (and logged once). Lines longer than 1 MiB are cut there and sent with truncated: true.

command#

input:
  via: command
  with: { command: journalctl, args: [-f, -o, json, -u, nginx], parse: json }

• Runs command with args (no shell, like an exec step: P058 warns about spaces, and an untrusted pipeline file can't run one, P052), cwd (relative to the pipeline file; default its folder) and env (added to the runner's environment), from the runner's start to its stop, in its own process group (SIGTERM, then SIGKILL 2 s later).
• Each line the program writes to stdout is one packet: parse: text (default) gives data = {line}; parse: json gives the line parsed as JSON (a line that isn't JSON is logged and skipped). stderr lines go to the runner's log as warnings. meta.trigger is command, meta.source is the command.
• restart: always (default) starts the program again whenever it exits, on-failure only when it exits with an error or a signal, never leaves the input idle. Restarts wait 1 s, doubling to 60 s, and reset to 1 s once the program has run for a minute. Each exit is logged with its code and the program's last stderr line.
• At most once. A line the program printed but the runner hadn't journaled when it crashed is lost, and a restarted program starts over. Where it matters, use the program's own position (journalctl --cursor-file, kubectl get --watch --resource-version).

crawl#

input:
  via: crawl
  with:
    start: https://docs.example.com/
    every: 1d
    within: https://docs.example.com/guide/
    depth: 3

• Each crawl starts at start (a URL or a list), follows links breadth first to depth (default 3; start is depth 0), and fetches at most max_pages pages (default 500), one at a time, delay apart (default 1s). Only URLs that start with one of within (a URL prefix or a list; default each start URL's origin) are followed, minus any matching one of exclude (regular expressions). URLs are normalized (fragment removed, scheme and host lower-cased).
• every crawls at start and then every interval after the last crawl ended; cron at each tick. A crawl that is cut short by a crash or a stop is crawled again at the next start.
• Each HTML page (text/html, up to 10 MiB) becomes data = {url, status, title, text, links, html, depth} as with poll's parse: html, including select. Other content types are skipped. meta.trigger is crawl, meta.source is the page URL (after redirects).
• emit: changed (default) sends a page that is new or whose title, text or links changed since the last crawl; all sends every page of every crawl. remember (default 10000) caps the pages kept, as in poll.
• Manners. robots.txt is read once per host per crawl and obeyed (User-agent: pipo or *, Allow/Disallow with * and $, the longest match wins; Crawl-delay raises delay); robots: false turns that off, for your own sites. Requests carry User-Agent: pipo/<version> and headers. A 429 or 503 waits for Retry-After (up to 5 minutes) once, then skips the page. JavaScript isn't run, so pages that build their content in the browser crawl as what their HTML says.

3.4 nodes#

nodes is a map from node id to node. Each node has:

• from (required): where its packets come from. This can be an input (input, or a name from inputs:), a node id, a route branch (triage.urgent) or a list of these (fan-in: each packet flows through on its own).
• Exactly one kind key, listed in the table below.
• with, on_error, loop and label (optional).

Connections are declared once, via from. Downstream links are worked out from them. When one node feeds several nodes (fan-out), each branch gets a copy of the packet, and the packet counts as delivered only when every branch reaches a terminal state.

Agent nodes use with: { model, prompt, schema, timeout, max_tokens }. Their cost is limited by the pipeline's agent_budget (§3.11). The schema (JSON Schema) is required: the agent's output is checked against it, and a non-matching result counts as an error under the node's on_error policy. This keeps the uncertainty of the agent's output contained inside the node. timeout defaults to 60s and max_tokens to 4096; the provider's API key and price table come from the engine config (D36).

CLI agents. claude_code (Claude Code), codex (the OpenAI Codex CLI), pi and opencode hand the prompt to a coding-agent CLI installed and logged in on the machine, which runs as the user, on the user's own account or subscription (D67). They use with: { model, prompt, schema, timeout, cwd, allow_tools }. model is optional (the CLI's own default when left out). timeout defaults to 5m. The schema is required and checked the same way: Claude Code and Codex are made to answer in its shape, and pi and opencode are asked for JSON in the prompt. Each call runs in a fresh, empty folder with the CLI's tools off, or read-only where it can't turn them off. cwd (relative to the pipeline file) names the folder to run in instead, and allow_tools: true lets the CLI use its tools (edit files, run commands) without asking. A pipeline using a CLI agent that isn't installed or logged in refuses to start, with the reason and the command that fixes it.

Exec nodes. tap: exec and transform: exec run a program installed on the machine (ffmpeg, ImageMagick, pandoc, a script), as the user (D71). They use with: { command, args, cwd, env, stdin, timeout, success, result, outputs }, and every string is a template. command is the program: a name found on PATH, or a path relative to the pipeline file. args is a list, and each item goes to the program as one argument. There is no shell, so a file name with spaces or quotes can't break the command. cwd (relative to the pipeline file, which is the default) is where it runs. env adds environment variables, stdin is fed to it, and timeout (default 5m) kills it. success lists the exit codes that count as success (default [0]); any other exit is a node error carrying the end of stderr. outputs lists the files the program must write, relative to cwd like the program's own relative paths: their folders are created first, a missing one after the run is a node error, and a transform returns them in files. result decides what a transform's data becomes: info (default) is {exit_code, stdout, stderr, duration_ms, files} with the last 64 KB of each stream, text is stdout, and json is stdout parsed (both up to 1 MB). A pipeline whose command isn't installed refuses to start, with a hint. If the runner crashes before the step commits, the program runs again, so name output files after the packet (${meta.packet_id}, or the input file's name) and a rerun overwrites rather than duplicates.

render:
  from: input            # a watch input with read: path
  transform: exec
  with:
    command: ffmpeg
    args: [-y, -loop, "1", -i, ./cover.jpg, -i, "${data.path}", -c:v, libx264, -tune, stillimage,
           -c:a, aac, -pix_fmt, yuv420p, -shortest, "./out/${data.name}.mp4"]
    outputs: ["./out/${data.name}.mp4"]
    timeout: 10m

Loops. The graph may not contain cycles, with one exception: a node may declare a bounded loop back to an earlier node.

loop:
  back_to: draft            # an upstream node id (or the node itself)
  until: data.approved == true
  max: 3                    # required
  then: dead_letter         # what happens when `max` is reached without `until` being true

max is how many times a packet may be sent back. meta.iteration counts how many times it has been sent back so far. The count belongs to the packet (or fan-out copy), not to one loop: a packet that passes two loops starts the second with the first one's count, and both max checks see the running total. Any cycle not declared through loop.back_to is a validation error.

3.5 output#

Every pipeline has exactly one output, by design (D78): it is where a packet is written once and verified, so "delivered" always means one thing. Side effects that need no verification belong in tap nodes. When data has to reach a second destination with the same guarantees, send it to another pipeline (to: pipeline, §3.14), whose own output writes and verifies it. The envelope is the same for every connector. Only with: differs, and it is checked against that connector's schema.

output:
  from: <node | [nodes]>
  to: sqlite                # sqlite | file | http | stdout | telegram | pipeline | (phase 2: postgres, s3, queue, mqtt)
  batch:                    # optional — write several packets in one go (§3.5.1)
    size: 100
    within: 2s
  with: {...}               # connector settings
  validate: [...]           # last check before writing (deterministic, no retry)
  on_invalid: {...}
  on_error: {...}           # failures at write time (retryable)

3.5.1 Batching#

batch collects packets that are ready to be written. They are flushed together when size packets have collected, or within has passed since the first packet arrived, whichever comes first. A flush is one write: a single transaction for sqlite, a single append for file.

• Packets stay separate. Batching affects only how packets are written. Each packet keeps its own packet_id, its own delivered check, its own state and its own place in the dead-letter queue.
• Failure isolation. If a batch write fails, it is retried under on_error. If the retries run out, the batch is split in half and retried, repeatedly, until single packets remain. Only the packets that still fail go to the dead-letter queue, so one bad packet can't take 99 good ones with it.
• Validation (output.validate) runs per packet before a packet joins a batch. Invalid packets never enter a batch.
• Pause, stop and drain flush any partial batch straight away.
• Crash safety. A packet is journaled as waiting in the batch before it joins it, so a crash loses nothing. After a restart, waiting packets join a new batch (its within starts again), and anything already written is skipped by the idempotency key. within defaults to 1s (D20).
• Supported in v1: sqlite and file. On http and stdout, pipo check rejects batch. A batch mode for http (one request carrying an array) is planned for Phase 2.

3.6 User functions#

fn: ./people.fn.ts points to a TypeScript or JavaScript module. Each exported function is available as fn.<name>:

// people.fn.ts
type Meta = { packet_id: string; pipeline: string; version: number; node: string; [key: string]: unknown };
export function textTransformer(data: any, meta: Meta) {
  return { ...data, name: data.name.trim(), bio: (data.bio ?? "").slice(0, 500) };
}

Functions run in an embedded JavaScript engine (QuickJS) inside the pipeline's runner, as plain JavaScript. A module gets the ES built-ins QuickJS has (JSON, Math, Date, Promise, Map, Set, RegExp, typed arrays, …), console (to the runner log), and setTimeout/clearTimeout (and setInterval/clearInterval), and nothing from Bun or Node: no node:/bun: imports and no Bun, process, require, Buffer or fetch. A module may be TypeScript and may import local files and installed packages: pipo compile bundles it (with Bun.build) into one self-contained ES module, and pipo check reports P059 when it can't (§5). A function gets (data, meta) and returns the new data, or a Promise of it; values cross as JSON. Each version pins the code it was compiled with, so a packet pinned to an older version runs that version's functions. console.log, console.info, console.debug, console.warn and console.error write to the runner's log. Calls run concurrently, as in Bun: while one awaits a timer (await new Promise((r) => setTimeout(r, 500)), say for a rate-limit backoff), the runner's other workers' calls run. A call that takes longer than 30 s from its start until its result, time spent waiting on timers included, is stopped and fails like any step error; so is a call whose Promise waits on nothing that could ever settle it. Timers belong to the call that set them and are cancelled when it ends, so nothing runs in the background between calls. Anything that needs the system (files, network, programs) belongs in a connector or an exec step. Functions are trusted code. A crash affects only that pipeline. Function modules that arrive through a template you didn't write are untrusted until you run pipo trust <template>. pipo check and pipo start refuse untrusted modules (§11).

3.7 Secrets#

Secret values never appear in a .pipo file. They are declared by reference and resolved by the runner at runtime:

secrets:
  telemetry_token: op://Pipo/telemetry/api-token   # 1Password (recommended)
  intake_token: env:PIPO_INTAKE_TOKEN              # environment variable

• Supported providers: op:// (1Password) and env:.
• Secrets can only be used inside with: blocks, as ${secrets.<name>}.
• They are redacted from the journal, logs, the UI and every response to agents.
• pipo check warns about literal values that look like credentials.

3.8 lifetime#

lifetime:
  ttl: 30m                       # wall-clock lifetime from start
  max_packets: 10000             # stop accepting after N accepted packets
  until: stats.delivered >= 1000 # stop when this becomes true
  on_end: drain                  # drain (default) | stop
  drain_timeout: 2m

Reaching any of ttl, max_packets or until journals a pipeline.lifetime event whose reason names it (ttl, max_packets, until), then the pipeline drains or stops. If nothing is set, the pipeline runs until it is stopped. The engine can also have its own TTL (§7.5), which drains every pipeline.

pipo start <name> --ttl 30m (and pipo restart --ttl) replaces ttl for that start only. It counts from the same anchor (§7.4), and on_end and drain_timeout still come from the file. A later start without --ttl uses the file's ttl again (D57).

3.9 Error policies#

Every error hook has the same shape. The top-level errors: block sets defaults, and any step can override them.

on_error:
  retry: 3                  # attempts after the first one
  backoff: exponential      # fixed | exponential
  delay: 500ms
  max_delay: 1m
  then: dead_letter         # what happens once retries run out (see below)
  message: "…${error.message}…"

on_invalid (for validation failures) is never retried, because the same input would fail the same way again. At input.on_invalid, the packet was never accepted, so it is always rejected (and the sender told: respond for http). The only then there is agent, which also tells the agent; it can't make the packet wait. Any other then is P033 (D50, D80).

3.10 delivered: delivery verification#

When the output connector acknowledges a write, the packet is written. The delivered block decides when it is delivered.

delivered:
  check: record_exists      # what to verify (must be supported by output.to — see matrix)
  with: {...}
  within: 10s               # keep re-checking until this deadline
  on_fail: {...}            # error policy
  stall:                    # pipeline-level health ("jammed")
    after: 5m               # no delivery progress for this long while packets are pending
    then: pause             # notify | pause | agent (flag it as notify does and tell the agent, D50)
    message: "…"          # rendered with stats.* and stall.*; see D23 for when it fires and clears

Compatibility matrix. pipo check rejects any combination that isn't listed here:

with: fields per check. All values are templates, rendered per packet.

Check queries run on a read-only connection, so a check can never change data. Put data in params, never into the SQL text. A check that cannot run (bad SQL, unreadable file, unreachable follow_up URL) counts as a failed attempt and is retried until within; its message becomes the error.

Example of a rejection:

people-intake.pipo:61:10  error  P031  delivered.check 'file_exists' is not supported by output 'sqlite'
  supported: ack, record_exists, row_count, query, external, none

3.11 agent_budget#

Agent nodes cost money, so the budget caps that cost for the whole pipeline. It has two limits:

agent_budget:
  per_day: 5.00             # USD across all agent nodes in this pipeline, per calendar day
  reset_at: "00:00"         # optional — when the day starts, in the engine's time zone (engine.timezone)
  per_packet: 20000         # tokens (input + output) one packet may use across all agent nodes and loop passes
  warn_at: 80%              # emit a `budget.warning` event

• Per-packet cap reached: that packet goes to the dead-letter queue with error code budget.packet. The pipeline keeps running.
• Daily cap reached: the pipeline pauses with the reason budget. It keeps journaling new packets up to buffer.max, so no data is lost. It resumes automatically when the next budget window opens, or straight away with pipo resume (which asks for confirmation, because it goes over the cap).
• Caps are checked before each call, not reserved. Calls already in flight when a cap is reached (up to the pipeline's concurrency) still finish and are counted, so the day's spend (and the engine cap, D58) can end up above the cap by what those calls cost.
• Why a calendar day rather than a rolling 24 hours: a calendar day gives one clear moment to resume, matches how providers bill and how people think about "$5 a day", and avoids the pause/resume flapping a rolling window causes as old spending ages out minute by minute.
• Cost is calculated from the token counts each provider reports and a price table in the engine config (agents.<provider>.pricing). A CLI agent's cost is what the CLI reports (Claude Code, pi and opencode report USD), else its tokens priced from that table, else $0, so a cap can't stop a CLI agent that reports nothing (D67). pipo check doesn't ask CLI agents for a cap (no P053). The UI and pipo status show spending to date.
• An engine-wide cap (engine.agent_budget.per_day, USD, in the home's config.yaml) can also be set. It applies on top of the pipeline caps, to the spend of every pipeline of the Pipo home together, per calendar day in engine.timezone (from midnight). It is checked like the daily cap, before each call. When the total reaches it, the next agent call of any pipeline of the home is refused the same way: the packet is held and that pipeline pauses with the reason budget (cap: engine) until the next engine day, or pipo resume overrides the engine cap for that pipeline until the day ends. To allow more today, raise the cap in config.yaml and restart the pipeline. pipo status and GET /api/agent-budget show the day's spend per pipeline and the total against the cap (D58).

3.12 retention#

retention:
  data: 7d                  # payloads of delivered/filtered packets
  trail: 30d                # packet metadata + event trail (states, timings, errors)
  rejected: 3d              # payloads of rejected input
  dlq: forever              # dead letters stay until replayed or purged (`pipo dlq purge`)

These are also the defaults. The journal is compacted once a day. Data from packets that are still pending, and from version history, is never removed.

3.13 Chat bots#

A pipeline can read from a Telegram bot and send with one (D69). Bots are set up once per Pipo home, in <home>/bots.json, from the dashboard's 🤖 Bots page (or /api/bots, §8). Each bot has a name, a token (the raw token from @BotFather, or a reference: op://…, env:…), an allow list and an optional poll_every. One bot is the default.

input:
  via: telegram             # messages to the default bot
nodes:
  alert:
    from: input
    tap: telegram           # send mid-pipeline, here with another bot
    with: { bot: alerts, chat_id: 123456, text: "new message: ${data.text}" }
output:
  from: alert
  to: telegram              # no chat_id: replies to the chat the message came from
  with: { text: "Thanks, got it" }

• Which bot. Every telegram with: takes bot (a name from bots.json, never a template) or token (inline, e.g. ${secrets.tg_token}). With neither, the default bot is used. A pipeline that names a missing bot, or has no bot at all, refuses to start and says how to add one. Tokens are resolved at start and redacted like secrets (§3.7). A change to bots.json applies from a pipeline's next start.
• Input. The runner long-polls getUpdates only while the pipeline runs: one request waits up to poll_every (default 25s, at least 1s) and returns as soon as a message arrives. Only messages whose chat or sender id is in allow (with.allow replaces the bot's list) become packets. Others are ignored, and the log names the sender's ids so they can be added. With no allow list, every message is ignored. The update offset is saved in the transaction that journals the packet, so each message becomes exactly one packet across crashes. data is {message_id, date, chat_id, chat_type, from: {id, username, name}, text, file}; text is the text or the caption. meta.trigger is telegram and meta.source the chat id.
• Files. An attached document, photo (the largest size), audio, voice, video or animation is downloaded to <home>/pipelines/<name>/files/<unique id>-<name> and described in data.file: {kind, file_id, name, mime_type, size, path}. download: false keeps only the description. Telegram doesn't let bots download files over 20 MB; such a file comes with path: null and a warning in the log.
• Sending (tap: telegram, to: telegram). chat_id defaults to the chat a telegram input's packet came from; with another input it is required (P057). text defaults to data (as JSON when it isn't a string), cut at Telegram's 4096 characters. photo or document (a path relative to the pipeline file, or an http(s) URL) sends a file, with text as its caption (1024 characters). parse_mode is Markdown, MarkdownV2 or HTML; it is off by default, so text from data can't break the formatting. A 429 is retried up to 3 times after Telegram's retry_after, then the step's on_error applies. The output's result is {chat_id, message_id}, and its only delivery checks are the universal ones.
• One poller per bot. Telegram lets one reader take a bot's updates. A runner records the bot id it polls in its registry entry (telegram_bot), and a second pipeline polling the same bot refuses to start, naming the first. pipo check on several files warns about it (P056). Sending needs no poll, so any number of pipelines can send with one bot.

3.14 Chains: one pipeline feeding another#

A pipeline's output can be another pipeline's input. This is how one stream reaches several verified destinations, and how a large pipeline is split into parts that are versioned, paused and restarted on their own. Both ends declare the link (D77):

# people-intake.pipo: the sender
output:
  from: clean
  to: pipeline
  with: { pipeline: people-store }
delivered:
  check: downstream          # optional: wait until people-store has delivered it
  within: 2m

# people-store.pipo: the receiver
inputs:
  intake:
    via: pipeline
    with: { from: [people-intake] }
    schema: ./person.schema.json
  webhook:
    via: http
    with: { path: /people }

• Both ends declare it. The receiving input lists the pipelines allowed to feed it in with.from; anything else is refused (rejected, nothing journaled). One pipeline can't list the same sender in two inputs, and can't feed itself (P063). pipo check on several files (a folder) warns when a pipeline names another one in the same check that doesn't name it back, and when chains form a cycle (P064).
• Same home. The sender finds the receiver through its registry entry (<home>/run/<name>.json) and writes over its control socket (op deliver, §7.2). Both must run under the same Pipo home; distributed engines (§12) will carry chains across machines.
• Handed over exactly once. The receiver journals the packet (or its rejection) with the sender's name and key (meta.key) in one transaction before it answers. A write repeated after a crash or a retry gets the packet it already made (duplicate: true). The sender then commits the write with the result {pipeline, packet_id, duplicate}. So across the hand-off, each sender unit (a packet or a fan-out copy) becomes exactly one receiver packet.
• When the write fails. A receiver that isn't running, is draining or stopping, or has a full buffer, is the write error downstream_unavailable, with a hint to start it. The output's on_error retries it, packets wait in the sender's journal, and delivered.stall sees a lasting outage. Nothing starts the receiver for you. A packet the receiver's input rejects (schema, validate) is the write error downstream_rejected with the receiver's message; the receiver keeps it as rejected and applies its own on_invalid. The same data is rejected again on every retry, so check what matters earlier with output.validate.
• When it is delivered. With the default check: ack, the sender's packet is delivered once the receiver has journaled it: from then on, the receiver's journal owns it. check: downstream waits for the receiver's packet to settle, re-checking until within (default 10s): delivered passes. dead_lettered, filtered or rejected, or a packet the receiver no longer has, fails at once with the receiver's reason, and on_fail applies. A receiver that isn't running is a failed attempt, re-checked until within. Give within room for the receiver's own retries.
• What the receiver sees. meta.trigger is pipeline, meta.source the sender's name, and meta.upstream {pipeline, packet_id, key, depth}: the sender's packet and its key, and how many hand-offs the data has made. A packet past depth 16 is refused (rejected, nothing journaled), so a cycle of pipelines can't run forever.
• Replaying across a chain. A sender's packet dead-lettered by check: downstream is about the receiver's packet, which the receiver owns: fix the cause there and replay the receiver's dead letter first, then the sender's. The sender's replay writes again, the receiver answers with the packet it already has, now delivered, and the check passes. Replaying only the sender finds the same dead-lettered packet and fails again.
• Pause, change and test. A paused receiver still journals what it receives, like any input. Changing output.with.pipeline is a live change; changing output.to or a receiver's from list needs a restart (D38). The dry run (§9.3) and pipo test mock the output as they always do: the hand-off is rendered and recorded, nothing is sent.

4. Full example#

The refined version of the example in idea.txt. The cycle in the original (node1 ↔ node2) has been removed, and every node now reaches the output.

pipo: 1
name: people-intake
description: Accept people from a webhook, normalise them, report telemetry and store them in SQLite.

fn: ./people.fn.ts

secrets:
  intake_token: op://Pipo/people-intake/webhook-token
  telemetry_token: op://Pipo/telemetry/api-token

lifetime:
  ttl: 30m
  on_end: drain

errors:
  retry: 2
  backoff: exponential
  delay: 500ms
  then: dead_letter

input:
  via: http
  with:
    path: /people
    method: POST
    auth:
      header: X-Pipo-Token
      equals: "${secrets.intake_token}"
  format: json
  validate:
    - exists(data)
    - type(data.name) == "string" && len(trim(data.name)) > 0
    - data.age > 30
  on_invalid:
    respond: 422
    message: "Rejected: rule '${error.rule}' failed"

nodes:
  logger:
    label: Logger
    from: input
    tap: log
    with:
      level: info
      message: "Received ${meta.packet_id} via ${meta.trigger} from ${meta.source}"

  normalize:
    from: logger
    transform: fn.textTransformer

  telemetry:
    label: Telemetry
    from: normalize
    tap: http
    with:
      method: POST
      url: https://telemetry.example.com/v1/events
      headers:
        Authorization: "Bearer ${secrets.telemetry_token}"
      body:
        event: pipo.packet.processed
        pipeline: "${meta.pipeline}"
        version: "${meta.version}"
        packet_id: "${meta.packet_id}"
        node: "${meta.node}"
        trigger: "${meta.trigger}"
        received_at: "${iso(meta.received_at)}"
        latency_ms: "${now() - meta.received_at}"
        payload_bytes: "${size(data)}"
    on_error:
      retry: 3
      then: continue            # telemetry must never block delivery

output:
  from: telemetry
  to: sqlite
  with:
    path: ./data/people.db
    table: people
    create: true
    mode: upsert
    key: id
    columns:
      id: "${meta.packet_id}"
      name: "${data.name}"
      age: "${data.age}"
      bio: "${data.bio}"
      received_at: "${meta.received_at}"
  validate:
    - type(data.age) == "number"
    - len(data.name) <= 200
  on_invalid:
    then: dead_letter
    message: "Packet ${meta.packet_id} failed output rule '${error.rule}'; not written"
  on_error:
    retry: 5
    backoff: exponential
    max_delay: 1m
    then: dead_letter
    message: "Could not write ${meta.packet_id} to ${output.with.path}#${output.with.table} after ${error.attempts} attempts: ${error.message}"

delivered:
  check: record_exists
  with:
    where: { id: "${meta.packet_id}" }
  within: 10s
  on_fail:
    then: dead_letter
    message: >
      Delivery unverified: sqlite acknowledged packet ${meta.packet_id} (rowid ${result.rowid}),
      but no row with id=${meta.packet_id} exists in ${output.with.table} after ${error.elapsed}.
      Look for a rolled-back transaction, a trigger deleting rows, or another process replacing ${output.with.path}.
  stall:
    after: 5m
    then: pause
    message: >
      Pipeline jammed: ${stall.pending} packets pending and none delivered for ${stall.duration}.
      Oldest packet ${stall.oldest.packet_id} has been at '${stall.oldest.node}' for ${stall.oldest.age}
      (attempt ${stall.oldest.attempt}, last error: ${stall.oldest.last_error}).

agent:
  control: true
  actions: [pause, resume, replay, push]
  edit:
    - nodes.normalize
    - nodes.telemetry.with
    - errors

4.1 Agent node with a bounded loop and routing#

pipo: 1
name: ticket-triage

agent_budget:
  per_day: 3.00
  per_packet: 15000

input:
  via: http
  with: { path: /tickets }
  format: json
  validate:
    - len(data.subject) > 0

nodes:
  draft:
    from: input
    agent: claude_api
    with:
      model: claude-sonnet-5-5
      prompt: |
        Classify this support ticket (priority: high|normal) and draft a first reply.
        If reviewer feedback is present, address it.
        ${json(data)}
      schema: ./schemas/draft.schema.json
      timeout: 60s

  review:
    from: draft
    agent: claude_api
    with:
      model: claude-sonnet-5-5
      prompt: |
        Check the classification and the reply. Set approved=true only if both are correct;
        otherwise put concrete feedback in `feedback`.
        ${json(data)}
      schema: ./schemas/review.schema.json
    loop:
      back_to: draft
      until: data.approved == true
      max: 3
      then: dead_letter

  triage:
    from: review
    route:
      urgent: data.priority == "high"
      normal: else

  page:
    from: triage.urgent
    tap: http
    with:
      method: POST
      url: https://pager.example.com/hooks/support
      body: { ticket: "${data.id}", subject: "${data.subject}" }

output:
  from: [triage.normal, page]
  to: file
  with:
    path: ./out/tickets.jsonl
    format: jsonl

delivered:
  check: line_contains
  with: { value: "${meta.packet_id}" }

5. Validation (pipo check)#

Every finding is reported as file:line:col severity code message. The validator runs before any start or change, including changes proposed by agents.

6. CLI (pipo)#

Every command except pipo run supports --json for scripts and agents (pipo run is the foreground data plane and streams its logs to the terminal). If the engine isn't running, pipo starts it on demand.

On a terminal, human output is decorated: colour, ✔/▲/✖ symbols, a spinner while the CLI waits (starting the engine, starting or stopping a pipeline) and a banner for pipo ui. Anything else gets plain text, which is the stable format: --json, output that isn't a terminal, --plain (any command), PIPO_PLAIN=1, NO_COLOR, CI and TERM=dumb. pipo run, pipo logs, pipo attach, status --watch and the runner are never decorated (D74).

pipo run <file> [--listen <port>] runs one pipeline in the foreground without the engine, which is useful during development. Its HTTP input is served at the same /in/<pipeline><path> route the gateway uses, on the runner's own port. Ctrl-C drains the pipeline, and a second Ctrl-C stops it at once.

$ pipo status
PIPELINE        STATE    VER  UPTIME   IN/MIN  PENDING  DELIVERED  DLQ  LAST DELIVERY
people-intake   active   v3   12m04s       42        3      5,112    2  1s ago
ticket-triage   jammed   v1   2h10m         0       17        230    0  6m ago   ⚠ stalled at 'review'

7. Engine#

7.1 Architecture#

            ┌───────────────────────── pipod (control plane) ─────────────────────────┐
  webhooks ─┤ HTTP gateway /in/*   REST /api/*   MCP /mcp   SSE /events   UI /ui      │
  CLI, UI  ─┤ supervisor · scheduler · version store · proposal queue                 │
            └──────┬──────────────────────┬──────────────────────┬────────────────────┘
                   │ unix socket          │                      │
            ┌──────▼──────┐        ┌──────▼──────┐        ┌──────▼──────┐
            │ runner      │        │ runner      │        │ runner      │   data plane:
            │ people-…    │        │ ticket-…    │        │ …           │   one Rust process
            │ journal.db  │        │ journal.db  │        │ journal.db  │   per pipeline
            └─────────────┘        └─────────────┘        └─────────────┘

• The engine is built on Bun and uses Bun.serve, bun:sqlite and Bun subprocesses.
• The runner is a Rust binary (pipo-runner, crates/pipo-runner), small enough to run one per pipeline: about 9 MB resident when idle. TypeScript checks and Rust executes: the runner runs pipo compile (the CLI's checker) at start, on apply/rollback and when it validates a proposal, and stores each version's compiled form in the journal, so Bun runs only for the moment of a compile, never while packets flow. fn modules run in the runner's embedded QuickJS (§3.6). The binary also answers pipo test (pipo-runner test) and reads without a runner (pipo-runner read, D34). D73 has the details.
• One runner process per pipeline. The runner owns that pipeline's journal, its inputs (in detached mode), its schedules and its connectors. The engine restarts crashed runners with backoff.
• Engine config and restarts. The engine (pipod) reads config.yaml (an engine: map: ttl, detached, listen, host and public_url (D85), auth (D84), mail (D86), idle, env_allow, start_timeout, stop_timeout, restart, timezone, agent_budget, mcp (the agent endpoint's tokens, §9.2, and auto_approve, §9.4), workspace (where the builder saves new pipelines, D72); and an agents: map of provider settings: pricing, plus api_key and base_url for claude_api or command (the executable) for a CLI agent, D36, D67), writes its own entry to run/engine.json, and sends each runner's output to logs/<name>.log. Restart backoff and when the engine gives up are in D25.
• Gateway. With engine.listen set (0 picks a free port, which run/engine.json records), the engine serves /in/<pipeline>/… (proxied to the runner's own port), REST /api/*, SSE /events and the dashboard at /ui (§8, D40) on 127.0.0.1, or on every interface with engine.host (D85). With engine.auth: owner (the default) /api and /events need the owner's session or the engine key (D84). REST includes GET /api/pipelines/<name>/graph, the nodes, edges and per-node counters the UI draws, and GET /api/agent-budget, the day's agent spend of the home against engine.agent_budget.per_day (D58). Details in D28. An engine the CLI starts (pipo engine start, or any command that needs one) listens on port 4747 unless engine.listen or --listen says otherwise, so MCP clients and their tokens have a stable URL; PIPO_ENGINE_PORT changes that default, and when the port is taken the CLI warns and uses a free one (D83). pipod itself still has no gateway without listen.
• State lives in ~/.pipo/: config.yaml; owner.json, sessions.json and engine.key (D84); pipelines/<name>/journal.db (packets, events and every version's source and compiled form, D38); run/<name>.json (the runner registry, §7.2) and run/<name>.sock; logs/.
• By default the server binds to 127.0.0.1. Exposing it is an explicit choice (engine.host, which needs the owner's login and engine.public_url), and so is exposing an input without auth (public: true, D85).

7.2 Detached mode#

With pipo start --detached, or detached: true in engine config, a runner keeps running if the engine dies:

• Inputs other than HTTP (schedule, watch, system) fire from the runner itself.
• Runners enforce their own lifetime and stall detection while the engine is away. When the engine reattaches it follows them through the journal events rather than arming them again, so nothing fires twice (D30).
• Status events are written to the journal while the engine is away.
• The runner runs in its own session, so neither a signal to the engine's terminal nor the engine's death reaches it. Stopping the engine leaves detached runners running; pipo stop <name> stops one. Details in D30.

HTTP ports. Pipo never assigns ports automatically. A detached HTTP input gets its own port in one of two ways:

• in the file: input.with.listen: 8081
• when starting it: pipo start people-intake --detached --listen 8081. This overrides the file and is saved in the runner registry, so restarts and reattaches reuse the same port.

Before starting, the engine checks that the port is free (the CLI's start goes through it). If it isn't, the start is refused with an error that names the process holding the port, or the pipeline if another runner holds it. pipo check warns when two pipelines in the same project declare the same port. While the engine is up, the gateway also proxies /in/<pipeline>/… to the runner, so the input is reachable both through the gateway and directly on its port. While the engine is down, only the direct port works. An HTTP input without listen is unreachable until the engine returns.

Runner registry. Each runner writes ~/.pipo/run/<name>.json atomically and removes it on a clean stop:

{ "pipeline": "people-intake", "version": 3, "pid": 41822, "socket": "~/.pipo/run/people-intake.sock",
  "listen": 8081, "detached": true, "ttl": "30m", "started_at": "2026-10-03T09:12:44Z", "engine_id": "e_01J9…" }

ttl is there only when the start passed --ttl (D57). Like listen, the engine passes it again when it restarts a crashed runner, adopts a live one, or restarts one that died while no engine ran.

Control socket. Each runner listens on run/<name>.sock (owner-only) and writes its registry entry only once the socket listens. The protocol is newline-delimited JSON: a request {id, op, args} gets one response {id, ok: true, result} or {id, ok: false, error: {code, message, hint}} (an invalid_pipeline error adds diagnostics, as pipo check --json gives them, D60). The ops are hello (pipeline, version, pid, state, status, started_at: the handshake), status (adds stats, paused_reason, awaiting_ack, last_seq), pause (reason: manual | agent), resume, drain, stop, push (data, optional source, and input, the input's name, needed when there are several, §3.3.1; replies with the packet_id once the packet is journaled), deliver (from, key, data, upstream: {packet_id, depth}: a packet from another pipeline's to: pipeline output, §3.14; replies {packet_id, duplicate} once the packet is journaled), ack (packet_id, for the external check, §3.10), events (after_seq, limit: journal events with their seq, so a reattaching engine replays what it missed), the packet reads and DLQ ops of D33 (packets (state, limit, after), packet (id), dlq, and replay and purge (ids or all)), and the version ops of D38: versions, version (version), diff (from, to), apply (source, reason, by) and rollback (version, by), and apply_proposal (id, by: a change proposal becomes the next version, D49), and resolve (ids or packet_id, action, by, by_kind, reason: retry, dead-letter or drop packets waiting for the agent, D50). Every response is redacted. A runner refuses to start when a live runner answers on its socket, and removes a socket nothing answers on. Details in D24.

Discovery and reattach. The engine and the CLI both use the registry:

• When the engine starts, it reads every registry entry. For each one, it checks that the pid is alive and performs a socket handshake confirming the pipeline name and version.
  • If the runner is alive, the engine reattaches: it replays the events it missed from the runner's journal and takes back TTLs, stall detection and control. The pipeline never notices the outage.
  • If the runner is dead, the engine removes the stale entry. If the pipeline was active, the engine restarts it, and the journal resumes the packets that were in flight.
  • If the pid is alive but the handshake fails, the engine leaves the runner alone (never kills it or starts a second one) and shows it as unreachable until pipo attach succeeds. Adopted runners, restarts and the event cursor are in D27.
• pipo runners lists every runner the registry and socket checks find: pid, port, version and whether it is attached, detached or stale.
• pipo attach [name] tells the engine to scan again and reattach now, for example after a failed handshake. Without a name, it reattaches every runner it can.
• If the engine is down or can't start, or when --no-engine is passed, the CLI talks to detached runners directly through their sockets. status, logs, inspect, pause, resume and stop all work, and their output is marked engine: down. Any command that does start the engine reattaches every runner as part of starting up.

7.3 Delivery guarantees#

• At-least-once. A packet is journaled before it is acknowledged, and each node's result is committed before the next node runs. After a crash, processing resumes from the last committed node, so a tap may run twice.
• Idempotency. Outputs use meta.packet_id as their idempotency key (§3.5; packet_id:<branch> for a fan-out copy, D22), so a repeated write does not create a duplicate. The same applies to a batch written again after a crash.
• Version pinning. A packet finishes on the pipeline version that accepted it, with that version's own fn code and schema files as compiled when the version was stored (versions.compiled, D73), never re-read from disk. A new version applies only to newly accepted packets.
• Chains. A hand-off to another pipeline (to: pipeline, §3.14) is deduplicated by the receiver on the sender's name and key, in the transaction that journals the packet, so each sending unit becomes exactly one receiving packet, whatever crashes or retries happen on either side.
• Ordering. concurrency: 1 gives FIFO order. Above that, packets may complete out of order.

7.4 Time and schedules#

Each runner owns the clocks of its pipeline: schedules, lifetime (ttl, max_packets, until), stall detection and retention clean-up, attached or detached (D30). The engine never arms them itself, so a reattach can't fire anything twice. It learns what happened from the journal events it replays (pipeline.lifetime, pipeline.stall, pipeline.paused, pipeline.stopped). The engine keeps only its own clocks: the engine TTL and idle sleep (§7.5).

A pipeline's ttl is anchored to the first pipeline.started since the last clean pipeline.stopped, pipeline.completed or halt (pipeline.failed). A deliberate stop or pipo restart resets it; a crash or a new version does not. A start's --ttl (D57) counts from the same anchor. After a crash restart the runner arms only the time that is left, and ends at once if it is already over, so a crashing pipeline can't run past its TTL.

schedule inputs (§3.3) use a 5-field cron expression (*, lists, ranges, steps; when both day fields are restricted, either may match) or an every duration. Cron is evaluated in UTC. every counts from when the runner starts. Each tick becomes one packet with meta.trigger: schedule and meta.source set to the tick time (ISO 8601). The packet is journaled before the next tick is armed. Ticks missed while the runner was down or suspended are skipped, not replayed. poll and crawl (§3.3.2) time their fetches and crawls the same way, except that every counts from the last fetch's start (the last crawl's end), which they keep in the journal, so a restart doesn't fetch early.

7.5 Engine lifetime#

• The engine runs while any pipeline is active, paused, draining or jammed, or has pending packets.
• When every pipeline is stopped or completed and nothing is pending, the engine sleeps: it exits, and the next pipo command starts it again. It waits engine.idle (default 5m) with nothing to do before it sleeps (D31). Detached runners it supervises count as busy, so pending packets always have a runner.
• An open event stream keeps it awake. While any /events stream is connected (an open dashboard, say), the engine does not sleep; the idle clock starts when the last one closes. The TTL still applies (D61).
• Engine TTL (pipo engine start --ttl 8h, pipod --ttl 8h, or engine.ttl in config): when it expires, every pipeline drains and the engine stops. Detached runners keep running and end on their own lifetime, which the runner owns (D30, D31).

7.6 Observability#

Every packet state change and pipeline state change is an event. Events are written to the journal, streamed over /events (SSE, event ids <pipeline>:<seq>, D28) to the UI and CLI. Built-in metrics cover throughput, pending count, DLQ size, latency per node and the age of the oldest pending packet. They appear in the runner's stats (the control status op), in /api (GET /api/pipelines/<name>, under runner.stats) and in pipo status. Export as OpenTelemetry (engine.otlp) is a Phase 2 item (D41). A telemetry node like the one in §4 is only needed for custom business telemetry.

The latency and oldest-pending fields are read from the journal, so they survive restarts and work for detached runners (D54):

• stats.latency: one entry per node id of the version in force, plus output (the output write), each {count, p50_ms, p95_ms, max_ms}. They cover the node's latest stats.latency_window (100) completed steps, whatever their age. count is the number of steps in that window. Percentiles use the nearest rank. With no completed step, count is 0 and the rest are null. A step's latency is the time from when the runner starts it to the commit of the transition that completes it, retries and their waits included. A completed step is a node's result (moved on, fanned out, filtered or dead-lettered) or the output's write. For a batch, it is the write of the packet's group. An escalated or held step has no latency until it runs again and completes. A step killed by a crash has none either, and its rerun after the restart counts once. Delivery checks are not included.
• stats.oldest_pending_age_ms and stats.oldest_pending_received_at (ISO): the age and received_at of the oldest packet that is not terminal. Pending here means what stats.pending counts: in flight, held by a pause, in a batch, waiting on its fan-out copies or escalated to an agent. Both are null when nothing is pending. A replayed DLQ packet keeps its received_at.
• pipo status <name> prints them under the table (oldest pending: 3m12s (received …), and a NODE COUNT P50 P95 MAX table). --json has the raw fields. The list without a name keeps its columns.
• Agent spend (§3.11, D58): the runner's stats.agent_spend_today (USD in the pipeline's own budget day) shows under pipo status <name> as agent spend today: $…. Under the table, pipo status prints the home's total for the engine day against engine.agent_budget.per_day and the spend per pipeline, shown when a cap is set or anything was spent. --json has it as agent_budget (per_day or null, spent_usd, window_start, resets_at, timezone, pipelines), the same object as GET /api/agent-budget.

8. UI#

Phase 1: observe and debug The engine serves it at /ui (pipo ui starts the engine when none runs, then prints the URL and opens it, D61); it is static files with no build step (D40). It has the pipeline list (with the lifetime remaining, counting down live, and packet counts per state, the oldest pending age and throughput_per_min (packets accepted in the last 60 s, as in_per_min in D21) from GET /api/pipelines, stats, null when the journal can't be read), the live graph, the packet inspector, a live log tail, the agent feed (proposals with status, author and reason, and what was handed to the agent or resolved), the DLQ and the versions with diffs (D59).

• A list of pipelines showing state, version, throughput, pending, delivered and DLQ counts, and lifetime remaining.
• A graph view drawn from the .pipo file, with live counters and error badges on each node.
• A packet inspector that traces one packet through every node, showing its data, timings, attempts and errors. The first step shows the data the packet came in with; each later step shows the top-level keys it set or changed, and the keys it dropped as removed: − key (never as null, which is a value a step can really write). On a settled packet, each node step has 🔁 Rerun from here: it shows the rerun plan (the steps that will run again, flagged when they spend or reach outside) and runs it after a confirm (D79).
• Live logs, the DLQ (with replay), version history with diffs, and a feed of agent activity and proposals.
• Pause, resume, drain, stop, restart and start controls, on each pipeline's page and in the task manager.
• A task manager (#/top, 🧮 Tasks): every pipeline with its state, runner pid, CPU, memory, process count, throughput and pending count, refreshed every 2 s and sorted by CPU, with its controls on each row (D68).
• A bots page (#/bots, 🤖 Bots): add, edit, rename, test, make default and remove the Telegram bots of the home (§3.13, D69). A token is typed into a password field and never shown again.
• An agents page (#/agents, 🔑 Agents, §9.2, §9.4, D81, D82): pipelines agents wrote that wait for approval (the agent, its reason, the effects the pipeline has outside its folder from GET /api/approvals's effects (runs_programs, calls_out, spends, listens, writes_files, reads_files, each with the places), and every file it wrote), with Approve, Approve & start and Reject; agent proposals held on agent-written pipelines, with Apply and Reject; the OAuth grants (scope, pipelines, last use, Disconnect) and the static tokens; and the Auto-approve agent pipelines toggle (POST /api/settings/agents). The OAuth consent page (#/connect/<id>) needs the owner's login and shows the client's self-declared name as unverified and where the token goes (a program on this machine, or a website), and takes the scope, the pipelines and the grant's name; Approve or Deny send the browser on to the client.
• The owner's pages (D84, D86): #/setup[/<code>] creates the owner with the setup code from the engine log (pipo ui opens it with the code filled in), #/login, #/forgot (a reset link by email, shown only when the engine can send mail; otherwise the page names pipo owner reset) and #/reset/<token>. A 401 from /api sends the browser to the login (or setup) page and back afterwards. ⚙️ Settings has an Owner section: the email with Send test email, a password change, Log out and Log out everywhere; in open mode it says there is no login.
• A settings page (#/settings, ⚙️ Settings): the workspace, the folder the builder saves new pipelines in (D72). Pick one by walking folders from the current one, or type a path; a missing folder can be created. The choice applies at once and is saved as engine.workspace in config.yaml.

The dashboard has a friendly, playful look (D63). It has light and dark themes, follows the system theme until one is picked, and respects prefers-reduced-motion. Each pipeline's overview has a 🧪 Send a test packet card that pushes a packet into the running pipeline (POST …/push); on the live graph, packets travel along the edges as they flow. A pipeline's page always has a ✏️ Edit button that opens its file in the builder, even before the pipeline has a journal (it never started, or failed to start); a tab that can't load shows its error in place of its content. When the engine can't be reached, the dashboard shows a "napping" screen that names the command to wake it and reconnects by itself (D61). When the engine is older than Pipo's code on disk (it was updated since the engine started), the dashboard shows a banner asking for a restart, with a 🔄 Restart now button (D65), and an API route the engine doesn't know yet is reported as that, not as a bare 404 (D64). The task manager and the settings page have a ⏻ Stop engine button (POST /api/engine/stop, answered 202 with the pipelines that drain and the detached runners that keep running): after a confirm, the engine stops as on pipo engine stop (pipelines drain, detached runners keep running, crashed pipelines are remembered), and the dashboard shows the offline screen with the command to start it again.

Phase 2: build (the builder, #/build)

• A visual node editor that saves back to the .pipo file. The file remains the source of truth, and edits run through pipo check live. Blocks are dragged (or clicked) from a palette of inputs, steps and outputs onto a canvas. A new block slots in after the selected block, or before the output, and takes over what was fed there. Blocks are wired by dragging from an out-port (one per route branch) onto another block. A block or wire is removed with Delete, and removing a block wires its parents straight to what it fed. Undo and redo cover every edit. Block positions are kept in the browser, not in the file, and so are unsaved edits, per file: reopening a file brings them back (its saved version one undo away) as long as the file hasn't changed since. Otherwise the saved version opens and the old edits are dropped with a note. Work started in one file (a check, a dry run, a save) finishes for that file even if another is opened meanwhile. Agents sit in one 🤖 Agents row of the palette that folds open in place (Claude API, Claude Code, Codex, pi, opencode). One that can't run on this machine (not installed, not logged in, no API key; GET /api/builder/agents) is greyed out: clicking it says why and what to do instead of adding it, and an agent block using it shows the same in its inspector, with a 🔄 check again. Changing an agent block's provider keeps the settings the new one also takes (prompt, schema, timeout) and drops the others. Its model field suggests the provider's models but takes any text; a new Claude Code block starts on sonnet, a Codex block on Codex's first listed model, and pi and opencode on their own default (D67). Diagnostics show as badges on their block and in a problems list. Some come with a one-click fix: a schema file the pipeline names but that doesn't exist (P014 on an agent's with.schema or the input's schema) is created with a starter schema, and an agent node without a cost cap (P053) gets agent_budget: {per_day: 1}. Every block's inspector shows 📦 data: an example of data as it reaches the block (as it leaves, for the input), and clicking one of its paths puts it in the field last used, as ${data.x} in a template or data.x in an expression (D70). An agent node's inspector edits its schema file in place (the "answer shape"), and the dry run's stub for it is generated from that schema. Until a draft is saved, its file is the one Save offers, <workspace>/<name>/<name>.pipo, and checks, dry runs and schema files resolve relative paths from there.
• 🏙️ City view (D75). A 🔀 Graph | 🏙️ City switch on the canvas shows the same draft as a small isometric town; both edit the same value and the choice is remembered per browser. Each block is a building whose shape says what it does (the input a station with a radio mast, a map a workshop, transform: http an office tower with a dish, exec a factory with a chimney, a filter a checkpoint with a funnel and a gate, a route a roundabout with a signpost board per branch, a tap a kiosk, an agent a domed lab, the output a depot whose look follows to:), with the connector's emoji on a sign and a label above it. Each wire is a coloured data lane along the roads, with arrows in the direction packets go; a route's branches each have their own colour. Grey roads, houses, parks and cars are scenery and never carry data: ambient traffic only uses roads no lane is on. The pipeline reads left to right, a lower row is further down and to the side, and lots in front of a building only hold low scenery, so lanes and buildings stay in view. Everything the graph does works here: click or drop palette items (on a building to feed from it), drag a building to another lot (onto another building to swap), pull a lane from a ● port onto a building, select a lane and press Delete, the inspector, undo, tidy up. Problems show as a red ring, a 🚧 barrier and a badge on the label; a dry run sends a packet van along the lanes it took and marks each building ✅, 💥 or 🫥; for a running pipeline, live events send a van down each lane a packet takes and show retries (🔁), dead letters (💥), filtered packets (🧹) and deliveries (✅) at their building. Lots are kept in the browser per file, like block positions. A pipeline's page has the same 🔀 Graph | 🏙️ City switch on its overview (remembered per browser): the town of the version it runs (GET …/graph includes the parsed definition), read-only (no ports to lay lanes from; buildings can be moved, sharing the builder's lots for that file), with live vans and marks from the event stream, and a click on a building shows its counters as on the graph.
• Pickers for connectors and node kinds, driven by connector schemas, plus a test run against fixtures. Each connector's with: form is generated from its JSON Schema in the manifests. The test panel dry-runs the draft against a sample packet or the pipeline's fixtures (it starts on the file's first fixture, and a picked fixture runs with its own meta and stubs unless its data is edited), lights up the path on the canvas and shows the data leaving every step (D62).
• Save writes the file (a new one goes in the engine's workspace, D61). ▶️ Save & run saves and starts it. For a running pipeline, 🚀 Apply live sends the draft as a human proposal (by: ui, apply: true, §9.3) and, once it is applied, saves the file too, so the next start runs the same version. A refused proposal leaves the file as it was.

The builder talks to the engine through /api/builder/* (D62). All of it is local /api (loopback only, JSON POSTs), and none of it starts, stops or changes a running pipeline; the builder uses the existing routes for that (POST /api/pipelines, proposals).

• GET /api/builder/catalog: {workspace, inputs, taps, transforms, agents, outputs, checks, node_kinds, then}. Each connector is {description, with} (the JSON Schema of its with: block from the connector manifests); outputs add checks (the delivery checks they support besides ack, external and none) and batch; agents add label, runs (api or cli) and their default timeout.
• GET /api/builder/agents[?refresh=1]: {agents, checked_at}, which agents can run on the engine's machine (D67). Each is {label, runs, ready, version?, reason?, hint?, models}: a CLI agent is ready when it's installed and logged in (the check a runner makes at start), claude_api when its env: key is set (an op:// one is assumed fine). models are the ones to offer for with.model. The answer is cached for 5 minutes; refresh=1 checks again.
• GET /api/builder/files: {workspace, files: [{file, rel, name}]}, the .pipo files under the workspace (D61), four folders deep at most, skipping dot folders, node_modules, fixtures, __snapshots__ and target, sorted by rel, at most 500. name is null when the file doesn't parse. The walk is cached: a listing under 10 s old is answered as is, an older one is answered and walked again in the background (so the list page never waits on a big workspace), and saving a new file drops it. ?refresh=1 waits for a fresh walk.
• GET /api/builder/open?file=<abs> or ?pipeline=<name>: {file, source, pipeline, diagnostics, fixtures, fixtures_error?, stubs, running}. pipeline is the parsed value (null on a YAML error), diagnostics is pipo check's. fixtures ([{name, data, meta?, stubs?}]) and stubs come from the file's fixtures/ folder as pipo test reads them (§10.3); a bad fixture leaves the list empty and says why in fixtures_error. running is {name, state, version, version_source} when the engine knows a pipeline with that file (version_source is the definition of the version in force, from its journal), else null.
• POST /api/builder/check {source | pipeline, base?, file?}: a live pipo check of a draft. Send exactly one of source (.pipo text) and pipeline (a parsed value, written to text with base, see D62). Always 200 {source, pipeline, diagnostics, ok}; ok is false when any diagnostic is an error. file (default <workspace>/<name>.pipo) is where relative paths (fn, schemas) are looked up.
• POST /api/builder/save {file, source, overwrite?}: writes the file (atomically) and returns {file, created, diagnostics}. A file with errors is saved too: it is a draft, and pipo start refuses it with the same diagnostics.
• GET /api/builder/schema?file=&path= and POST /api/builder/schema {file, path, schema, overwrite?}: read or write a schema file named by the pipeline at file (which need not exist yet). path is relative, ends in .json and must stay in or below the pipeline's folder (real paths), and file follows the open/save rule (D62). The read returns {file, path, exists, schema, problem}; the write refuses a value that isn't a usable JSON Schema (400, as P054 judges it), an existing file without overwrite (409), and creates missing folders only inside the workspace.
• POST /api/builder/test {source, file?, fixtures, stubs?}: runs the draft against 1–50 fixture packets as pipo test does (§10.3: taps and the output mocked, agent nodes and transform: http stubbed, nothing sent or written), with the real clock and a trace, and returns the test report: {pipeline, fixtures: [{fixture, outcome, units, taps, calls, error?}], counts}. Each unit has steps: [{step, data}]. A draft with check errors or a feature the runner refuses is 422 invalid_pipeline with diagnostics and gaps; a bad fixture is 400.

9. Agents in Pipo#

9.1 Four roles#

1. Author. Agents write .pipo files using the published JSON Schema, the generators and pipo check/pipo test, or over MCP with create_pipeline, test_pipeline and start_pipeline; what they write runs once a human approves it (§9.4). They do not change engine code. That happens only in Pipo's own development.
2. Operator. Agents use the agent endpoint (§9.2) to watch pipelines, pause, resume and replay them, push packets and handle stalled packets or packets escalated with then: agent.
3. Editor. Agents change a running pipeline through the change protocol (§9.3), within the pipeline's agent: policy.
4. Node. agent: nodes run inside the graph with a required output schema, a timeout and a budget (§3.4).

9.2 Agent endpoint#

The engine serves an MCP server at /mcp (Streamable HTTP, revision 2025-06-18 or later: one JSON-RPC message per POST, answered as JSON; no sessions, no server stream). It also offers the same operations over REST and through pipo … --json; its tools call the REST handlers. Access uses scoped bearer tokens, from either of two places:

• OAuth (D82), the usual way: claude mcp add --transport http pipo http://127.0.0.1:4747/mcp, and the client's first request gets a 401 whose WWW-Authenticate: Bearer resource_metadata="…/.well-known/oauth-protected-resource" leads it through the MCP authorization flow. The browser opens the dashboard's consent page (§8), where the human picks the scope, the pipelines and the name the agent's actions are recorded under; the client gets a token.
• Static tokens listed in engine config, each a reference (op:// or env:) resolved when the engine starts:

engine:
  mcp:
    tokens:
      - name: ops-agent                  # author of its proposals, `by` of what it does
        token: op://Pipo/mcp/ops-agent   # or env:PIPO_MCP_TOKEN; never the token itself
        scope: operate                   # read (default) | operate | edit | author
        pipelines: [people-intake]       # optional allowlist (default: all)

• Scopes. read: list_pipelines, get_status, get_events, inspect_packet, list_dlq, check_pipo, get_proposal and every resource. operate adds replay, rerun, push, ack, pause, resume and resolve. edit adds propose_change and rollback. author adds create_pipeline, start_pipeline and stop_pipeline (§9.4); test_pipeline and list_approvals are read. Requests send Authorization: Bearer <token>: a missing or unknown token is 401 with the OAuth challenge and a hint. A pipeline outside a token's allowlist does not exist for it.
• The pipeline's policy (its version in force) applies on top: every pipeline tool and resource needs agent.control: true; pause, resume, replay, rerun, push and ack also need that action in agent.actions, as do start_pipeline and stop_pipeline (start, stop) on a pipeline the agent didn't write; proposals are checked against agent.edit (§9.3). Proposals and rollbacks made over MCP are recorded author_kind: agent with the token's name. An agent's rollback is a proposal of the older version's source, so the same checks apply.
• Tools: list_pipelines, get_status, get_events, inspect_packet, list_dlq, replay, rerun, push, ack, pause, resume, check_pipo, propose_change, get_proposal, rollback, resolve for escalated packets (D50), and the authoring tools create_pipeline, test_pipeline, start_pipeline, stop_pipeline and list_approvals (§9.4).
• Resources: pipeline sources and versions (pipo://pipelines/<name>/source, …/versions, …/versions/<n>), connector schemas (pipo://connectors/<kind>/<name>), the .pipo JSON Schema (pipo://schema/pipo.schema.json), this spec (pipo://docs/spec.md), every page of the guide (pipo://docs/guide/<page>.md, from docs/guide/) and a Markdown summary of every connector's with: fields (pipo://docs/reference/connectors.md).
• OAuth server (D82). The consent page needs the owner's login (D84), and the grant records who approved it (approved_by). With engine.public_url set, a request addressed to its host gets it as the issuer and resource origin (D85). Metadata at /.well-known/oauth-protected-resource (RFC 9728; also under /mcp) and /.well-known/oauth-authorization-server (RFC 8414); dynamic client registration at POST /oauth/register (RFC 7591, public clients, redirect URIs on a loopback host or https); GET /oauth/authorize (authorization code with PKCE S256, resource must be this engine's /mcp); POST /oauth/token (authorization_code, refresh_token). Access tokens live 1 hour, refresh tokens 30 days and rotate on use. Grants are in <home>/mcp-grants.json (0600, digests only), listed by GET /api/mcp/grants and pipo mcp grants, revoked by POST /api/mcp/grants/<id>/revoke and pipo mcp grants revoke <id>.
• Every response is redacted: secrets always, plus any extra paths listed in agent.redact (paths in packet data such as data.email; their values are masked wherever else they appear in the response too). Details in D53.

9.3 Change protocol: propose → validate → apply#

agent:
  control: true                        # enables the endpoint for this pipeline
  actions: [pause, resume, replay, push]
  edit:                                # paths an agent may change; everything else is forbidden
    - nodes.*.with.prompt
    - nodes.normalize
    - errors
  redact: [data.email]
  on_stall: notify                     # notify | handle (agent takes over stalled packets)

1. Propose. The agent submits a patch against version N, with a short reason.
2. Validate. The patch must pass pipo check, and it must touch only paths listed under agent.edit. Changes to output, delivered, secrets, agent and agent_budget are always forbidden for agents (D47).
3. Dry-run (optional). If agent.verify: last 20 is set, the new version replays the last 20 delivered packets in memory, with outputs and taps mocked and agent nodes replayed from what they returned before. The proposal is rejected if results diverge beyond the schema: a replayed packet fails a step, or no longer passes the input's or the output's validate rules or an agent node's schema (D48).
4. Apply at a safe point. A proposal that passed validation (and the dry run, when set) is applied right away; agent.edit is the grant, there is no separate human approval (D51). It can be held instead (apply: false, pipo proposals … propose --hold): it stays validated or verified until anyone with control applies it (apply_proposal) or rejects it (reject_proposal, only from those two states). The control ops are propose, proposals, proposal, apply_proposal and reject_proposal; REST has GET/POST /api/pipelines/<name>/proposals, GET …/proposals/<id>, POST …/proposals/<id>/apply and …/reject; reads work from the journal with no runner. Version N+1 is activated for newly accepted packets. In-flight packets finish on version N (§7.3). The version and the proposal's applied state are committed together, so a crash leaves one or the other, never half (D49).
5. Audit. Each version records who made it (human or agent id, and which of the two), the reason, the proposal it came from, the diff and the content hashes of the fn module and schema files it runs with. pipo rollback restores any earlier version's definition; its files are read as they are now, and each one that changed since that version was recorded is a warning (D60).

Every packet records which version processed it, so any result can be traced back to the exact definition that produced it.

Escalated packets. A packet handed to the agent (then: agent, or a stall with agent.on_stall: handle) waits in state escalated until it is resolved with retry (from the step it stopped at, on its pinned version), dead_letter or drop (not at the output). The runner control op resolve does it today; REST, the CLI and MCP call it (D50). Without an agent the packets simply wait; nothing is dropped.

9.4 Agent-authored pipelines and approval#

An agent with the author scope can build and run a pipeline end to end over MCP (D81):

1. check_pipo until the source has no errors, then create_pipeline {pipeline, source, files?, reason?}: the source and its files (an fn module, schemas, fixtures/*.json; an object of relative path → text) are written to a staging folder in the workspace and checked there; with errors nothing is written (invalid_pipeline with the diagnostics). Otherwise they land in <workspace>/<name>/, the source as <name>.pipo. The name must be the source's name:. Paths are relative and stay in the folder (no .., absolute paths, dot files or other .pipo files); at most 32 files, 256 KiB each, 1 MiB together. It never writes over a folder an agent didn't write, a pipeline another agent wrote, or one that is live (that one changes through propose_change); files it wrote before and leaves out are removed.
2. test_pipeline {pipeline | source, fixtures?, stubs?} dry-runs it as POST /api/builder/test does (taps and the output mocked, exec, http and agent steps stubbed, fn steps in the sandbox), with the folder's fixtures/ when fixtures is left out. It touches nothing, so it is read and needs no approval, like check_pipo.
3. Approval. The pipeline is pending until a human approves it: the dashboard's Agents page (§8), pipo approvals approve <name> [--start], or POST /api/approvals/<name>/approve {start?, by?}. Starting it as a human (pipo start, the dashboard, POST /api/pipelines) approves it too. reject (with a reason the agent sees) refuses it until the agent writes it again. Approval is tied to a SHA-256 of the files the agent wrote (path and content of each), so a rewrite, or anyone's edit to those files, makes it pending again.
4. start_pipeline {pipeline, ttl?} starts it once approved; until then it answers approval_pending or approval_rejected with a hint. stop_pipeline {pipeline, now?} drains (or stops) it. On a pipeline the agent didn't write, both need agent.control and start/stop in agent.actions. list_approvals lists the agent's own pipelines and their state.

Approvals live in <home>/approvals.json (mode 0600): per pipeline {name, file, files, by, reason, hash, state, requested_at, decided_at, decided_by, note}. REST: GET /api/approvals (each with effective, changed, pipeline_state and contents, the text of each file up to 64 KiB), GET /api/approvals/<name>, POST …/approve and …/reject.

Changes. An agent's proposal (or rollback) to a pipeline any agent wrote is held: it is sent with apply: false, so it stays validated or verified until a human applies it, and the answer carries held_for_approval: true. Otherwise an approved, harmless pipeline could be turned into another one through agent.edit. Proposals to pipelines a human wrote keep §9.3's rule: agent.edit is the grant.

Guidance. A Claude Code plugin (plugins/pipo, listed by the marketplace in .claude-plugin/marketplace.json) bundles a Pipo skill (the authoring loop and the file's rules) and the engine as an HTTP MCP server at 127.0.0.1:4747/mcp (D83); any other agent reads the guide as MCP resources (§9.2).

Auto-approve. engine.mcp.auto_approve: true (default false; the Agents page has a toggle that saves it to config.yaml and applies at once) approves what agents write as it is created, by auto_approve, and lets their proposals apply as usual.

10. Toolchain#

10.1 Editor support (VS Code first)#

• The .pipo JSON Schema is published by pipo schema. Connector with: blocks use conditional schemas based on via, to, tap and so on.
• The VS Code extension:
  • registers .pipo as YAML and attaches the schema, which gives completion, hover help and structural checks;
  • adds a TextMate injection that highlights ${…} templates and expressions;
  • includes a small Pipo language server for what a schema can't express: completing node ids in from, going to the definition of fn.* in the TS module, showing pipo check diagnostics live, offering quick-fixes (e.g. "unsupported delivery check: choose from …") and showing a graph preview.

10.2 Generators#

• pipo new <name> --template <t> scaffolds a pipeline folder: the .pipo file, an fn stub, schemas, fixtures and a test.
• Built-in templates: webhook-to-sqlite, cron-to-file, watch-to-http, agent-classifier, blank.
• Your own templates live in ./.pipo/templates/ or ~/.pipo/templates/. A template is a folder plus template.yaml, which defines prompts and variables. Files use {{var}} placeholders.
• pipo generate node inserts a node into an existing file, wires its from and keeps the file formatted.

10.3 Testing#

pipo test runs a pipeline in the runner binary (pipo-runner test, D73) against fixture packets, with no journal and no live connectors. Outputs and taps are mocked, agent nodes can be stubbed from recorded responses, and results are compared with snapshots. This is the same machinery as the dry-run in §9.3.

• Fixtures. pipo test <file|dir> reads <pipeline folder>/fixtures/*.json (or --fixtures dir): a packet's data as JSON, or {"data": …, "meta": {…}, "stubs": {…}}. Run-wide stubs (node id → response) go in fixtures/stubs.json; a fixture's own stubs win per node. A folder argument must hold exactly one .pipo file.
• Snapshots. Each fixture's result is kept in fixtures/__snapshots__/<fixture>.snap.json. A missing snapshot is written and reported as new. A different result fails with the changed paths and a hint; --update-snapshots accepts it (updated). A fixture that cannot run (a missing stub, say) fails whatever its snapshot says.
• Output. One line per fixture (✓/✗, its outcome) and a summary. The exit code is 0 only when every fixture passes. --json prints {pipeline, fixtures: [{name, status: pass|fail|new|updated, outcome, diff?, error?}], passed, failed}. Check errors, a bad fixture or a missing fixtures folder are errors with a hint, and nothing runs.

The dashboard's test run uses the same machinery with a trace: each unit also lists its steps, the data leaving every step (D62). pipo test and its snapshots don't include them.

11. Security#

• Secrets are referenced through op:// (1Password, recommended) or env: and never stored in .pipo files, the journal, logs or agent responses (§3.7).
• The engine binds to localhost by default. The dashboard and the API need the owner's login (or the engine key, <home>/engine.key, which the CLI sends), unless engine.auth: open, which only a loopback engine may use (D84). An engine on every interface needs engine.public_url and answers only requests addressed to it or to loopback (D85). HTTP inputs support token-header or HMAC signature auth, and one without either takes requests from the network only with public: true.
• The owner's password is hashed with argon2id; sessions, reset links and OAuth tokens are random and stored as SHA-256 digests only. Failed logins, setup codes and reset links are rate-limited. A forgotten password is reset with pipo owner reset on the engine's machine, or with a link by email when engine.mail is set (D86).
• Agent access uses scoped tokens, is limited by each pipeline's agent: policy and is fully audited. Tokens come from config.yaml references or from OAuth grants a human approved on the local dashboard (only SHA-256 digests are stored, D82). A pipeline an agent writes runs only once a human approved its files, and agent changes to it wait for a human too, unless engine.mcp.auto_approve is on (§9.4).
• exec nodes run programs as the user, with no shell and no sandbox (D71). A .pipo file with exec nodes that came from a template outside your project is untrusted, like an fn module, until you run pipo trust, and again after you edit it.
• fn modules and connectors are trusted code. fn modules run in the runner's embedded QuickJS without Bun or Node APIs (§3.6), with a time limit per call and a memory limit; that keeps them off the system, but it is not a reviewed sandbox. Each pipeline is isolated at the process level. Templates from outside your project are untrusted until you run pipo trust, which records a hash of the content. If the content changes, it has to be trusted again. Sandboxing arrives with the Phase 2 template registry.

12. Roadmap#

Phase 1: MVP

• The spec v1 format, pipo check, pipo fmt and pipo test.
• The engine, with one runner per pipeline, the journal, detached mode, the scheduler and the gateway.
• Inputs: http, schedule, watch, push, system. Outputs: sqlite, file, http, stdout. Node kinds: tap, transform, filter, route, agent, and loops.
• Delivery checks and stall detection, plus the DLQ and replay.
• Output batching (sqlite, file), agent budgets and retention.
• The CLI, generators, the JSON Schema and the VS Code extension.
• The Phase 1 UI dashboard, the MCP agent endpoint and the change protocol.

Phase 2

• The visual editor in the UI (built: the builder, §8, D62).
• Active inputs that fetch on their own: poll, tail, command, crawl (built: §3.3.2, D87).
• Agents through local coding-agent CLIs: claude_code, codex, pi, opencode (built: §3.4, D67).
• More connectors: postgres, mqtt (sensors), s3, queues.
• Join and window nodes, and batch mode for http and pipeline outputs. (Multiple outputs are not planned: a pipeline has one output by design, and chains carry data further, D78.)
• A connector SDK and a template registry.
• Distributed: sub-engines on other machines or operating systems, with runners placed remotely and controlled from one engine.
• OpenTelemetry export (engine.otlp, D41).
• Pipelines declaring the env vars they need (§14.2, D44).
• A configurable watch content limit, and emitting files that already exist on first start (§14.3, D44).
• Sandboxing fn modules (§11).
• The editor language server.

13. Decision log#

14. Open questions#

1. Fan-out and idempotency. Resolved by D22: each branch copy is its own journal unit with the key packet_id:<branch>, and pipo check warns (P026) when an explicit key can't tell copies apart.
2. Runner-level env allowlist. Expressions see env only for variables in engine.env_allow. Until the engine exists, pipo run --env-allow A,B sets it. Should a pipeline also be able to declare the variables it needs? Deferred to Phase 2 (D44).
3. Watch and system input details. watch finds events by comparing folder snapshots (fs.watch only triggers a scan; a 1 s poll covers drives where fs events are lost), so bursts of writes collapse into one change. The snapshot is kept in the journal and updated in the same transaction as each packet, so changes made while the runner was down are emitted exactly once on the next start (D19). The first start, or a changed glob, takes the files already there as a baseline. data is {event, path, name, size, mtime, content?}; meta.source is <path>#<event>@<mtime ms>. read: content reads UTF-8 text up to 1 MiB; larger files are sent with content: null, truncated: true. events defaults to [create, change]. system data is {<metric>: {...} | null, sampled_at}; unavailable metrics (battery on servers, WSL) are null, and cpu.percent is a delta between samples. Should the size limit be configurable, and should a first start be able to emit the files already there? Deferred to Phase 2 (D44).
4. Matching the output key in delivery checks. With fan-out, the default output key is packet_id for unbranched packets and packet_id:<branch> for copies (D22), so a delivered.with lookup by meta.packet_id misses copies. Resolved by D44: yes, expressions get meta.key, holding the resolved output key (§3.2). Built in D55.
5. Who owns the clock for detached runners. Resolved by D42: D30 stands. Runners own these timers and the engine learns outcomes from replayed events (§7.4, §7.5).
6. Observability scope in Phase 1. Resolved by D41: per-node latency, oldest-pending age and MCP /mcp are built in Phase 1. OTLP export moves to Phase 2.
7. Can an agent change its own agent_budget? agent.edit: ["*"] covered it, so an agent could raise the cap that limits its own spending. Resolved by D47: no, agent_budget is human-only like output, delivered, secrets and agent.
8. Can an agent fix a waiting packet's data? resolve retries an escalated packet with the data it had (D50), so a packet that fails on its data (output.on_invalid, a schema) fails again, and the agent can only dead-letter it or push a corrected copy. Should retry take replacement data? It would let an agent change data the pipeline never produced, and agent.redact hides parts of what it would edit.