pi 扩展机制：加载、执行与能力

pi 的扩展系统解决一个问题：把 agent 内部流程以事件形式暴露出来，让外部代码能观察、拦截、修改它的行为。Claude Code 这类 agent 默认是黑盒——你不知道它中途改了哪些文件、跑了哪些命令、每一轮花了多少 token、有没有切模型。pi 的做法是给一套扩展机制，把内部流程白盒化，挂回调就能拿到这些信息。

这篇文章讲 pi 扩展机制怎么设计：扩展长什么样、怎么被加载、怎么被执行、能做哪些事。

扩展是什么：一个 `.ts` 文件，导出一个 factory

pi 扩展的契约极简——一个 TypeScript module，export default 一个 factory：

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";

export default function (pi: ExtensionAPI): void {
  // 在这里订阅事件、注册工具/命令/provider
}

无需预编译：loader 用 jiti.import 在运行时编译 .ts，写完直接 pi -e my-ext.ts 就能用。这是 pi 降低扩展开发门槛的第一个决策——去掉 build 步骤。

用后端工程师熟悉的话说：pi 是一个带完整能力的框架（能调 LLM、执行工具、管会话），扩展是一组注册到框架的回调，挂在框架流程的切面上。类似 Spring 的 @EventListener + @Component。

加载流程：两阶段初始化

这是理解 pi 扩展最关键的部分。整个加载分两个阶段：加载阶段（只注册，不执行） 和 就绪阶段（stub 替换为真实实现）。

加载阶段：factory 执行，但 action 是 throwing stub

加载一个扩展的真实顺序：

1. createExtensionRuntime()          造一个带 throwing stub 的 runtime
2. createExtension()                  造一个空 extension 容器（handlers 是空 Map）
3. createExtensionAPI(ext, runtime)   组合成 api  ← api 在这步就组合好，先于扩展加载
4. jiti.import(my-ext.ts) → factory   加载你的扩展文件
5. factory(api)                       执行你的 export default
     └─ pi.on("tool_execution_end", h)  把 h 塞进 extension.handlers Map
                                         (h 函数体没执行，只是存起来)
6. 返回 extension（带满 handlers）

几个反直觉的点：

api 先于扩展组合好。先把 api 搭好当门面，再执行 factory 让你往里填 handler。api 是贯穿全程的稳定门面，内部持有的 extension 和 runtime 引用不变。
加载阶段执行的是 factory 函数体（你的 export default function(pi) { ... } 会跑），但只能做注册（on/registerTool/registerCommand）。
handler 函数体不会执行。pi.on("tool_execution_end", handler) 只是把 handler 塞进 Map，handler 要等运行时 loop emit 事件才被调。

throwing stub：防加载阶段误调 action

stub 是占位实现，调用即抛错。加载阶段，runtime 上所有 action 方法（sendMessage/abort/setModel 等）都是 throwing stub：

function createExtensionRuntime(): ExtensionRuntime {
  const notInitialized = () => {
    throw new Error("Extension runtime not initialized. Action methods cannot be called during extension loading.");
  };
  return {
    sendMessage: notInitialized,         // action 方法全是 stub
    setActiveTools: notInitialized,
    setModel: () => Promise.reject(...),
    // ...
    registerProvider: (name, config) => { runtime.pendingProviderRegistrations.push({...}); },  // 允许：只是排队
  };
}

它能防止在加载阶段误调 action。如果扩展在 factory 里写了 pi.sendMessage("hello")，此时 session/agent 还没建好，会触发 stub 抛错，避免拿到半初始化对象产生难以定位的 bug。

区分清楚：on/registerTool 这些注册类方法在加载阶段允许（只往 Map 塞数据），action 类方法才是 stub。

throwing stub 的本质是加载阶段与运行阶段的时序分离。扩展 factory 在启动早期执行，但 runtime 的真实实现依赖的东西（agent、session、modelRegistry）要到后续启动流程才创建。加载阶段没有这些依赖，action 方法无从委托，于是用 throwing stub 占位（有点像Spring的IOC机制）。stub 的作用是双重的：

结构完整：给 ExtensionAPI 一个结构完整的对象，factory 里可以正常调 on/registerTool 这些注册类方法。

快速失败：action 被误调时立即抛错，而非静默返回半初始化状态。

