切换会话与重新订阅

宿主里有个很常见的结构：一个侧边栏列出所有会话，点一下换到另一个，右边的对话区跟着刷新。

在 Pi 的 SDK 里，这个"换一下"不是改个变量就完事。AgentSessionRuntime 的替换操作会销毁旧的 AgentSession，创建一个全新的。旧对象上挂的订阅跟着旧对象一起没了——旧会话不会继续向新视图写消息，但你必须在新会话上重新订阅，否则新视图什么也收不到。

这两件事是同一个陷阱的两面。这篇把订阅归属讲清楚，并把四个替换操作的行为差异列全。

读它需要会用 TypeScript，并了解 Pi 的会话与扩展绑定机制（bindExtensions() 什么时候触发、订阅挂在哪个对象上）。Pi 迭代很快，动手前先记录 pi --version；本文依据 Pi 官方文档 与 官方仓库 编写。

一、契约

二、四个替换操作

AgentSessionRuntime 提供四个方法。它们共享同一条流程：先拆掉当前运行时，再建一个新的，然后应用。

官方 SDK 文档的原话：每次操作都会替换活动的 AgentSession，并为目标工作目录重建服务。

替换流程拆解

源码里这条路径是这么走的：

1. 发 session_before_switch（或 session_before_fork）事件，允许扩展取消。
2. teardownCurrent()：先 session.abort() 让当前响应结束（这样被中止的那轮，包括工具结果，会写进即将退出的会话），再发 session_shutdown，最后 session.dispose()。
3. createRuntime()：为目标 cwd 重建服务，创建新的 AgentSession。
4. apply()：把 _session 和 _services 换成新的。
5. finishSessionReplacement()：调用 rebindSession 回调（如果设置过），再调用 withSession 回调。

第 2 步的 session.dispose() 就是订阅消失的原因——它会移除所有事件监听。

三、核心陷阱：订阅属于旧会话

先看官方示例的核心模式（runtime 的创建见下一节）：

// 重新订阅的唯一模式：先退订旧的，再订阅新的
let unsubscribe: (() => void) | undefined;

async function bindSession() {
  unsubscribe?.();                              // 先退订旧的
  const session = runtime.session;              // 再取新的
  await session.bindExtensions({});             // 重新绑定扩展
  unsubscribe = session.subscribe((event) => {  // 在新会话上重新订阅
    if (event.type === "queue_update") {
      console.log("Queued:", event.steering.length + event.followUp.length);
    }
  });
  return session;
}

let session = await bindSession();
// newSession() 会销毁旧 AgentSession 并建一个全新的
await runtime.newSession();
// 必须重新绑定：旧订阅属于旧对象，不重绑就收不到新会话的任何事件
session = await bindSession();

三个要点：

1. unsubscribe?.() 必须在订阅新会话之前调用。 否则旧监听还挂着，你会在同一个回调里收到两个会话的事件。
2. session.bindExtensions({}) 也要重新调。 扩展绑定是会话级的，不是运行时级的。MCP 扩展在 session_start 连服务器——不重新绑定，新会话不会连接 MCP。
3. runtime.session 每次读都是当前会话。 不要把旧 session 引用缓存到别处。

反例：订阅没重新绑定会怎样

// 反例：只在启动时订阅一次
const unsubscribe = runtime.session.subscribe((event) => {
  console.log("event:", event.type);
});

await runtime.newSession();
// 切换之后：旧会话被 dispose，监听被移除。
// 新会话上没有任何监听，回调再也不会被调用。

表现出来就是"换了会话之后界面就不更新了"。反过来，如果你在替换时没有退订旧监听（假设 Pi 还没 dispose 掉它），你会看到旧会话的事件继续往新视图里写——消息串到错误的会话里。

正确的模式是先退订旧的，再订阅新的。

四、完整示例

下面这份代码可以跑，它覆盖了切换、新建、分叉和导入四种情况，每一步都重新绑定：

import {
  type CreateAgentSessionRuntimeFactory,
  createAgentSessionFromServices,
  createAgentSessionRuntime,
  createAgentSessionServices,
  getAgentDir,
  SessionManager,
} from "@earendil-works/pi-coding-agent";

const cwd = process.cwd();

// 运行时工厂：为目标 cwd 重建服务
const createRuntime: CreateAgentSessionRuntimeFactory = async ({
  cwd,
  sessionManager,
  sessionStartEvent,
}) => {
  const services = await createAgentSessionServices({ cwd });
  return {
    ...(await createAgentSessionFromServices({
      services,
      sessionManager,
      sessionStartEvent,
    })),
    services,
    diagnostics: services.diagnostics,
  };
};

const runtime = await createAgentSessionRuntime(createRuntime, {
  cwd,
  agentDir: getAgentDir(),
  sessionManager: SessionManager.create(cwd),
});

// 当前订阅的归属。变量名里带上 sessionId 是为了读代码时能想起这件事。
let unsubscribe: (() => void) | undefined;
let boundSessionId: string | undefined;

async function rebind(label: string): Promise<void> {
  // 1. 先退订旧会话的监听
  unsubscribe?.();
  unsubscribe = undefined;

  // 2. 取当前活动会话
  const session = runtime.session;

  // 3. 重新绑定扩展（这一步会触发 session_start，MCP 等扩展在这里连接服务器）
  await session.bindExtensions({});

  // 4. 在新会话上重新订阅
  unsubscribe = session.subscribe((event) => {
    if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
      process.stdout.write(event.assistantMessageEvent.delta);
    }
  });

  boundSessionId = session.sessionId;
  console.log(`[${label}] bound to session ${boundSessionId}`);
}

