pi-workflow-engine

A pi extension that brings Claude-Code-style dynamic workflows to pi: a JavaScript orchestration sandbox that spawns dozens of isolated subagents, keeps intermediate results in script variables (out of the host context window), and returns only the final synthesized value.

What it registers

• workflow tool — the LLM calls it with { script, args?, budgetTokens? }; the engine runs the script in a node:vm sandbox.
• /workflow <file> [json-args] command — run a saved .js script.
• REPL-mode subagent tools — persistent, stateful subagent sessions the host drives across multiple turns (see REPL subagents below): subagent_create, subagent_block, subagent_interact, subagent_suspend, subagent_kill.

Primitives injected into the sandbox

Primitive	Purpose
agent(prompt, { schema?, label?, phase?, model? })	spawn one isolated subagent; returns validated JSON when schema is given, else final text. null on hard failure.
parallel(thunks[])	barrier: run all, await all, null on failure. filter(Boolean) downstream.
pipeline(items, ...stages)	no barrier: item B stage-1 while item A is stage-3. Stage sig: (prevStageResult, originalItem, index).
phase(title) / log(msg)	progress UI.
args / budget	caller input; { total, spent(), remaining() }. Hard cap via budgetTokens → agent() throws when exceeded.

Determinism: Date / Math.random throw (keeps resume replays consistent). Limits: min(16, cpus-2) concurrent agents, 1000 total per run.

Install

Auto-discovered at ~/.pi/agent/extensions/pi-workflow-engine/. /reload picks up edits. For other machines: pi install from git/npm, or copy the dir.

Script format

export const meta = {
  name: 'security-audit',
  description: 'Parallel security scan',
  phases: [{ title: 'Scan' }, { title: 'Synthesize' }],
}

const SCHEMA = {
  type: "object",
  required: ["refuted", "confidence"],
  properties: {
    refuted: { type: "boolean" },
    confidence: { enum: ["high", "medium", "low"] },
  },
}

phase('Scan')
const [auth, db] = await parallel([
  () => agent('Audit auth flows'),
  () => agent('Audit DB queries', { label: 'db', schema: SCHEMA }),
])

phase('Synthesize')
const report = await agent(`Compile findings:\n${auth}\n${JSON.stringify(db)}`)
return report

Token & cache behavior

• Token accounting: session.getSessionStats() → real input/output/cacheRead/cacheWrite. budget.remaining() is finite when budgetTokens is set, so while (budget.remaining() > N) loops terminate.
• Prefix-cache affinity: one shared ResourceLoader across all subagents → byte-stable system+tools prefix → concurrent subagents hit the same cached prefix (verified: 100% cache-read on a 4-agent pipeline run). 1h retention via PI_CACHE_RETENTION=long (set on pool creation if unset) so inter-phase gaps >5min don't evict the cache.

Structure

pi-workflow-engine/
├── index.ts          # extension entry: registers tool + command + REPL tools, wires UI
├── runtime.ts        # sandbox + primitives (agent/parallel/pipeline/phase/log/budget)
├── subagent.ts       # isolated subagent spawn, concurrency cap, schema validation, createSession
├── repl.ts           # REPL-mode persistent subagent registry (create/block/interact/suspend/kill)
├── test/selftest.ts  # no-LLM self-check (mock host): npx tsx test/selftest.ts
├── test/repl_selftest.ts # no-LLM REPL self-check (mock session): npx tsx test/repl_selftest.ts
├── test/tool_selftest.ts # no-LLM tool-layer self-check (mock pi+pool): npx tsx test/tool_selftest.ts
├── docs/AUDIT.md     # token-efficiency & cache-affinity audit log
├── package.json      # pi-package manifest
└── README.md

Run the self-test

cd ~/.pi/agent/extensions/pi-workflow-engine
npx tsx test/selftest.ts        # 31 checks, no LLM calls (runtime + schema + extractMeta)
npx tsx test/repl_selftest.ts   # 109 checks, no LLM calls (REPL lifecycle + fork)
npx tsx test/tool_selftest.ts   # 64 checks, no LLM calls (tool layer + /workflow command)