bindCore 是阶段切换点：之前 action 是 stub，之后替换为真实实现。这是两阶段生命周期的防御性设计，不是循环依赖解法——pi 这里压根没有循环依赖，扩展与 runtime 之间是单向时序依赖。

就绪阶段：bindCore 替换 stub

bindCore 把 throwing stubs 替换为真实实现：

runtime.sendMessage = actions.sendMessage;   // stub → 真实
runtime.abortFn   = ...                       // ctx.abort() 这时才真正工作
// flush pending provider registrations（加载时排队的）

这之后 action 才能用。bindCore 是"加载完成 → 可用"的分界点。

所以需要注意的是插件里的注册和执行需要分开：factory 里只做注册（on/registerTool），别做真正的初始化（构造上下文）。初始化代码该放在 session_start 事件（reason: startup）的 handler 里——那时 runtime 已 bindCore 完成，action 可用。factory 是注册阶段，session_start 才是初始化阶段。

执行流程：loop emit → 扩展先于系统其他处理

事件直接同步调扩展，没有中间缓冲

emit是AgentEventSink 对象的命名，它的作用是事件发送器。

我看源码时以为会存在中间层来转发：“loop emit → 系统中间层消费 → 再转发扩展”。实际没有中间缓冲层，是 AgentSession 直接 await 调扩展 handler。

private _handleAgentEvent = async (event: AgentEvent): Promise<void> => {
  // steering/follow-up 队列处理 ...
  await this._emitExtensionEvent(event);   // ① 扩展先处理（await，阻塞等）
  this._emit(event);                        // ② 用户 listener（--mode json 输出在这）
  // ③ session 持久化（appendMessage）
};

事件分发是有三路顺序固定：

扩展先：await _emitExtensionEvent，扩展 handler 在这里被同步 await 调用。扩展改的结果才是后续看到的。
用户 listener：_emit 通知 --mode json 的 stdout 输出、TUI 重绘等。
持久化：sessionManager.appendMessage。

这个顺序意味着扩展先于用户看到结果。扩展可以修改消息内容、拦截工具调用，用户看到的已经是扩展处理后的版本。

事件翻译层

底层 AgentEvent（loop 协议）和扩展事件（应用层协议）是解耦的。_emitExtensionEvent 负责翻译：

对 tool_execution_end：字段一致，只包装。
对 turn_start/end：附加 turnIndex、timestamp。
对 agent_end：附加 willRetry。
对 message_end：走 emitMessageEnd（链式，可替换消息内容）。

loop 不关心 turnIndex/retry，应用层不关心 partial 的 delta 结构，各管各的。

runner.emit：分发 + 异常隔离 + 延迟 ctx

async emit<TEvent extends RunnerEmitEvent>(event: TEvent): Promise<...> {
  const ctx = this.createContext();        // 延迟构建 ctx
  let result;
  for (const ext of this.extensions) {      // 按扩展加载顺序
    const handlers = ext.handlers.get(event.type);
    if (!handlers || handlers.length === 0) continue;
    for (const handler of handlers) {       // 按 handler 注册顺序
      try {
        const handlerResult = await handler(event, ctx);   // ← 你的 handler 在这被调
        if (this.isSessionBeforeEvent(event) && handlerResult?.cancel) return result;
      } catch (err) {
        this.emitError({ extensionPath: ext.path, event: event.type, error, stack });
        // 异常被吃掉，不影响其他 handler 和主流程
      }
    }
  }
  return result;
}

三个设计要点：

按扩展顺序、handler 顺序执行：可预测的执行顺序。
handler 异常被 try/catch：单个 handler 报错不影响其他，emitError 记录，loop 不崩。扩展自身出错不会影响 pi 主流程。
**session_before_* 事件支持 cancel**：result.cancel 立即返回取消。扩展能拦截会话切换、压缩等关键决策。

链式 emit：让多个扩展依次修改同一结果

普通 emit 只是通知。有些事件需要修改——多个扩展依次改同一个结果，后一个看到前一个的修改。这类事件有专用 emit：

emitMessageEnd：链式替换消息内容（校验 role 不变）。
emitToolResult：链式修改 content/details/isError/usage。
emitToolCall：event.input 可 mutate（修改工具参数），返回 {block: true} 阻止调用。
emitContext：链式修改 messages。
emitBeforeProviderRequest：替换发往 LLM 的 payload。

