pi-dynamic-workflows 完整教學

把 Anthropic 在 Claude Code 推出的 dynamic workflows (DWF; 動態工作流) 概念，移植到 earendil-works/pi 編程代理框架。一個 Pi extension，註冊 workflow 工具；主模型寫一段 JavaScript script，工具會在 vm sandbox 沙箱裡跑，期間 fan-out 多個 isolated subagent 並行做事，最後把結果合回。
適合：codebase audit、multi-perspective review、large refactor、fan-out research。不適合：單檔讀寫、簡單 grep、純對話、不需要 fan-out 的任務。

1. 專案定位

維度	說明
屬性	TypeScript library（Pi extension）
Size	約 1.6 K LOC（含 tests），核心 6 個 `src/*.ts`
核心 abstraction	`workflow` 工具 — 把「主模型生成 JS script」變成「沙箱化 + AST 驗證 + subagent 並行執行」的一等公民
對標來源	Anthropic Claude Code dynamic workflows（README L9 自承）
跑在哪	你必須先有 Pi（`@earendil-works/pi`）這個 coding agent framework；此 repo 只是 extension
何時用	「請你幫我用 workflow 跑 X」這種使用者明確要求多代理編排的情境 — README 反覆強調別濫用
何時不用	單一檔案的 read/edit、簡單 search、不需要 fan-out 的線性任務

1.1 「dynamic」這詞的具體意義

不是「workflow 動態被產生」這種空話，而是兩個具體技術點：

腳本由 LLM 即時撰寫：主模型在被要求時，把整個工作流寫成一段 JavaScript，傳進 workflow 工具的 script 參數。沒有預先定義的 DAG (DAG; 有向無環圖)。
Phase 在執行期被發現：meta.phases 只是可選的「outline 文件」；真正驅動 live progress UI 的是腳本執行中動態呼叫 phase("Name")。所以 if/else 分支或 loop 裡的 phase 可以條件出現，不會跑出空的 progress row。

1.2 三層架構：tool / runtime / agent

 1Pi session
 2  └─ workflow tool（src/workflow-tool.ts）
 3       ├─ parseWorkflowScript()  ← AST 驗 + meta 抽取
 4       └─ runWorkflow()（src/workflow.ts）
 5            ├─ vm.createContext(...)
 6            ├─ 注入 agent / parallel / pipeline / phase / log / args / cwd / budget
 7            ├─ new vm.Script(wrapped).runInContext(context)
 8            └─ 每個 agent() → WorkflowAgent.run()（src/agent.ts）
 9                              └─ createAgentSession(SessionManager.inMemory)
10                                 ├─ 注入 createCodingTools(cwd)
11                                 ├─ 若有 schema → createStructuredOutputTool(...)
12                                 └─ session.prompt(...)

2. 安裝指南

2.1 系統要求

Node.js（README 沒明示版本，但 package.json 用 tsx、type: module、"ecmaVersion": "latest"，建議 Node 20+）
Pi 已安裝且可執行（pi install、pi reload）
@earendil-works/pi-ai、@earendil-works/pi-coding-agent、@earendil-works/pi-tui ≥ 0.78.0（peerDependencies）

2.2 安裝步驟

 1# 方式 A：從 npm
 2pi install npm:pi-dynamic-workflows
 3
 4# 方式 B：從 local checkout（適合改 source）
 5git clone https://github.com/Michaelliv/pi-dynamic-workflows.git
 6cd pi-dynamic-workflows
 7npm install
 8npm test               # biome check + tsc + 單元測試
 9pi install /path/to/pi-dynamic-workflows
10
11# 方式 C：直接 dev
12npm run dev            # 等同 tsx src/index.ts

進入 Pi session 後：

1/reload

extension 會在 session_start 事件自動把 workflow 加進 activeTools。

2.3 安裝流程圖