REPL-mode subagents

The workflow tool's agent() is one-shot: it spawns a fresh isolated subagent, runs it to completion, returns the final text, and tears the session down. For long-running, interactive, or multi-turn sub-tasks you need a persistent subagent — one that keeps its conversation history across turns. That's what the subagent_* tools provide.

Tool	Purpose
subagent_create({ mode?, forkFrom?, prompt?, model?, systemPrompt? })	Spawn a persistent REPL subagent. mode:"new" (default) starts a fresh empty session; mode:"fork" copies another agent's conversation history to branch it. If prompt is given, the turn is kicked off non-blocking — join it with subagent_block. Returns the agentId. See Forking below.
subagent_block({ agentId })	Block until the agent's current turn finishes (join). No-op if idle. Returns the accumulated text + token delta.
subagent_interact({ agentId, prompt })	Interact: send a prompt and block until the turn completes. Conversation history is retained across calls. Throws if a turn is already running.
subagent_suspend({ agentId })	Suspend: abort the current turn but keep the session alive. Returns partial text. The agent stays usable for future subagent_interact calls.
subagent_kill({ agentId })	Kill: permanently dispose the session and remove it from the registry. Idempotent. Always call when done to avoid leaking sessions.

Lifecycle: create (→ optional block) → interact×N → suspend (if needed) → kill.

One in-flight turn per agent. interact rejects if a turn is already running; block to join it or suspend to abort first. suspend and kill resolve the in-flight prompt() via session.abort() (matches the SDK's abort semantics — the partial assistant message is retained).

Token accounting: each turn returns a per-turn delta (stats-after minus stats-before), not cumulative totals — so you can budget individual turns.

Forking

subagent_create with mode:"fork" seeds a new subagent with an existing conversation, producing a divergent branch you can drive independently — the source keeps its own history untouched. forkFrom selects the source:

• forkFrom: "<agentId>" — fork from another live REPL subagent. The source must be idle (no turn in progress); suspend or block it first. Inherits the source's resolved model + system prompt unless overridden.
• forkFrom: "current" — fork from the current (host) session: the extension reads the host transcript via ctx.sessionManager + buildSessionContext() and seeds the subagent with it. This lets you hand off the in-progress work to an isolated context window (e.g. to let a subagent run a long tool-heavy investigation without polluting the host's context). Inherits ctx.model + ctx.getSystemPrompt().

The fork copies the transcript via session.state.messages = [...src] (a shallow array copy — message objects are shared, the list is independent). Both fork modes accept an optional prompt to kick off a first turn on the branch immediately.

Example (fork from current, then diverge):

subagent_create({ mode: "fork", forkFrom: "current", prompt: "Take over: finish the refactor and run the tests" })
  → agentId: repl-2, status: running   # started with the host transcript + a first turn
subagent_block({ agentId: "repl-2" })
  → text: "Done — tests pass.", status: idle
subagent_kill({ agentId: "repl-2" })

Example (interactive debugging session):

subagent_create({ prompt: "Reproduce the bug in src/auth.ts", systemPrompt: "..." })
  → agentId: repl-1, status: running
subagent_block({ agentId: "repl-1" })
  → text: "I found the bug: ...", status: idle
subagent_interact({ agentId: "repl-1", prompt: "Write a fix and run the tests" })
  → text: "Tests pass.", status: idle
subagent_kill({ agentId: "repl-1" })
  → status: killed

Footgun

pipeline stage signature is (prevStageResult, originalItem, index). For a first stage that only needs the item, write (_prev, item) => ... — (item) => silently binds the undefined prev to your param. Matches Claude Code's API.

Not implemented (ponytail deferrals)

• No cross-session resume (ephemeral journal; same-session resume is a v2 candidate).
• No git worktree isolation (isolation:'worktree' accepted, not implemented).
• Minimal schema validator (no $ref/oneOf/allOf).
• No interactive /workflows pane beyond the setWidget status lines.