多个扩展能协作修改同一条消息或工具调用，互不覆盖。

createContext：延迟求值 + assertActive

每次 emit 都 createContext()。ctx 用 getter 实现延迟求值：

createContext(): ExtensionContext {
  const runner = this;
  return {
    get model() { runner.assertActive(); return getModel(); },  // 调用时才取当前 model
    abort: () => { runner.assertActive(); runner.abortFn(); },  // ← ctx.abort() 真身
    // ...
  };
}

延迟求值：ctx 反映调用时的最新状态，不是注册时的快照。handler 注册时 model 可能还没选，调用时取的是当前 model。
assertActive：检查 runtime 是否 stale（会话替换后）。扩展若用捕获的旧 ctx 会抛错，配合 withSession 回调防 use-after-replace。

扩展能做什么：三类能力

扩展的 API（ExtensionAPI）既是"观察者"（on 事件）又是"参与者"（register/动作），一个对象覆盖扩展的全部交互能力。归纳成三类。

订阅事件（on）：观察/拦截 pi 内部流程

pi.on("tool_execution_end", (event, ctx) => { ... });
pi.on("message_end", (event, ctx) => { ... });
pi.on("session_before_compact", (event, ctx) => { ... });

pi 定义了 30+ 事件，分七类：资源事件、会话事件、Agent 事件、Provider 事件、工具事件、模型事件、输入事件。部分事件有返回值（hook 语义）：context 返回 messages、tool_call 返回 block、message_end 返回替换消息、session_before_compact 返回 cancel/compaction。

这是审计类扩展的主要形态——只挂回调，不增加新能力。

注册工具（registerTool）：给 LLM 增加新工具

pi.registerTool({
  name: "tic_tac_toe",
  label: "Tic-Tac-Toe",
  description: "Execute ONE tic-tac-toe action as Player O. ...",   // LLM 看到的工具说明
  promptSnippet: "Play a tic-tac-toe action ...",                   // 注入 prompt 的片段
  promptGuidelines: ["When it is your tic-tac-toe turn, ..."],       // 注入 prompt 的规则
  parameters: Type.Object({ ... }),                                  // 工具参数 schema
  async execute(toolCallId, params, signal, onUpdate, ctx) { ... }, // LLM 调用时执行
});

on 是挂回调观察 pi 内部流程；registerTool 是给 LLM 增加新工具，LLM 可以主动调用。工具的 execute 由 loop 在 LLM 决定调用时触发，收到的 ctx 和事件 handler 拿到的是同一套。

注册命令/provider/快捷键：扩展 UI 入口和模型接入

pi.registerCommand("timed", { handler: async (args, ctx) => { ... } });   // 斜杠命令
pi.registerShortcut("ctrl+x", { handler: (ctx) => { ... } });             // 键盘快捷键
pi.registerFlag("verbose", { type: "boolean", default: false });          // CLI flag
pi.registerProvider("my-llm", { baseUrl, apiKey, models: [...] });        // 自定义 LLM provider

命令：用户输入斜杠命令触发，handler 收 ExtensionCommandContext（有会话控制权）。
快捷键：键盘绑定。
Flag：CLI 参数，值存 runtime。
Provider：扩展可注册完整 LLM provider（自定义 baseUrl、API、OAuth、模型列表），让 pi 接入企业内部 API、代理、自定义协议。

能力	方法	触发方	ctx 类型
订阅事件	`pi.on(event, handler)`	loop 内部流程自动 emit	ExtensionContext
注册工具	`pi.registerTool(tool)`	LLM 主动调用	ExtensionContext
注册命令	`pi.registerCommand(name, opts)`	用户输入斜杠命令	ExtensionCommandContext
注册 provider	`pi.registerProvider(name, config)`	启动时加载模型列表	—
注册快捷键/flag	`pi.registerShortcut/registerFlag`	用户按键/CLI 传参	—

Context 三级权限：权限递增

扩展 handler 收到的 ctx 按场景分三级，权限递增：

ExtensionContext（普通事件）——基础能力：UI 方法、cwd、model、signal、abort、compact、getSystemPrompt 等。
ExtensionCommandContext（命令 handler）——继承 ExtensionContext，增加会话控制：waitForIdle/newSession/fork/switchSession/reload。
ReplacedSessionContext（会话替换后）——继承 ExtensionCommandContext，增加 sendMessage/sendUserMessage。会话替换后（newSession/fork/switchSession）用 withSession(ctx) 回调拿新 ctx，防 use-after-replace。

