v1.1.0 Pi v1.1.0 源码分析 Analysis of Pi v1.1.0 · 2026-10-07
durable durable 持久化的会话、任务与文档运行时 Durable conversation, task, and document runtime for Pi
Infra Infra pi-ai
← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言 ← → navigate · F fullscreen · O overview · T theme · L language
01 · 模块结构 01 · Module structure Infra
模块结构 Module structure pi-durable modules (20,404 lines of TypeScript) 核心类型 Core types src/*.ts · 1,786 Session session/ · 2,009 Harness harness/ · 7,015 存储后端 Storage backends storage/ · 3,221 执行环境 Execution env env/ · 2,250 内置工具 Built-in tools tools/ · 1,313 一致性测试套件 Conformance suites testing/ · 2,810 11 个导入路径与依赖 11 import paths and dependencies 1,107 types.ts Storage · TaskState · 记录 Storage · TaskState · records 208 documents.ts defineDoc 34 entries.ts pi.user … pi.compaction 8 tasks.ts defineTask 1,042 transaction.ts 一次原子提交 one atomic commit 569 session.ts 唯一的提交线 the single commit line 301 observation.ts view · watch 1,435 scheduler.ts 任务调度与恢复 task scheduling & recovery 681 generation.ts pi.generation 任务 the pi.generation task 509 tool.ts pi.tool 任务 the pi.tool task 453 compaction.ts pi.compaction 任务 the pi.compaction task 437 harness.ts Harness.open() · root() 425 events.ts watchEvents() 934 sqlite/storage.ts SQLite 核心 SQLite core 845 jsonl/storage.ts JSONL 核心 JSONL core 864 memory.ts MemoryStorage 327 index.ts ExecutionEnv 接口 the ExecutionEnv interface 1,226 node.ts NodeExecutionEnv 522 node-watch.ts 文件监视 file watching 500 edit-diff.ts edit 的匹配与差异 edit matching & diffs 216 read.ts read 工具 the read tool 25 index.ts CodingTools：4 个工具 CodingTools: 4 tools 38 runner.ts register…Conformance() 1,662 storage-conformance.ts 存储用例 storage cases 511 env-conformance.ts 环境用例 environment cases . /env /env/node /tools /storage/memory /storage/jsonl /storage/jsonl/node /storage/sqlite /storage/sqlite/cloudflare /storage/sqlite/node /testing chord ^1.1.0 pi-ai ^1.1.0 diff 8.0.4 typebox 1.3.27 依赖： deps: 被导入：coding-agent 17 个文件 · env 3 个文件 imported by: coding-agent (17 files) · env (3 files) session harness 存储 storage env 11 个导入路径让宿主只加载需要的部分：例如 ./storage/sqlite 与 ./storage/jsonl 的可移植核心不依赖 Node API，可以跑在 Bun 或 Cloudflare Durable Object 上。 11 import paths let a host load only what it needs: the portable cores ./storage/sqlite and ./storage/jsonl, for example, need no Node APIs and run on Bun or in Cloudflare Durable Objects.
02 · 概念模型 02 · The model Infra
概念模型 The model The model: everything is a record committed to storage 一个 Conversation（句柄不持有状态，按 id 比较） One Conversation (handles hold no state; compare by id) Entry：不可变记录 Entries: immutable Document：类型化 JSON Documents: typed JSON Task 与 Submission Tasks & submissions Agent：会话用什么跑 Agent: what a conversation runs with 宿主程序 Host program submit · wait · viewState · watch Harness 一份打开的存储 + 运行 agent 的机制 one open storage + the machinery that runs agents Registry 本进程安装的 extension（不存储，可随时变） extensions this process installed (not stored, may change) Session：唯一的原子提交线 Session: the single line of atomic commits 提交前什么都不会被看到 nothing is shown before it is stored pi.user pi.system pi.assistant pi.tool-result pi.reset pi.compaction pi.agent pi.provider pi.live pi.inbox pi.usage app.* pi.generation pi.tool pi.compaction 用户输入 user input 提示词变化 prompt changes 模型回答 model answer 工具结果 tool result 新上下文起点 new context start 摘要 summary 模型 · 工具 · cwd model · tools · cwd sessionId 运行中的生成与工具 running work 排队的提交 queued submissions 花费 spend 你自己的 your own 调用模型 calls the model 一次工具调用 one tool call 做摘要 summarizes Task 每一步都存检查点， A task checkpoints every step, 重启后从最后一步继续。 so a restart continues from it. Submission：交给会话的 A submission: input or an entry 输入或条目，可以 wait()。 handed over; you can wait(). 要么全存，要么全不存 all of it or none model · thinkingLevel 模型与思考强度 model and thinking extensions · tools 按名字存储 stored by name instructions 最后渲染的提示词段 rendered as the last section cwd 环境由它构建 the env is built from it 存在 pi.agent 文档里；每次使用 Stored in the pi.agent document; 时对照 Registry 解析。 resolved against the registry at each use. 扩展被卸载后，选它的会话只是 After an extension is uninstalled, 暂时拿不到它，名字仍保留。 conversations just stop getting it; the stored name stays. settings：每次使用时读取，从不存储 settings: read at every use, never stored 条目与会话 entries & conversations 文档 documents 任务 tasks 扩展与 agent extensions & agent conversation.commit(tx => …) 可以在一次原子写入里追加条目、改文档、建任务——要么全部存下，要么全不存。conversation.commit(tx => …) can append entries, edit documents and create tasks in one atomic write — all of it is stored or none of it.
03 · 回答一次输入 03 · Answering one input Infra
回答一次输入 Answering one input One answered input, as entries and tasks over time pi.generation 的检查点阶段 pi.generation checkpoint phases harness/generation.ts 进程崩溃之后 After a crash harness/tool.ts submit(input) pi.user pi.generation 第 1 轮 turn 1 pi.system pi.assistant 工具调用 tool calls pi.tool #1 pi.tool #2 pi.tool-result × 2 pi.generation 第 2 轮 turn 2 pi.assistant 回答 answer wait() → done prepare request retry poll tools 宿主 Host 任务 Tasks 提交的条目 Committed entries pi.system 只在提示词或工具变化时写入 pi.system only when the prompt or tools changed generation 拥有并等待它们 owned and awaited by the generation 组装请求；需要时先等一次压缩 build the request; wait for a compaction if needed 流式调用模型（部分回答每 100 ms 提交） stream the model (partials committed every 100 ms) 退避后重试：最多 3 次，基数 2000 ms back off and retry: up to 3, base 2000 ms 延迟的回答：默认每 5000 ms 轮询 deferred answers: poll every 5000 ms by default 拥有本轮的 pi.tool 任务并等待 own this round's pi.tool tasks and wait pi.tool 先提交意图（phase "execute"），再运行 execute()。 pi.tool commits its intent (phase "execute") before execute() runs. replay: "safe" 重新打开时重跑 rerun on reopen 默认 "unsafe" default "unsafe" 模型收到 interrupted 错误结果： the model gets an interrupted error: "Tool … was interrupted and may have partially run" 其余工作保持 pending；resume()（或下一次 submit / wait）启动调度器， Everything else stays pending; resume() (or the next submit / wait) starts the 从最后的检查点继续。 scheduler, which continues from the last checkpoints. 同一个 requestId 的重复 submit 返回原来的 submission。 A resubmit with the same requestId returns the existing submission. 条目 entries 任务 tasks 产生 produces 作为输入 feeds generation 的检查点阶段：prepare → request → retry（退避重试，默认最多 3 次）→ poll（延迟回答，默认每 5000 ms）→ tools。 Generation checkpoint phases: prepare → request → retry (back-off, up to 3 by default) → poll (deferred answers, every 5000 ms by default) → tools.
04 · 任务：状态机与所有权 04 · Tasks: state machine and ownership Infra
任务：状态机与所有权 Tasks: state machine and ownership Tasks: a resumable state machine with ownership TaskOutcome 所有权：每个任务都有主人 Ownership: every task has an owner pending 可被调度 eligible to run running 一次内存中的调用 one in-memory invocation waiting on: [...] · policy completing 结果已定，等下属工作 outcome held terminal 永久结果 permanent receipt 会话 conversation pi.generation pi.tool subagent 子会话 child conv 提交 waiting commit waiting on 都结束 all of on done 没有下属工作时直接结束 straight to terminal when nothing below is live 重启后，running After a restart, running 显示为 pending shows as pending 每次状态转换都在 Session 的 Every transition is serialized 提交线上串行完成。 on the Session's commit line. 活状态：pending · running · Live: pending · running · waiting · completing completed 带 result carries result failed 实现主动提交的失败 failure the task committed aborted 走了 abort 协议 handled abort orphaned 定义或迁移不可用 definition / migration missing faulted 运行时检测到违约 runtime contract failure · 中止调用会中止子会话；父会话要等子会话空闲才算空闲。 · Aborting the call aborts the child; the parent is idle only once the child is. · 中止自下而上：先中止它拥有的工作，再运行它自己的 abort 处理。 · Abort runs bottom-up: owned work first, then the task's own abort handler. · { background: true } 的任务是边界：它拥有的工作不受父中止影响，也不让父忙碌。 · A { background: true } task is a boundary: its work survives the parent's abort and keeps it idle. · 等待策略：failFast（第一个失败就中止其余）或 allSettled。 · Join policy: failFast (the first failure aborts the rest) or allSettled. harness.taskGraph() 把所有活任务显示为一个 Chord state（例：24-child-tasks.ts） harness.taskGraph() shows every live task as one Chord state (example: 24-child-tasks.ts) README：子任务 README: child tasks // 子任务：创建后进入 waiting // child tasks: create, then wait pay: async (task, runtime, ctx) => { await runtime.commit(async tx => { const ids = [...] // tx.createTask(…) return { status: "waiting", checkpoint: { phase: "decide", ids }, on: ids, policy: "failFast" }; }, ctx); }, 转换 transition 恢复 / 捷径 resume / shortcut 重启 restart completing 每次状态转换都在 Session 的提交线上串行完成；重启后 running 的任务显示为 pending，直到再次运行。 Every transition is serialized on the Session's commit line; after a restart, tasks that were running show as pending until they run again.
05 · 会话忙时的提交 05 · Submitting to a busy conversation Infra
会话忙时的提交 Submitting to a busy conversation Submitting to a busy conversation: steer · followUp · reject · write 它们什么时候被放进转录 When they are placed into the transcript pi.inbox 文档 The pi.inbox document harness/inbox.ts 撤回、重置与幂等 Withdraw, reset and idempotency root.submit(x, ctx) 会话正在跑一个 run a run is in progress whenBusy: "reject" 立即抛出 ConversationBusy throws ConversationBusy at once whenBusy: "steer" 进 pi.inbox，mode steer into pi.inbox as steer 默认：followUp default: followUp 进 pi.inbox，mode followUp into pi.inbox as followUp { type: "write", entry } 排队写条目，不问模型 queue an entry, no model call 第 1 轮 turn 1 工具轮 tool round 第 2 轮 turn 2 回答 answer steer：当前工具轮之后，加入正在跑的工作 steer: after the current tool round, joins the running work followUp：run 回答后放入， followUp: placed when the run 开启下一个 run answers; starts the next run steeringMode / followUpMode 默认 "one-at-a-time"：每轮放一条；"all"：一次全放。 steeringMode / followUpMode default "one-at-a-time": one per turn; "all": every queued item at once. run 失败时，排队项留在 inbox，等下一次提交时按先后放入。 If a run fails, queued items stay until the next submission places them, oldest first. { items: [ { id, mode: "steer", content }, { id, mode: "followUp", content }, … writes and resets queue here too ] } UI 从 view.docs["pi.inbox"] 直接画出排队区。 A UI draws its queue straight from view.docs["pi.inbox"]. submission.abort(ctx) 撤回仍在排队的提交 withdraw a queued submission root.abort(ctx) 撤回排队输入（写入保留），中止当前工作 withdraw queued inputs (writes stay), abort work root.reset(note?, ctx) 忙时像写入一样排队 queued like a write while busy requestId 同一 ID 重复提交 → 原来那个 submission same ID again → the existing submission steer followUp write reject steeringMode / followUpMode 默认 "one-at-a-time"：每轮放一条；设为 "all" 时一次放入全部排队项。steeringMode / followUpMode default to "one-at-a-time": one per turn; "all" places every queued item at once.
06 · 文档与观察 06 · Documents and watching Infra
文档与观察 Documents and watching Documents and watching: everything a UI needs is committed state 内置文档（scope: conversation） Built-in documents (scope: conversation) 三种观察方式 Three ways to watch pi.agent pi.provider pi.live pi.inbox pi.usage viewState(ctx) 只读 Chord state：entries + docs read-only Chord state: entries + docs watch(ctx) 同一个视图 + 每次提交的精确 Chord op the same view + each commit's exact Chord ops watchEvents(…) coding-agent 风格事件（实验性） coding-agent style events (experimental) 文档 document 内容 holds history fork 模型 · 扩展 · 工具 · cwd model · extensions · tools · cwd rewindable asOf sessionId = UUIDv7 latest initial 运行中的生成、工具、压缩 running generation, tools, compactions latest initial 排队的提交 queued submissions latest initial 按模型 / 工具的花费 spend per model / tool latest initial pi.agent 可回溯：分叉拿到分叉点那一刻的 agent；其余分叉时从初始值开始。 pi.agent is rewindable: a fork gets the agent as of the fork entry; the rest start fresh. 你自己的文档（README） Your own document (README) const Todos = defineDoc<{ items: string[] }>({ kind: "app.todos", version: 1, scope: "conversation", history: "latest", // 或 "rewindable" history: "latest", // or "rewindable" fork: "initial", // "current" | "asOf" initial: () => ({ items: [] }), }); await root.commit(async tx => (await tx.doc(Todos, root.id)).items.push("write docs"), ctx); 每次相关提交后更新 updated after every commit that touches it 慢消费者最多 100 帧，之后换成一帧完整视图 slow consumers: 100 frames, then one whole-view frame 落后超过 100 批 → 一个新的 snapshot 事件 more than 100 batches behind → a fresh snapshot event 迟到或重连的客户端从当前视图开始，不会重放历史；部分回答与工具输出默认最多每 100 ms 提交一次。 Late or reconnecting clients start from the current view, nothing is replayed; partials and tool output commit at most every 100 ms. 文档 documents 观察 watching history: "rewindable" 的文档可以用 snapshotAsOf() 读旧值；fork 决定分叉时的初值：initial、current 或 asOf。Documents with history: "rewindable" can be read as of the past with snapshotAsOf(); fork decides a fork's start value: initial, current or asOf.
07 · 存储后端 07 · Storage backends Infra
存储后端 Storage backends Storage: one commit(writes) interface, four backends StorageWrite 的 8 种类型 The 8 StorageWrite types 读取接口 Read interface Memory SQLite (Node) JSONL (Node) Cloudflare DO Session 把一次事务变成一组写入 turns a transaction into writes Storage.commit(writes, ctx) → Seq 全部写入或全部不写；Seq 单调递增 all writes or none; Seq increases conversation entry task submission document.create document.copy document.change document.retire conversation() scanConversations() entry() scanEntries() findLatestHeadMarker() task() scanTasks() submission() submissionByRequest() scanSubmissions() document() findDocument() scanDocuments() mintId() close() 文档还有修订历史（rewindable） documents also keep revisions 每个查询都分页（limit + cursor）；同一时刻只能有一个进程拥有一个存储，没有跨进程锁。 Every scan is paged (limit + cursor); one process owns a storage at a time, with no cross-process locking. MemoryStorage 什么都不持久化；测试与示例 nothing persisted; tests and examples 从包根导入 exported from the package root 与其他后端跑同一套一致性测试 runs the same conformance suite openNodeSqliteStorage(file) 一个数据库文件 one database file WAL · synchronous = NORMAL wal_autocheckpoint 1000 · busy 5000 ms 崩溃不丢提交；断电可能丢最新一个 survives crashes; power loss may drop the newest openNodeJsonlStorage(dir, ctx) 一个目录里的追加文件 append-only files in a directory main.jsonl：每次提交追加一行标记 main.jsonl: one marker line per commit {format:1, type:"commit", seq, writes} task-<id>.jsonl / doc-<id>.jsonl 旁路文件 task-<id>.jsonl / doc-<id>.jsonl sidecars { fsync: true } openDurableObjectSqliteStorage(…) 一个 Durable Object 的 SQLite one Durable Object's SQLite 提交 = 它的存储事务 commits = its storage transactions 8 张表： 8 tables: durable_metadata record_ids conversations entries tasks submissions documents document_revisions 可移植核心 /storage/sqlite、/storage/jsonl 不依赖 Node：可跑在 Bun 或 Durable Object 上 portable cores /storage/sqlite and /storage/jsonl need no Node APIs: Bun, Durable Objects 提交 commit 实现 implements SQLite：journal_mode = WAL、synchronous = NORMAL、wal_autocheckpoint 1000 页、busy 超时 5000 ms——提交能挺过进程崩溃，断电可能丢最新一个。 SQLite: journal_mode = WAL, synchronous = NORMAL, wal_autocheckpoint 1000 pages, a 5000 ms busy timeout — commits survive process crashes; power loss may drop the newest.
08 · 扩展、钩子与重载 08 · Extensions, hooks and reload Infra
扩展、钩子与重载 Extensions, hooks and reload Extensions: tools · sections · hooks · wraps · tasks defineExtension({ name, … }) 钩子挂在三个内置任务上 Hooks attach to the three built-in tasks BUILTIN_TASKS 系统提示词 The system prompt harness/prompt.ts 重载 Reload tools sections hooks wraps tasks GenerationTask pi.generation ToolTask pi.tool CompactionTask pi.compaction defineTool：TypeBox 参数 defineTool: TypeBox params section(name, fn) hook(Task, {…})：见右 hook(Task, {…}): see right wrapTool / wrapSection 自己的持久任务 your own durable tasks 同名工具：后安装的覆盖先安装的 Same-name tool: the later extension wins beforeRequest 替换一次请求的消息 replace one request's messages afterResponse 看到回答 observe the answer onYield { continue } 让 run 继续 { continue } keeps the run going afterTools 一轮工具都结束后 after a round's tools beforeTool { block } 或改写参数 { block } or rewrite args afterTool 替换结果 replace the result beforeCompact 拒绝或自给摘要 decline or supply a summary 钩子只在选中该扩展的会话里生效：configure({ extensions: { add: [Guard] } }) Hooks act only in conversations that select their extension: configure({ extensions: { add: [Guard] } }) 按顺序渲染所选扩展的段；instructions 最后渲染。 Selected extensions' sections render in order; instructions last. <cwd>\n/work/repo\n</cwd> ← section("cwd", i => i.env?.cwd) 段与工具变化存成转录里按位置的 pi.system 条目， Section and tool changes are stored as positional pi.system 只重发变化的部分 → provider 提示缓存保持命中。 entries; only changes are resent → prompt caches stay warm. 每次都变的段（如当前时间）会破坏缓存。 A section that changes every time (e.g. the time) defeats that. registry.install(ext)：同名就原地替换，一步完成。 registry.install(ext): same name replaces in place, in one step. 已开始的工作保留它拿到的代码： Work that started keeps the code it took: · 运行中的工具调用用旧实现跑完 · a running tool call finishes on the old code · 每个任务阶段开始时解析一次钩子和 agent · each task phase resolves hooks and agent once · 下一个阶段 / 请求 / 调用用新代码 · the next phase / request / call uses new code 重启后重新安装同样的扩展，其 pending 任务就会继续。 After a restart, install them again; their pending tasks resume. 扩展 extensions 内置任务与钩子 built-in tasks & hooks 提示词 prompt 重载 reload 钩子：generation 有 beforeRequest、afterResponse、onYield（{ continue } 让 run 继续）、afterTools；工具有 beforeTool（{ block } 或改参数）、afterTool；压缩有 beforeCompact。 Hooks: generation has beforeRequest, afterResponse, onYield ({ continue } keeps the run going), afterTools; tools have beforeTool ({ block } or rewrite arguments), afterTool; compaction has beforeCompact.
扩展点 Extension points pi-coding-agent
durable 的扩展点 Where to plug into durable 1
defineExtension({ tools, sections, hooks, wraps, tasks }) defineExtension({ tools, sections, hooks, wraps, tasks })
把功能打包成一个可安装、可重载的扩展。 Package functionality as an installable, reloadable extension.
2
defineTool({ parameters, execute, replay }) defineTool({ parameters, execute, replay })
TypeBox 参数；replay: "safe" 让崩溃后重跑。 TypeBox parameters; replay: "safe" reruns after a crash.
3
hook(Task, { … }) hook(Task, { … })
观察或调整内置任务：beforeTool、onYield、beforeCompact… Observe or adjust built-in tasks: beforeTool, onYield, beforeCompact…
4
defineDoc / defineEntry / defineTask defineDoc / defineEntry / defineTask
你自己的文档、条目和持久任务。 Your own documents, entries and durable tasks.
5
Storage Storage
实现 commit + 读取接口，用一致性套件验证。 Implement commit + the read interface, verify with the conformance suite.