flowchart TD
    A["npm:pi-dynamic-workflows
or local checkout"] --> B["pi install"]
    B --> C["Pi session start"]
    C --> D["extensions/workflow.ts"]
    D --> E["pi.registerTool(workflowTool)"]
    D --> F["pi.on('session_start')
setActiveTools(+workflow)"]
    E --> G["workflow tool ready"]
    F --> G
    G --> H["User: 'Run a workflow to ...'"]
    H --> I["LLM writes JS script
+ calls workflow tool"]

3. 核心架構解析

3.1 全貌


flowchart LR
    subgraph User["User prompt"]
        U["'Run a workflow to inspect repo'"]
    end
    subgraph Parent["Parent Pi session"]
        M["LLM writes JS"]
        T["workflow tool execute()"]
    end
    subgraph Tool["workflow runtime (vm sandbox)"]
        P["parseWorkflowScript
(AST 驗證)"]
        R["runWorkflow
vm.createContext"]
        C["agent / parallel / pipeline
phase / log / budget"]
    end
    subgraph Sub["WorkflowAgent (per agent call)"]
        S1["createAgentSession
(SessionManager.inMemory)"]
        S2["createCodingTools(cwd)"]
        S3["structured_output tool
(if schema)"]
        S4["session.prompt(task)"]
    end
    U --> M --> T --> P --> R --> C
    C -->|agent('...')| S1
    S1 --> S2
    S1 --> S3
    S1 --> S4
    S4 -->|final text or schema obj| C
    C -->|return value| T
    T -->|structured-cloneable result| M

3.2 6 個 source 檔的職責

檔案	LOC	職責
`src/workflow.ts`	455	AST validator + vm sandbox runtime；定義 `agent` / `parallel` / `pipeline` / `phase` / `log` / `budget` 語意
`src/workflow-tool.ts`	209	Pi tool definition（`workflow` 工具的 schema、prompt guidelines、live progress 串流、abort 處理）
`src/agent.ts`	131	`WorkflowAgent` — in-memory Pi subagent runner；裝載 coding tools 與 structured output tool
`src/display.ts`	235	Workflow snapshot 結構 + compact text renderer（給 live progress 用）
`src/structured-output.ts`	47	`structured_output` 工具 — schema 驗證 + `terminate: true` 讓 subagent 一次結束
`src/index.ts`	30	純 re-export，給 npm consumer 用
`extensions/workflow.ts`	14	Pi extension entry — `registerTool` + `setActiveTools` 兩件事

3.3 deterministic sandbox (DS; 決定性沙箱) — 三層保護

層	機制	程式碼位置
L1 AST 驗證	`parseWorkflowScript` 用 acorn 解析腳本後，呼叫 `assertDeterministicAst` walk 整棵樹，遇到 `Date.now()` / `Math.random()` / `new Date()` 直接 throw	`src/workflow.ts:312–344`
L2 meta 字面值限制	`evaluateLiteral` 只允許 `ObjectExpression` / `ArrayExpression` / `Literal` / `TemplateLiteral`（不含 interpolation）/ `UnaryExpression`（負號 only）；不允許 spread、computed key、method、function call	`src/workflow.ts:267–303`
L3 vm.createContext	沒注入 `require` / `import` / `fs` / `process.env` / network API；只暴露 `agent` / `parallel` / `pipeline` / `phase` / `log` / `args` / `cwd` / `budget` / `console` 與部分 builtin（`JSON / Math / Array / Object / String / Number / Boolean / Set / Map / Promise`）	`src/workflow.ts:186–212`

⚠️ 重點：作者在 README 明說「This keeps meta parseable, runs reproducible, and the surface area small」— 不是當作 security boundary。Node vm.createContext 過去多次出現 obj.constructor.constructor("...")() 這種逃逸；要做敵對輸入隔離，請額外用 process / worker 圍堵。詳見 §6 資安掃描。

3.4 核心 primitive 對照表

Primitive	語意	失敗行為
`agent(prompt, opts)`	起一個 isolated Pi subagent，return final text（或 `opts.schema` 給的物件）	例外時 log 並 return `null`；不會 abort 整個 workflow（除非 signal 被 abort）
`parallel(thunks)`	thunks = `Array<() => Promise<T>>`（不是 promises）；`Promise.all` 平行跑	個別 thunk fail → 該位置 `null`；其它繼續
`pipeline(items, ...stages)`	每個 item 走完所有 stages 才算完成；items 間平行；每個 stage 收 `(prev, original, index)`	某 item 某 stage fail → 該 item 結果 `null`
`phase(title)`	標記目前 phase；之後的 `agent()` 落在此 phase 群組底下；UI 顯示	不會 throw（只改 state）
`log(message)`	加一條 workflow-level log	不會 throw
`budget`	`{ total, spent(), remaining() }`；spent 用 `JSON.stringify(result).length / 4` 估 token	`agent()` 進入時若 `remaining()<=0` → throw
`args`	tool 呼叫時帶進來的 JSON value	由 LLM 在 tool call 傳入
`signal`	由 tool execute() 傳入；Esc 取消會 abort	各 primitive 進入時檢查 `signal.aborted` → throw

3.5 Concurrency / abort / budget 三件事

Concurrency limit：createLimiter() 在 runWorkflow 開頭算上限 = min(16, max(1, navigator.hardwareConcurrency - 2))。parallel() 與 pipeline() 都過此 limiter。
Abort：tool 收到 signal，runtime 內 throwIfAborted 在每個 primitive 入口檢查。subagent 在 WorkflowAgent.run 內注入 signal.addEventListener("abort", () => session.abort())。
Token budget：可選的 tokenBudget 設定；agent() 進入時若 remaining()<=0 就 throw workflow token budget exhausted。注意 estimateTokens() 是粗估（JSON.stringify(result).length / 4），不是真實 API 計費。

3.6 Structured output 流程


sequenceDiagram
    participant W as Workflow runtime
    participant A as WorkflowAgent
    participant S as Subagent (Pi session)
    participant T as structured_output tool
    W->>A: agent('...', { schema: JSONSchema })
    A->>S: session.prompt(task + 'must call structured_output')
    S->>S: thinking, reading files
    S->>T: tool call with params (validated by schema)
    T-->>S: { terminate: true }
    S-->>A: session ends
    A-->>W: capture.value (validated object)

關鍵：createStructuredOutputTool 回傳 terminate: true，讓 subagent 不需要再做一輪 assistant text turn，省 token + 強制結構化。

4. 6 個檔案的詳細用法

4.1 src/workflow.ts — runtime 核心

最重要的兩個 export：

1parseWorkflowScript(script: string): { meta: WorkflowMeta; body: string }
2runWorkflow<T>(script: string, options: WorkflowRunOptions): Promise<WorkflowRunResult<T>>

如果你想繞過 Pi 直接呼叫 runtime（例如寫 CLI 跑 workflow）：

 1import { runWorkflow, WorkflowAgent } from "pi-dynamic-workflows";
 2
 3const result = await runWorkflow(scriptText, {
 4  cwd: process.cwd(),
 5  args: { input: "..." },
 6  tokenBudget: 200_000,
 7  concurrency: 8,
 8  signal: AbortSignal.timeout(60_000),
 9  onPhase: (p) => console.log("Phase:", p),
10  onAgentStart: (e) => console.log("Start:", e.label),
11  onAgentEnd: (e) => console.log("End:", e.label, e.result === null ? "FAIL" : "ok"),
12});
13console.log(result.result);

4.2 src/workflow-tool.ts — Pi tool 包裝

裡頭有非常詳細的 promptGuidelines（行 56–71）— 幾乎是「LLM 寫 workflow 時必讀」的 RFC：

只在使用者明確要 workflow 才用
第一條 statement 必須是 export const meta = { name, description }
不能用 TS syntax、imports、require、fs、Date.now、Math.random、new Date
parallel() 收 functions 不收 promises（這條最容易踩雷）
每個 agent() 都要有 unique short label（2–5 字）
failed branch return null — 合成前要先檢查
多 subagent 合成時，最後加一個 synthesis agent 算 verdict
subagent 沒有 parent 的 repo context — 要在 prompt 帶足資訊

4.3 src/agent.ts — WorkflowAgent

可以獨立使用。在你自己的工具裡：

1import { WorkflowAgent } from "pi-dynamic-workflows";
2
3const wa = new WorkflowAgent({
4  cwd: "/path/to/repo",
5  instructions: "You are a senior reviewer.",
6});
7const text = await wa.run("Summarize the auth module.", { label: "auth-review" });

若要 schema：

1import { Type } from "typebox";
2
3const schema = Type.Object({
4  files: Type.Array(Type.String()),
5  risk: Type.Union([Type.Literal("low"), Type.Literal("medium"), Type.Literal("high")]),
6});
7const obj = await wa.run("Audit src/.", { schema });

4.4 src/structured-output.ts — 通用化的好工具

createStructuredOutputTool 並不綁 workflow 場景。任何 Pi 任務都可以塞給它：

1import { createStructuredOutputTool } from "pi-dynamic-workflows";
2
3const capture = { called: false, value: undefined };
4const tool = createStructuredOutputTool({ schema, capture });
5// 把 tool 加進你自己的 Pi session.customTools

terminate: true 是關鍵設計 — Pi 看到 terminate flag 後不會再要求 assistant 寫 prose final answer，直接以 tool 回傳作為終態，省一輪 round trip。

4.5 src/display.ts — snapshot model

定義了 WorkflowSnapshot 與兩個 display 工廠：createToolUpdateWorkflowDisplay、createWidgetWorkflowDisplay。前者用在 workflow 工具內部串流 progress；後者給 UI 框架嵌入用。

preview(value) 函式會把 subagent 結果壓成 ≤ ~80 字的縮圖，避免大物件淹沒 progress UI。

4.6 extensions/workflow.ts — 14 行 entry

整個 extension 就 14 行。createWorkflowTool() 出工具實例，session_start 時把它加進 active tools。Pi extension 寫得多薄，這就是範例。

5. 應用場景

5.1 經典「inspect + summarize」（README 範例）

 1export const meta = {
 2  name: 'inspect_project',
 3  description: 'Inspect a repository and summarize the main modules',
 4  phases: [
 5    { title: 'Scan' },
 6    { title: 'Analyze' },
 7  ],
 8}
 9
10phase('Scan')
11const inventory = await agent('Inspect the repository structure.', {
12  label: 'repo inventory',
13})
14
15phase('Analyze')
16const summary = await agent(
17  'Summarize the main modules from this inventory:\n' + inventory,
18  { label: 'module summary' },
19)
20
21return { inventory, summary }

5.2 多檔案並行 review（fan-out）

 1export const meta = {
 2  name: 'parallel_file_review',
 3  description: 'Review N files in parallel with diverse reviewer roles',
 4}
 5
 6phase('Review')
 7const files = args.files   // 透過 args 傳入清單
 8const reviews = await parallel(
 9  files.map((f) => () => agent(
10    `Review ${f} for bugs, performance, and security. Return findings.`,
11    { label: `review:${f}` },
12  )),
13)
14
15phase('Synthesize')
16const verdict = await agent(
17  `Combine these reviews and give a final verdict:\n${JSON.stringify(reviews)}`,
18  {
19    label: 'synthesis',
20    schema: {
21      type: 'object',
22      properties: {
23        ok: { type: 'boolean' },
24        critical: { type: 'array', items: { type: 'string' } },
25      },
26      required: ['ok', 'critical'],
27    },
28  },
29)
30return verdict

5.3 pipeline 模式（每篇 paper 過三段 stage）

 1export const meta = {
 2  name: 'paper_pipeline',
 3  description: 'Per paper: read → extract methods → score',
 4}
 5
 6const papers = args.papers   // [{ path, title }, ...]
 7
 8const results = await pipeline(
 9  papers,
10  (_, p) => agent(`Read ${p.path}, return key claims.`, { label: `read:${p.title}` }),
11  (claims, p) => agent(`Extract Methods section from:\n${claims}`, { label: `methods:${p.title}` }),
12    label: `score:${p.title}`,
13    schema: { type: 'object', properties: { score: { type: 'number' }, why: { type: 'string' } }, required: ['score', 'why'] },
14  }),
15)
16
17return results

5.4 適合的真實 use case

codebase audit：fan-out 每個 src/ 目錄，最後 synthesizer 合成 risk register
multi-perspective review：同一份 PR 派 3 個 subagent 用不同角度（security / performance / correctness）審
fan-out research：paper-tutorial 把 N 篇 paper 並行讀（README 範例 §5.3 已是現成模板）
migrate：每個檔案 pipeline（讀 → 改寫 → 驗證），同型工作平行做

6. 資安掃描報告

掃描範圍：src/、tests/、extensions/。掃描關鍵字：eval/exec/os.system/subprocess/shell/curl/wget/http/secret/token/api_key/pickle/__import__/input/fs/process.env。

6.1 紅黃綠燈

等級	項目	說明
🟢	無對外網路請求	整個 repo 沒有 `fetch` / `http` / `https` / `curl` / `wget` 用法
🟢	無 shell exec	沒有 `child_process` / `exec` / `spawn`；subagent 跑 shell 走 Pi 既有 coding tools，責任在 Pi 框架
🟢	無 secret 處理	沒有讀寫 `process.env`、`API_KEY`、`token`；扣除 `tokenBudget`（與「auth token」無關）不算
🟢	AST 黑名單	`Date.now / Math.random / new Date / spread / computed key / function call inside meta / require / import` 都在 parse 階段被擋
🟢	Structured clone guard	`runWorkflow` 結尾 `assertStructuredCloneable(result)` 防 Promise 漏 await（commit `1c61834`）
🟡	vm sandbox ≠ security boundary	作者明說「surface area small」但不保證敵對輸入安全；Node `vm.createContext` 有 `obj.constructor.constructor('return process')()` 之類逃逸記錄。信任場景 OK；給陌生人代入 script 要再隔離（worker / process / firewall）
🟡	Subagent 繼承 coding tools	`createCodingTools(cwd)` 注入後，subagent 可讀寫檔、執行 shell。權限 = 父 Pi session。若使用者下惡意 workflow，可在 sandbox 外做事
🟢	Abort 串聯	abort signal 從 tool → runtime → 每個 subagent session 都會傳，確保 Esc 真的中斷
🟢	License	MIT，使用無授權風險

6.2 結論

🟢 整體低風險（給你自己用）。 🟡 中等風險（給多租戶 / 陌生使用者用） — 需要追加：

把 vm runner 放進 worker / child process
subagent 不要繼承 full coding tools，改成 whitelist
限制 cwd，避免讀寫 repo 外路徑

6.3 對 AI-knowledge template 場景

本機跑、自己寫 workflow → 可放心安裝
若要把 workflow tool 暴露給外部 demo / 自動化 routine → 補一層 process 隔離

7. FAQ

Q1：和 Anthropic Claude Code 的 Workflow tool 差在哪？ Issue #11 列了「6 個 untracked faithfulness gaps」。最明顯：(1) 無 persisted / resumable（issue #1 epic）(2) 無 /workflows manager (3) 無 saved workflow registry (4) 沒 consent prompt。功能性 parity 約 70%，核心 primitive（agent / parallel / pipeline / phase / schema）齊全。

Q2：可以在 workflow 裡呼叫另一個 workflow 嗎？ 目前不行。沒有 workflow() nested primitive。要做 nested 必須在 subagent 裡讓它再呼叫 workflow 工具（但 subagent 預設沒有 workflow tool — 需要 customTools 注入）。

Q3：subagent 看得到主 session 的 context 嗎？ 看不到。SessionManager.inMemory(cwd) 是新的空白 session。所以 prompt guideline 第 N 條明說：不要假設 subagent 有 repo context，每個 agent prompt 要帶足資訊。

Q4：parallel(items.map(i => agent(...))) 為什麼錯？ 因為 agent() 是 async function，items.map(i => agent(...)) 馬上開始執行，傳給 parallel 的是已啟動的 promises，concurrency limiter 來不及生效。parallel 內會 throw expects an array of functions, not promises。正確寫法：items.map(i => () => agent(...)) — 傳「沒呼叫的 thunk」進去。

Q5：abort 後資料怎麼算？ abort 中的 agent 在 progress UI 顯示 skipped；workflow tool 整體 throw Workflow was aborted；budget 累計到中斷前那一刻為止。

Q6：tokenBudget 是真的 token 還是估計？ 估計。estimateTokens(result) = ceil(JSON.stringify(result).length / 4)。和 Anthropic billing 不同步；建議當 soft cap 用，不要當計費依據。

Q7：能不能不用 Pi、純當 npm package 跑 workflow？ runtime 層可以（runWorkflow + 自己包 WorkflowAgent），但 WorkflowAgent 內部用 @earendil-works/pi-coding-agent 的 createAgentSession — 等於還是要 Pi runtime。要徹底脫鉤須改寫 agent.ts。

Q8：phase 跟 meta.phases 不一致會怎樣？ 不會 throw。meta.phases 純 documentation；live UI 依 phase() runtime call 為準。可以 meta 宣告 5 個 phase 但實際只觸發 2 個（因 if branch 沒走到）。

8. 進階技巧

8.1 用 schema 強制結構

每個關鍵 synthesis 用 schema 強制收斂。LLM 很愛在 prose 加閒話，schema 直接掐斷。

8.2 budget guard 早收

1phase('Find')
2let found = []
3while (found.length < 10 && budget.remaining() > 30_000) {
4  const r = await agent('Find more bugs in src/', { label: `find-${found.length}` })
5  found.push(...JSON.parse(r))
6}

8.3 對照 phase 把 progress 切細

不需要在 meta.phases 預宣告所有 phase。loop 裡用 phase('Verify ' + i) 動態建立，UI 會 incrementally 多出 group。

8.4 用 agent({ phase: '...' }) 強制歸屬

某些情境主流程已經換 phase 了，但你想把這次 agent 歸到舊 phase（例如背景做清理），用 opts.phase 強制：

1phase('Synthesize')
2await agent('Final report', { label: 'final' })
3// 同時做背景驗證，歸屬 'Verify' 不污染 Synthesize phase
4await agent('Sanity check', { label: 'sanity', phase: 'Verify' })

8.5 IntelliSense

在你的 .workflow.js 檔頂端加：

1/// <reference types="pi-dynamic-workflows/workflow" />

VS Code 會給 agent / parallel / pipeline / phase / log / args / budget 的 IntelliSense（型別來自 types/workflow.d.ts）。

9. 整合進其他工作流

9.1 對映本專案 19 Layer

本專案 Layer	可加 workflow 編排的地方
Layer 9 paper-search	跨 6 資料庫平行查 → 目前用 bash 串；改 workflow 後可動態決定深度
Layer 12 gh-tutorial-qd	gh-save + tutorial md + 資安掃描 + qd compile ×2 → 6 step 平行
Layer 15 paper-tutorial	N 篇 paper 並行 tutorial（README 已有 case）
Layer 18 research-pipeline-v2	9-stage pipeline 本身就是天生的 workflow target
Layer 19 tu-plan-generator	12 domain × 4 candidate 同時跑（v3 case 已驗證）

9.2 把 workflow tool 接到 Pi 之外的編程代理

Claude Code：用 Workflow 工具（Anthropic 官方版）即可，這 repo 是 Pi 移植，不需要再接
Cursor / Cline / Aider：runtime 層 (runWorkflow) 可獨立用，需自己包 subagent runner 取代 WorkflowAgent
本專案內：目前不引入（避免和 superpowers chain 衝突），但作為「未來把 9-stage pipeline 框架化」的候選方案

9.3 跟 superpowers 的關係

superpowers 是 skill-based 編排（每個 skill 是 markdown + bash + python）。workflow 是 script-based 編排（每個 workflow 是一段 JS）。兩者互補不互斥：

高 level：superpowers skill 決定「做什麼」
低 level：skill 內部某個 stage 若需要 fan-out，可在 Claude Code 內呼叫 Workflow tool

10. 重點摘要 Checklist

Pi extension，註冊 workflow 工具
主模型寫 JS script → vm sandbox 跑 → fan-out subagent
AST 阻擋 Date.now / Math.random / new Date / spread / require / import
primitive：agent / parallel / pipeline / phase / log / budget / args
parallel(thunks) 收 functions、不收 promises（最常踩雷）
subagent 看不到 parent context — prompt 要帶足
schema 用 JSON Schema、terminate: true 省一輪 round trip
abort 串到每個 subagent session
tokenBudget 是估計、非計費
🟢 自用低風險、🟡 多租戶要額外隔離
License MIT、stars 630（4 天）、active dev、roadmap 在 issue tracker

11. 進一步閱讀

README：github.com/Michaelliv/pi-dynamic-workflows
Pi framework：github.com/earendil-works/pi
Anthropic dynamic workflows：claude.com/blog/introducing-dynamic-workflows-in-claude-code
Epic：issue #1（persisted/resumable workflows）
Faithfulness audit：issue #11
acorn（parse 用）：github.com/acornjs/acorn
TypeBox（schema 用）：github.com/sinclairzx81/typebox
Biome（lint/format）：biomejs.dev

Tutorial: pi-dynamic-workflows — Claude-Code-style 動態工作流移植到 Pi 的完整解讀

Contents