普通事件 handler 拿不到会话控制权（不能乱切会话），只有用户显式触发的命令才有。权限按需授予。

UI 能力也通过接口注入，不同 mode（tui/rpc/print）提供不同实现。hasUI 标志让 handler 判断是否有 UI 能力——在 --mode json 下跑的扩展拿不到 TUI 的 select/confirm，但能正常记日志。

会话替换的安全：assertActive + withSession

pi 支持会话切换（newSession/fork/switchSession/reload）。这带来一个危险：扩展如果捕获了旧 ctx，在会话替换后还用它，会操作到错误的会话。

pi 的解法是两手：

invalidate：会话替换后标记旧 runtime 为 stale。
assertActive：每次 action 调用前检查，stale 则抛错，错误信息明确：“Do not use a captured pi or command ctx after newSession/fork/switchSession/reload.”

配合 withSession(ctx) 回调模式——会话替换后用回调拿新 ctx，而不是用捕获的旧 ctx。从机制上杜绝 use-after-replace。

完整时序：从 `pi -e my-ext.ts` 到 handler 被调

[加载] loader.loadExtension
  createExtensionRuntime()              ← throwing stubs
  createExtension()                     ← 空 extension 容器
  createExtensionAPI(ext, runtime)      ← api 先组合好（先于加载扩展）
  jiti.import(my-ext.ts) → factory
  factory(api)                          ← 执行你的 export default
    └─ pi.on("tool_execution_end", h) → 存进 extension.handlers

[就绪] runner.bindCore(actions)
  runtime stubs → 真实实现（abortFn 等）

[运行] agent loop 执行 bash 工具 → emit AgentEvent{type:"tool_execution_end",...}
  └─ AgentSession._handleAgentEvent(event)
       ├─ await _emitExtensionEvent(event)           ① 扩展先
       │    └─ 翻译成 ToolExecutionEndEvent
       │    └─ runner.emit(extensionEvent)
       │         └─ createContext()                   ← ctx 延迟求值
       │         └─ for ext: for handler:
       │              try { await handler(event, ctx) }  ← 你的 handler
       │              catch { emitError }              ← 异常隔离
       ├─ _emit(event)                                 ② --mode json / TUI listener
       └─ session.appendMessage                       ③ 持久化

[退出] agent_end → disposeRuntime

整条链路白盒：从加载、就绪到运行、退出，每一步都能挂回调观察或拦截。

总结

回头看，pi 扩展系统的设计决策围绕几个目标：

加载无编译：基于 jiti 动态 import .ts 文件，扩展开发者写完代码就能跑，零构建成本。
能力可扩展：事件/工具/命令/快捷键/provider，无论是深入 Agent 推理循环，还是扩展 UI 交互入口，都有对应的接入点。
事件可拦截：session_before_* 支持 cancel，扩展能拦截关键决策。
修改可链式：专用 emit 让多扩展依次修改同一结果。
异常可隔离：handler 异常被 catch，不影响主流程。
会话替换可安全：assertActive + withSession 防 use-after-replace。
provider 可接入：扩展能注册完整 LLM provider，接入企业/自定义 API。

所有这些能力的根基，是 pi 本身的分层架构——每个模块职责单一、边界清晰，只处理自己关注的数据。

扩展是什么：一个 .ts 文件，导出一个 factory#

加载流程：两阶段初始化#

加载阶段：factory 执行，但 action 是 throwing stub#

throwing stub：防加载阶段误调 action#

就绪阶段：bindCore 替换 stub#

执行流程：loop emit → 扩展先于系统其他处理#

事件直接同步调扩展，没有中间缓冲#

事件翻译层#

runner.emit：分发 + 异常隔离 + 延迟 ctx#

链式 emit：让多个扩展依次修改同一结果#

createContext：延迟求值 + assertActive#

扩展能做什么：三类能力#

订阅事件（on）：观察/拦截 pi 内部流程#

注册工具（registerTool）：给 LLM 增加新工具#

注册命令/provider/快捷键：扩展 UI 入口和模型接入#

Context 三级权限：权限递增#

会话替换的安全：assertActive + withSession#

完整时序：从 pi -e my-ext.ts 到 handler 被调#

总结#