v1.1.0 Pi v1.1.0 源码分析 Analysis of Pi v1.1.0 · 2026-10-07
chord chord 应用组合运行时：服务、复制状态、RPC 与插件 Application composition runtime for services, replicated state, RPC, and plugins
Infra
← → 翻页 · F 全屏 · O 目录 · T 明暗 · L 语言 ← → navigate · F fullscreen · O overview · T theme · L language
01 · 模块结构 01 · Module structure Infra
模块结构 Module structure chord modules (8,817 lines of TypeScript) API 与类型 API & types Facet 宿主 Facet host 服务 Services Delta Context 与 Node Context & Node 导入路径、依赖与使用者 Import paths, dependencies and users 85 index.ts 根入口：重新导出 root entry: re-exports 110 api.ts createFacetHost() 等 createFacetHost() … 339 types.ts Facet · Service · Context … 120 json.ts isJsonValue · copyJson 906 host.ts FacetKernel：激活 · 重载 FacetKernel: activate · reload 6 loader.ts LoadedFacets 的释放 disposing LoadedFacets 713 consumer.ts 远程绑定：稳定外观 remote binding: stable facade 633 provider.ts RemoteServiceProvider 457 state.ts 复制状态 · 订阅 replicated state · subscriptions 248 wire.ts $chord.service 控制调用 $chord.service control calls 176 state-codec.ts 每个订阅一套路径字典 one path dictionary per subscription 151 instances.ts keyed 实例 · generation keyed instances · generations 113 handle.ts ServiceSlot · 代理视图 ServiceSlot · proxy views 2,205 tracker.ts track() · Draft · prepare/adopt 694 index.ts Op · WireOp · apply · 编解码 Op · WireOp · apply · codec 523 diff.ts diffRevisions() 128 apply-immutable-trusted.ts 结构共享地重放 structurally shared replay 75 revision-validator.ts 校验不可信的 op validates untrusted ops 121 index.ts 取消 · 调用范围的值 cancellation · scoped values 416 bundle-loader.ts 校验 SHA-256 · node:vm SHA-256 check · node:vm 222 bundle.ts esbuild → .cjs + 清单 esbuild → .cjs + manifest 232 package.ts package.json 中的 chord.facets chord.facets in package.json @earendil-works/chord @earendil-works/chord/context @earendil-works/chord/delta @earendil-works/chord/bundler @earendil-works/chord/node esbuild 0.28.2 durable · 41 coding-agent · 26 server · 7 client · 2 env · 2 protocol · 2 setup 同步声明依赖； setup declares deps synchronously; 宿主校验整张图后 the host validates the whole graph, 按依赖顺序激活。 then activates in dependency order. 较小的文件： Smaller files: services/errors.ts services/loopback.ts services/state-internals.ts delta/draft.ts node/manifest.ts bundler.ts · node.ts 依赖： dep: 被这些包导入（src 中的文件数）： Imported by (files in src): 标签 = 行数；不依赖任何其他 pi 包 tag = lines; depends on no other pi package facets 服务与状态 services & state delta context · node 最大的单个文件是 delta/tracker.ts：它让提供方像改普通对象一样改状态（Draft 代理），再算出 op。 The largest single file is delta/tracker.ts: it lets a provider change state like a plain object (a Draft proxy) and then computes the ops.
02 · Facet 的一代：装配与激活 02 · A facet generation: assembly and activation Infra
Facet 的一代：装配与激活 A facet generation: assembly and activation One facet generation: setup → validate → connect → activate → dispose setup(env) 能做的事 What setup(env) can do FacetEnvironment assembling：validateFacets() assembling: validateFacets() 例：两个 facet Example: two facets setup 同步调用 facet.setup(env) facet.setup(env), synchronously assembling 外部目录 · 校验 · 排序 catalogues · validate · order connecting 等所有绑定 ready() wait for every binding ready() activating 按拓扑顺序 activate() activate() in topological order active 服务可用 services usable disposing 逆序释放 reverse-order disposal models-facet env.provide(Models, impl) ui-facet env.use(Models) 任何一步失败：#terminate() 释放已建好的一切，原错误与清理错误一起以 AggregateError 抛出（"Facet generation startup and cleanup failed"）。 A failure at any step: #terminate() disposes whatever was built and throws the original error with any cleanup errors as an AggregateError ("Facet generation startup and cleanup failed"). env.use() 声明对单例服务的硬依赖，返回稳定句柄 hard dependency on a singleton; returns a stable handle env.observe() 依赖 keyed 服务，逐个观察活着的实例 depend on a keyed service; observe each live instance env.provide() 提供单例实现（必须是对象） provide a singleton implementation (an object) env.provideMany() 拥有 keyed 服务，返回 spawn(key, impl) own a keyed service; returns spawn(key, impl) env.replicatedState() 创建本 facet 的可变复制状态 create mutable replicated state env.own() 登记一个释放函数 register a disposal function env.onActivate() 依赖就绪后的异步初始化 async init once dependencies are ready env.onDeactivate() 最终拆除（= own） final teardown (same as own) 每个服务只能有一个提供者（facet 或宿主的外部来源） each service has exactly one provider (a facet or a host source) "is provided by both … and …" singleton / keyed 模式必须一致 singleton / keyed modes must match "provided as both singleton and keyed" 每个 use / observe 都必须有提供者 every use / observe needs a provider "… but no facet provides it" 依赖图无环：Kahn 拓扑排序给出激活顺序 no cycles: a Kahn topological sort gives the activation order "Facet dependency cycle: …" 激活：models-facet → ui-facet activate: models-facet → ui-facet 释放：ui-facet → models-facet dispose: ui-facet → models-facet 每个 facet 内部，own() 登记的释放 Inside each facet, own() disposals 函数也按登记的逆序执行； also run in reverse registration order; 错误收集后统一抛出。 errors are collected and rethrown. setup 返回 Promise 会被拒绝（"setup must be synchronous"）；use / provide 等只能在 setup 期间调用，句柄只能在 active 期间使用。 A setup that returns a Promise is rejected ("setup must be synchronous"); use / provide and friends work only during setup, and handles only while active. facet 宿主 facet host 远程绑定 remote bindings 阶段顺序 phase order 依赖 depends on setup 必须同步：返回 Promise 会被拒绝。这让依赖图在任何异步工作开始前就是完整、可校验的。setup must be synchronous: returning a Promise is rejected. That makes the dependency graph complete and checkable before any async work starts.
03 · 热重载 03 · Hot reload Infra
热重载 Hot reload FacetHost.reload(facets): replace implementations without disconnecting handles 失败时 On failure 为什么句柄不断开 Why handles stay connected services/handle.ts 检查 Check phase 必须是 active；ID 唯一且都已激活 phase must be active; IDs unique and all active 候选 setup Candidate setup sameFacetShape()：requires / provides 不变 sameFacetShape(): same requires / provides 激活候选 Activate candidates 按原激活顺序；旧实现仍在服务请求 in the existing order; the old ones still serve 切换 Cut over 单例：local → bindSingleton · 远程 → replace() singletons: local → bindSingleton · remote → replace() 退役旧 facet Retire old facets 逆序 dispose dispose in reverse order 连接 keyed Connect keyed 新实例拿到新的 generation new instances get fresh generations consumer 拿着代理视图 holds a proxy view ServiceSlot 宿主拥有 owned by the host old impl new impl 切换之前（第 1–3 步） Before the cutover (steps 1–3) 释放候选；宿主回到 active，旧实现照常工作。 candidates are disposed; the host returns to active and the old ones keep working. 只有清理也失败时才中止整个代。 Only if cleanup fails too is the whole generation aborted. 切换之后（第 4–6 步） After the cutover (steps 4–6) #abort(previous)：撤销所有 facet 的服务访问并终止整个代， #abort(previous): revoke every facet's service access and terminate 抛出 "Facet reload failed after cutover"。 the generation; throws "Facet reload failed after cutover". 每次访问成员时，代理都通过 slot.resolve() 读当前实现， Every member access resolves the current implementation through 所以 bind(new) 之后，已有句柄立即调用新代码。 slot.resolve(), so after bind(new) existing handles call the new code. unbind() 之后访问 → "Service … is disconnected"。 Access after unbind() → "Service … is disconnected". README 推荐的重载流程：先 load() 候选的 LoadedFacets → 把 facets 交给 reload() → 失败则释放候选； The README's reload recipe: load() the candidate LoadedFacets → pass its facets to reload() → dispose the candidate on failure; 成功切换之后才释放退役的那一代 LoadedFacets（代码随之可被回收）。 dispose the retired LoadedFacets only after a successful cutover (its code can then be collected). keyed 实例按代区分（ServiceInstanceAddress = key + generation）：重载后旧实例关闭，新实例以新 generation 出现。 Keyed instances are generation-specific (ServiceInstanceAddress = key + generation): old instances close, new ones appear with new generations. 1 2 3 4 5 6 步骤 steps 切换 cutover 失败处理 failure handling 形状不变是硬性要求：重载不能增删依赖或提供的服务（"must preserve its service requirements and provisions"）。改形状需要新的一代。 Unchanged shape is a hard rule: a reload cannot add or remove dependencies or provisions ("must preserve its service requirements and provisions"). Changing the shape needs a new generation.
04 · 服务、句柄与 Context 04 · Services, handles and Context Infra
服务、句柄与 Context Services, handles and Context Services: tokens · modes · stable handles · Context 服务令牌：defineService<T>(id, options?) Service tokens: defineService<T>(id, options?) 两种提供方式 Two provision modes 一个调用的路径 The path of one call Context：显式传递的调用上下文 Context: an explicitly passed call context chord/context 远程（默认） remote (default) 本地：{ local: true } local: { local: true } pi.transcript pi.models pi.agent-controller pi.local.presentation-ui singleton provide(S, impl) ↔ use(S) keyed provideMany(S) ↔ observe(S, …) models.select(m, ctx) facet 拿到的句柄 handle the facet holds ServiceSlot.resolve() 取当前实现（重载可换） current implementation (reload-safe) impl.select(m, ctx) 本地对象，或远程绑定的成员代理 local object, or a remote member proxy 成员只能是： members must be: (…json, Context) => Promise 或 ReplicatedState<json> or ReplicatedState<json> 任意 TypeScript 形状； any TypeScript shape; 只在本进程内提供， provided in-process only, 不能跨传输。 never crosses a transport. coding-agent/src/experimental/services 中的例子： Examples from coding-agent/src/experimental/services: 空 ID 会抛错；"$chord." 前缀保留给 chord 自己的控制调用。 An empty ID throws; the "$chord." prefix is reserved for chord's own control calls. 一个服务 ID 只有一个实现； One implementation per service ID; use() 返回稳定句柄。 use() returns a stable handle. spawn(key, impl) 返回 close()； spawn(key, impl) returns close(); observer 对每个活实例被调用， the observer runs for each live 实例关闭时执行其清理函数。 instance and cleans up on close. 来源：facet 提供，或宿主的 serviceSources（外部服务目录）提供—— Source: a facet, or one of the host's serviceSources (external catalogues) 两者不能同时提供同一个服务。 — never both for the same service. 远程成员最后一个 A remote member's last 参数必须是 Context， argument must be a Context, 否则 service_invalid_value。 else service_invalid_value. 绑定关闭后调用 → Calling after the binding service_stale_instance。 closes → service_stale_instance. BACKGROUND_CONTEXT · TODO_CONTEXT 根上下文（TODO 标记"尚待接线"） roots (TODO marks "not wired yet") createContextKey · withContextValue 调用范围内的键值 call-scoped values withAbortSignal · withCancel 合并取消信号（AbortSignal.any） combine cancellation (AbortSignal.any) withoutAbortSignal 保留值，去掉取消 keep values, drop cancellation awaitWithContext(p, ctx) 取消时只拒绝等待者，不中止 p cancellation rejects the waiter only, not p 远程服务 remote services 本地服务 local services facet 侧 facet side Context 远程约束在类型层面检查（RemoteServiceContract<T>），运行时在边界上再校验一次（validateRemoteServiceImplementation）。 The remote constraint is checked in the type system (RemoteServiceContract<T>) and again at runtime at the boundary (validateRemoteServiceImplementation).
05 · 跨进程的服务 05 · Services across processes Infra
跨进程的服务 Services across processes Services across processes: provider ⇄ transport ⇄ binding 提供方进程 Provider process 传输（应用自选） Transport (application's choice) RemoteServiceTransport 使用方进程 Consumer process 订阅更新的 6 种类型 The 6 subscription update types ServiceProviderUpdate 背压与错误 Backpressure and errors 服务实现 service implementation facet 的 provide / provideMany a facet's provide / provideMany RemoteServiceProvider invoke(call) · subscribe(id, mode) createRemoteServiceEndpoint() 处理 "$chord.service" 控制调用 serves "$chord.service" control calls createRemoteServiceBinding() 稳定外观：成员代理 + 复制状态 stable facade: member proxies + replicas facet 的 use() / observe() a facet's use() / observe() 像本地对象一样调用 calls it like a local object state.value · state.subscribe() 按序重放 delta 的副本 replica that replays deltas in order not_allowed not_found mode_mismatch member_not_found member_mismatch instance_not_found stale_instance invalid_value 调用 [...args, ctx] calls [...args, ctx] invoke(call, context) subscribe(serviceId, mode, listener, ctx) chord 不规定外层协议：这些值可以放进 chord does not prescribe the outer protocol: 任何路由 / 请求 / 事件信封。 wrap these values in any envelope. $chord.service 的三种控制调用： Three $chord.service control calls: · catalogue · subscribe(subscriptionId, serviceId, mode) · unsubscribe(subscriptionId) 同进程：createLoopbackServiceTransport(provider) In-process: createLoopbackServiceTransport(provider) 调用 / 订阅 invoke / subscribe 更新 updates state 某成员的一批 op，sequence 连续 a batch of ops for one member, contiguous sequence reset 整份快照：每个 state 都是 ["r", value] full snapshot: every state is ["r", value] unavailable 当前没有提供者 no provider right now replaced 提供者被替换（如重载） the provider was replaced (e.g. reload) spawned keyed 新实例出现 a new keyed instance appeared closed keyed 实例关闭 a keyed instance closed 每个订阅最多缓存 100 个待发更新；第 101 个到来时， At most 100 pending updates per subscription; when update 101 整个队列被替换成一个 { type: "reset", snapshot }。 arrives the whole queue becomes one { type: "reset", snapshot }. 普通 state 更新绝不允许 sequence 跳号。 An ordinary state update never permits a sequence gap. 可以跨边界的 8 个错误码： The 8 error codes that may cross the boundary: 错误码都带 service_ 前缀 all codes carry a service_ prefix 调用 calls 订阅更新 subscription updates 提供方 provider 使用方 consumer 订阅更新有 6 种：state（某成员的一批 op）、reset、unavailable、replaced、spawned、closed。 Subscription updates come in 6 kinds: state (a batch of ops for one member), reset, unavailable, replaced, spawned, closed.
06 · 复制状态 06 · Replicated state Infra
复制状态 Replicated state Replicated state: how one change() becomes a new value elsewhere 发布之后：两类订阅者 After publishing: two kinds of subscriber 远端副本：严格按序 The remote replica: strictly in order change(ctx, draft => …) 不可重入；回调必须同步 no reentrancy; callback must be sync tracker.beginChange() 拿到 Draft 代理 get a Draft proxy mutate(draft) 像普通对象一样改 mutate like a plain object change.prepare() 算出 ops（失败则 abort） compute ops (abort on throw) tracker.adopt(prepared) 换根指针：新不可变版本 swap the root: new immutable revision publish(value, ops) sequence + 1；没有 op 就不发布 sequence + 1; no ops, no publish 内部精确订阅 internal exact subscriptions (ops, sequence, ctx)：同步，拿到每个版本 (ops, sequence, ctx): synchronous, every revision state.subscribe(fn) (value, ctx, delivery)：每个订阅串行 (value, ctx, delivery): serialized per subscription hydrate(7, [["r", v]]) 基线 baseline update(8, ops) apply update(9, ops) apply update(11, …) 跳号！ gap! 回调抛错 → 原值和 sequence 都不变。 Callback throws → value and sequence unchanged. replace(ctx, value) 走同一条路径（prepareReplace）。 replace(ctx, value) takes the same path (prepareReplace). → 远程编码器 / provider 订阅 → remote encoders / provider subscriptions 每个 client × state 独立编码 encoded separately per client × state 先 hydrate（当前值），再 update； hydrate (current value) first, then updates; 最多 100 个待处理：溢出时只留最新的， at most 100 pending: on overflow keep the newest, 所以公开的 sequence 可能跳号。 so public sequences may skip. 异步回调里读捕获的 value，不要读 state.value。 In async callbacks read the captured value, not state.value. update 要求 sequence == 上一个 + 1；否则清空副本并抛出 update requires sequence == previous + 1; otherwise the replica clears and throws "Replicated state update sequence has a gap" provider 侧的 reset 是唯一合法的"重新起跑"：它带着新的基线 sequence， A provider reset is the only legal restart: it carries new baseline sequences, 并清空该订阅的路径字典。 and clears that subscription's path dictionaries. 值是不可变、结构共享的：没变的子树在版本之间共用。 Values are immutable and structurally shared: unchanged subtrees are reused. 1 2 3 4 5 6 顺序 order delta tracker 订阅 subscriptions 违约 violation 一次 change() 恰好发布一个原子版本；回调抛错时值和 sequence 都不变；没有产生 op 就不发布。 A successful change() publishes exactly one atomic revision; if the callback throws, value and sequence stay unchanged; no ops, no publication.
07 · JSON delta 07 · JSON deltas Infra
JSON delta JSON deltas Delta: 7 ops and on-the-wire compression Op（delta/index.ts） Op (delta/index.ts) 字符串怎样变成 op How a string change becomes ops WireOp：只多两种压缩 WireOp: just two compressions r s d a t p m ["r", value] 替换根 replace the root ["s", path, value] 设置字段 / 下标 set a field / index ["d", path] 删除字段 delete a field ["a", path, str] 字符串末尾追加 append to a string ["t", path, n] 字符串去掉前 n 个字符 drop the first n chars of a string ["p", path, start, del, items] 数组 splice array splice ["m", path, perm] 数组重排：new[i] = old[perm[i]] reorder: new[i] = old[perm[i]] 新值以旧值开头 new starts with old → ["a", path, tail] 旧值的后缀是新值的前缀 old suffix = new prefix → ["t", …] + ["a", …] （重叠扫描最多 65,536 字符：滚动日志） (overlap scan up to 65,536 chars: rolling logs) 其他情况 anything else → ["s", path, new] 超过 4,096 个 op → 折叠成一个 ["r", 整个根] more than 4,096 ops → fold into one ["r", whole root] ["#", id, path]：路径第二次出现时定义一个编号 ["#", id, path]: defines an id on a path's second use 之后用数字代替该路径 later ops reference it by number 缩短的元组：同一批内复用上一个 op 的路径 shortened tuples reuse the previous op's path (within a batch) 编码器 / 解码器每个订阅一对，每个实例 × 成员一本字典； one encoder / decoder per subscription, one dictionary per 替换、不可用、关闭、重新 hydrate 时清空。 instance × member; cleared on replace, unavailable, close, hydrate. 例：README 里的 { output, count } Example: the README's { output, count } draft.output += "done\n"; draft.count += 1; // 两次修改 draft.output += "done\n"; draft.count += 1; // two edits → ops [["a", ["output"], "done\n"], ["s", ["count"], 3]] → wire [["a", ["output"], "done\n"], ["s", ["count"], 3]] // 路径各用一次，不定义编号 → wire [["a", ["output"], "done\n"], ["s", ["count"], 3]] // each path used once: no ids 下一批再追加 output（第二次使用）→ ["#", 0, ["output"]], ["a", 0, "…"] next batch appends to output (second use) → ["#", 0, ["output"]], ["a", 0, "…"] 超过 4,096 个 op 时折叠成 ["r", 整个根]；去头 + 追加的重叠扫描最多 65,536 个字符。 More than 4,096 ops fold into ["r", whole root]; the truncate-plus-append overlap scan covers up to 65,536 characters.
08 · 打包与加载插件 08 · Bundling and loading plugins Infra
打包与加载插件 Bundling and loading plugins Bundling and loading facets: esbuild → manifest → verified, then run in node:vm 打包时 At bundle time node/bundle.ts 清单 chord-facets.json Manifest chord-facets.json node/manifest.ts 加载时：createFacetBundleLoader(...).load() At load time: createFacetBundleLoader(...).load() node/bundle-loader.ts 插件包 plugin package package.json 的 "chord": { facets } package.json "chord": { facets } bundleFacetPackage() esbuild：cjs · bundle · 单文件 esbuild: cjs · bundled entries 输出目录 output directory facet-<hash>.cjs + chord-facets.json createFacetBundleLoader load() → LoadedFacets 读清单 read manifest 格式 / 版本必须匹配 format / version must match 校验 SHA-256 verify SHA-256 与 integrity 不符就拒绝 reject on integrity mismatch node:vm 编译 compile in node:vm compileFunction(code, …) 受限 require restricted require 未声明的 external → 报错 undeclared external → error 校验 facets validate facets 然后交给 reload() then hand to reload() · 依赖包（peer）和 @earendil-works/chord 都标为 external peers and @earendil-works/chord stay external · 入口名 facet-<12 位短哈希>-[hash]，扩展名 .cjs entry names facet-<12-char hash>-[hash], .cjs · target node22.19 或 es2022；不支持动态 import；"use strict" 横幅 target node22.19 or es2022; no dynamic import; "use strict" banner · 先写到 .<out>.tmp-<uuid>，再改名替换 write to .<out>.tmp-<uuid>, then rename into place · 旧目录先移到 .old-<uuid>；替换失败 → 移回 old dir moved to .old-<uuid>; swap fails → moved back { "format": "chord.facet-bundle", "version": 2, "plugin": { "id": …, "version": … }, "entries": [{ "file": "facet-….cjs", "integrity": "sha256-…", "externalImports": ["…"], "sourceMap": … }] } 也可以把整个包读成一个 artifact（readFacetBundleArtifact，格式 chord.facet-bundle-artifact）再用 createFacetBundleArtifactLoader 加载。 A bundle can also be read as one artifact (readFacetBundleArtifact, format chord.facet-bundle-artifact) and loaded with createFacetBundleArtifactLoader. 1 2 3 4 5 打包 bundling 产物 artifacts 加载 loading 流程 flow 输出先写到临时目录，再整体改名替换；旧目录先挪到 .old-<uuid>，替换失败就挪回来——加载方永远看不到写了一半的包。 Output goes to a temporary directory and is renamed into place; the old directory is moved to .old-<uuid> first and moved back if the swap fails — a loader never sees a half-written bundle.
扩展点 Extension points pi-coding-agent
chord 的扩展点 Where to plug into chord 1
defineFacet({ id, setup }) defineFacet({ id, setup })
写一个功能单元：在 setup 里声明 use / provide。 Write a unit of functionality: declare use / provide in setup.
2
defineService<T>(id, { local? }) defineService<T>(id, { local? })
定义一个服务契约，远程或本地。 Define a service contract, remote or local.
3
RemoteServiceTransport RemoteServiceTransport
实现 invoke / subscribe，把服务接到任何传输上。 Implement invoke / subscribe to carry services over any transport.
4
FacetLoader FacetLoader
load() 返回 LoadedFacets：静态、组合或从打包文件加载。 load() returns LoadedFacets: static, combined or from bundles.
5
RemoteServiceSource RemoteServiceSource
catalogue() + open()：把另一个进程的服务目录接进宿主。 catalogue() + open(): plug another process's service catalogue into a host.