try {
  await rebind("startup");
  const firstSessionFile = runtime.session.sessionFile;
  console.log("initial session file:", firstSessionFile);

  // 新会话：替换活动会话，然后重新绑定
  await runtime.newSession();
  await rebind("after newSession");
  console.log("after newSession:", runtime.session.sessionFile);

  // 切回原会话
  if (firstSessionFile) {
    await runtime.switchSession(firstSessionFile);
    await rebind("after switchSession");
    console.log("after switchSession:", runtime.session.sessionFile);
  }
} finally {
  // 退订并释放整个运行时
  unsubscribe?.();
  await runtime.dispose();
}

注意 runtime.dispose() 会发一次 session_shutdown（reason 为 "quit"）并 dispose 当前会话——这和替换时的清理是同一条路径。

五、也可以让运行时替你重新绑定

AgentSessionRuntime 有一个 setRebindSession() 方法。设置它之后，运行时在每次替换完成后会自动调用它：

// 运行时自动重新绑定。注意退订仍在订阅之前
runtime.setRebindSession(async (session) => {
  await session.bindExtensions({});
  unsubscribe?.();
  unsubscribe = session.subscribe(handler);
});

这样宿主代码里就不用在每个替换点后面手动调 rebind()。

但两种方式只能选一种。 如果既设了 setRebindSession 又在替换后手动 rebind()，订阅会被绑定两次——一次在运行时回调里，一次在你自己的代码里。表现就是事件处理执行两遍。

选 setRebindSession 的代价是它只有一个回调槽位，多个视图要各自管理自己的订阅；选手动 rebind() 的代价是每个替换点都要记得调。

六、各操作的行为差异

importFromJsonl() 有两个值得注意的行为：

• 文件不存在时抛 SessionImportFileNotFoundError。
• 导入的会话 cwd 解析不出来且没有提供 cwdOverride 时，抛 MissingSessionCwdError。

另外，导入的会话会以 reason: "resume" 触发 session_start——不是 "new"。扩展如果要区分场景，注意 session_start 事件的 reason 字段有 "startup" | "reload" | "new" | "resume" | "fork" 五个取值。

fork() 的 position 参数决定分叉点："before"（默认）从选中条目的父节点分出，"at" 从选中条目本身分出。"before" 模式下选中的条目必须是用户消息，否则抛 "Invalid entry ID for forking"。

持久化会话分叉时还要求会话文件已经存在，否则抛 "This session has not been saved yet. Send a message before cloning or forking it."。内存会话则在内存里分叉，不受这个限制。

七、官方已定义、源码事实与教学设计

官方已定义的行为：

• AgentSessionRuntime 提供 newSession()、switchSession()、fork()、importFromJsonl()。
• 每次操作替换活动的 AgentSession 并为目标工作目录重建服务。
• 替换后订阅属于旧的 AgentSession，必须重新绑定。
• session.dispose() 会中止活动工作、使扩展上下文失效、断开 agent、移除事件监听。
• 扩展命令上下文里的会话替换操作（newSession、fork、switchSession）会让旧上下文失效——切换前只捕获纯数据，切换后用 withSession 给的新上下文做会话绑定的工作。

源码事实但官方文档没有单独成段说明的：

• 替换流程的精确顺序（abort → session_shutdown → dispose → 建新运行时 → apply → rebindSession → withSession）。
• setRebindSession() 这个回调槽位。
• importFromJsonl() 复制文件时的数字后缀规则。
• session.dispose() 本身不发出 session_shutdown；发出它的是 teardownCurrent() 和 runtime.dispose()。这条差别决定了 MCP 连接在什么时机被关闭。

教学设计：

• 用 bindSession() 这样一个命名的函数统一处理"退订 + 重新绑定"，避免散落在各个替换点。
• 在订阅变量旁标注它属于哪个会话（用 sessionId 而不用对象引用）。
• 用 try/finally 保证 runtime.dispose() 一定执行。

你现在能做什么

• 写一个运行时宿主的 rebind() 函数，封装"退订 → 取当前会话 → bindExtensions() → 订阅"，在每个替换点后调用它。
• 用 runtime.setRebindSession() 替换掉手动调用，然后故意两个都用一次，观察事件处理跑两遍。
• 在 rebind() 里打印 session.sessionId，连续做 newSession() 和 switchSession()，验证每次的 ID 都变了。
• 故意漏掉 bindExtensions()，在切换后观察 MCP 工具消失，再补回来。
• 用 switchSession() 切到一个 cwd 不同的会话，验证工具的路径基准跟着目标会话走。
• 用 fork(entryId, { position: "at" }) 从一条非用户消息分叉，捕获 "Invalid entry ID for forking"，再换成合法的条目 ID 成功。
• 用 importFromJsonl() 导入一个不存在的文件，捕获 SessionImportFileNotFoundError；再用 cwdOverride 兜住 MissingSessionCwdError。

排障速查

参考资料

• 官方文档：SDK
• 官方文档：会话与上下文
• 官方文档：扩展
• 官方仓库
• SDK 示例：会话运行时
• 会话运行时实现