在任务执行前建立监听，用对的事件判断一次运行是否真的结束，并在结束后把监听和会话一起交还。

订阅事件与释放会话

智能体的运行不是"调一次函数等它返回"。它是一串事件：模型开始说话、工具开始执行、工具结束、这一轮结束、这次运行结束。

作为宿主，你在这条链上只做两件事：在正确的时机挂上监听，在正确的时机摘掉它。

这两件事都有明确的规则，而且规则不复杂。复杂的是几个看起来都像"结束"的事件，含义完全不同。用错了，你会以为任务已经完成，实际上自动重试还在后面。

读完这篇你能有一张事件类型对照表、一份判断"真的结束了"的判据、一份完整的订阅与释放模板，并能让同一个会话连跑多次而不重复收到事件。

读它需要会用 TypeScript，并且已经跑通过一次 createAgentSession() 调用。你还要接受一个前提：监听器是会话的属性，不是某次运行的属性。

最终要跑通什么

在 prompt() 之前装上监听，在 finally 里退订并释放会话。
用 agent_settled 而不是 agent_end 判断"Pi 不会自动继续了"。
用 message_end 里的消息作为权威的最终消息，而不是累加 message_update 的增量。
同一个会话连跑两次，监听器只注册一次，事件不重复。

第 4 点是完成标准。它听起来平淡，但它是订阅机制最容易出错的地方。

输入、输出与责任边界

先把边界定下来，后面所有判断都基于它。

维度	内容
输入	一个 `AgentSession`；一个监听回调
输出	一串会话事件；一个 `unsubscribe` 函数
状态	监听器列表（属于会话，不是属于某次运行）
时间点	订阅必须在 `prompt()` 之前；释放必须在会话结束之后

监听器属于会话，不属于某次运行。 这是本篇所有结论的根源。一次 prompt() 触发一次运行，运行结束不代表监听器失效——监听器还在，下次 prompt() 照样会收到事件。

责任边界也很清楚：

职责	由谁承担
产生事件	Pi（agent 循环与会话层）
决定关心哪些事件	宿主
判断运行是否真的结束	宿主（依据事件语义）
摘掉监听器	宿主（`unsubscribe()` 或 `dispose()`）

事件类型

会话事件分两层。核心层来自 Agent Core，扩展层是会话自己加的。

事件	含义	关键字段
`agent_start`	一次底层运行开始	—
`turn_start`	一个回合开始（一次模型调用加它的工具执行）	—
`message_start`	任意消息开始（user / assistant / toolResult）	`message`
`message_update`	仅助手消息，流式增量	`message`、`assistantMessageEvent`
`message_end`	消息完成	`message`
`tool_execution_start`	工具开始执行	`toolCallId`、`toolName`、`args`
`tool_execution_update`	工具上报中间进度	`toolCallId`、`partialResult`
`tool_execution_end`	工具执行结束	`toolCallId`、`result`、`isError`、`durationMs`
`turn_end`	回合结束	`message`、`toolResults`
`agent_end`	一次底层运行结束	`messages`、`willRetry`
`agent_settled`	Pi 不会自动继续了	`aborted`

会话层额外提供这些：

事件	含义
`queue_update`	排队消息变化，含 `steering` 和 `followUp`
`compaction_start` / `compaction_end`	上下文压缩开始/结束
`entry_appended`	会话条目被追加（持久化）
`session_info_changed`	会话名变化
`thinking_level_changed`	思考级别变化
`auto_retry_start` / `auto_retry_end`	自动重试
`bash_execution_update`	Bash 命令输出增量

一次运行的典型序列

不带工具调用的一次 prompt()：

text

# 不带工具调用时一次 prompt() 的完整会话事件序列，缩进表示嵌套层级
agent_start
  turn_start
    message_start   { message: userMessage }
    message_end     { message: userMessage }
    message_start   { message: assistantMessage }
    message_update  { assistantMessageEvent: text_delta }   ← 流式增量，可能很多次
    message_end     { message: assistantMessage }
  turn_end          { message, toolResults: [] }
agent_end           { messages: [...] }
agent_settled       { aborted: false }

带工具调用时，工具事件插在助手消息结束之后、工具结果消息之前：

text

  message_end       { message: assistantMessage with toolCall }
  tool_execution_start  { toolCallId, toolName, args }
  tool_execution_update { partialResult }        ← 工具流式上报时才有
  tool_execution_end    { toolCallId, result, isError }
  message_start/end { message: toolResultMessage }

三个"结束"的区别

这是本篇最重要的一段。

事件	它说的是	它没说的是
`turn_end`	一个回合（一次模型调用加它的工具）结束了	运行结束了
`agent_end`	一次底层 agent 运行结束了	Pi 不会自动继续了
`agent_settled`	Pi 不会自动继续了	—

官方 SDK 文档的原话是：message_end 包含权威的完成消息。agent_end 标记一次底层 agent 运行的结束，但自动恢复或排队的工作仍可能跟在后面。当宿主需要知道 Pi 不会自动继续时，用 agent_settled。

这条区别有实际后果。agent_end 之后可能发生的事：

自动重试。 遇到可重试的错误，会话会自己再试一次。会话层的 agent_end 事件带着 willRetry 字段，直接告诉你后面还有没有。
排队的工作。 运行期间排进队列的消息（引导或后续）会在运行结束后触发新的运行。
压缩与分支摘要。 这些也可能在 agent_end 之后发生。

所以：判断"任务真正完成"用 agent_settled；判断"这一次底层运行结束"用 agent_end；两者都需要就两个都监听。

typescript

// 两个结束事件要分开判断：agent_end 只说明一次底层运行结束，agent_settled 才说明 Pi 不会再自动继续
const unsubscribe = session.subscribe((event) => {
	if (event.type === "agent_end") {
		// willRetry 直接告诉你后面还有没有自动重试
		console.log("一次运行结束", event.willRetry ? "（后面还会重试）" : "");
	}
	if (event.type === "agent_settled") {
		// aborted 为 true 表示是被中止而不是自然结束
		console.log("Pi 不会自动继续了", event.aborted ? "（被中止）" : "");
	}
});

监听器是 await 的

Agent Core 的 subscribe() 接收的监听器可以返回 Promise，它们按注册顺序被 await。这一点很重要，它决定了你的监听器能不能当屏障用。

typescript

// 核心层：监听器返回 Promise 会被等待
const unsubscribe = agent.subscribe(async (event, signal) => {
	if (event.type === "agent_end") {
		// 运行结束前的最后一道收尾工作，这里可以 await
		await flushSessionState(signal);
	}
});

官方文档明确写了两点：agent_end 是这次运行的最后一个事件，但被 await 的 agent_end 监听器仍然计入运行结束。await agent.waitForIdle() 和 await agent.prompt(...) 只有在那些被 await 的 agent_end 监听器完成后才 settle。

这意味着你可以把"落盘最终状态"这类收尾工作放在 agent_end 监听器里，它是运行的一部分，不是运行之后的补救。

一个必须注意的推论：你的监听器慢，整个运行就慢。 在监听器里做同步的重活（读大文件、等待网络）会直接拖慢模型响应的呈现。流式渲染这类轻量操作放监听器里没问题，重活应该放到 content 之外的通道去。

订阅要早于 prompt

session.subscribe() 只是往监听器列表里加一个函数。事件在运行过程中产生，产生时如果你的监听器不在列表里，就永远收不到。

typescript

// 反例：订阅晚了一步，快速完成的运行会被整个错过
await session.prompt("总结这个文件");
const unsubscribe = session.subscribe((event) => {
	// 前面的运行已经结束，这里什么都收不到
});

正确的顺序永远是先订阅再动作：

typescript

// 先订阅：事件在运行中产生，错过就永远收不回来
const unsubscribe = session.subscribe((event) => {
	// message_update 只带增量，要最终文本要去看 message_end 的 message
	if (event.type === "message_update" && event.assistantMessageEvent.type === "text_delta") {
		process.stdout.write(event.assistantMessageEvent.delta);
	}
});

try {
	await session.prompt("总结这个文件");
} finally {
	// 退订放在 finally：无论成功失败都不留悬挂的监听器
	unsubscribe();
}

完整的订阅与释放模板

typescript

// run.ts
import { createAgentSession, SessionManager } from "@earendil-works/pi-coding-agent";

const { session } = await createAgentSession({
	sessionManager: SessionManager.inMemory(),
});

// 订阅早于任何会触发运行的动作
const unsubscribe = session.subscribe((event) => {
	switch (event.type) {
		case "message_update":
			// message_update 只带增量，最终文本要去看 message_end 的 message
			if (event.assistantMessageEvent.type === "text_delta") {
				process.stdout.write(event.assistantMessageEvent.delta);
			}
			break;
		case "tool_execution_start":
			console.error(`[工具] ${event.toolName}`);
			break;
		case "tool_execution_end":
			console.error(`[工具结束] ${event.toolName} isError=${event.isError}`);
			break;
		case "agent_end":
			// willRetry 为 true 表示后面还有自动重试，这次运行还没真正结束
			console.error(`[运行结束] willRetry=${event.willRetry}`);
			break;
		case "agent_settled":
			// 只有这个事件能告诉你「Pi 不会自动继续了」
			console.error(`[已落定] aborted=${event.aborted}`);
			break;
	}
});

try {
	await session.prompt("用三句话说明这个仓库的结构。");
	console.log();
	console.log(session.getLastAssistantText());
} finally {
	// 退订和释放都放在 finally：无论成功失败都不留悬挂的监听器
	unsubscribe();
	session.dispose();
}

三段结构，每一段都有原因：

代码	为什么
`subscribe()` 在 `try` 之前	运行前就位，不漏事件
`unsubscribe()` 在 `finally` 里	无论成功失败都退订，不留悬挂的监听器
`dispose()` 在 `finally` 里	中止进行中的工作、失效扩展上下文、与 agent 断开、移除全部监听器

重复运行不会重复收到事件

这是本篇的完成标准，也是订阅机制最容易出错的地方。

正确的写法：订阅一次，跑多次。

typescript

import { createAgentSession, SessionManager } from "@earendil-works/pi-coding-agent";

const { session } = await createAgentSession({
	// 用内存会话：练习不写磁盘文件
	sessionManager: SessionManager.inMemory(),
});

const runs: { aborted: boolean }[] = [];

// 只订阅一次：监听器属于会话，不会因为跑了一次就失效
const unsubscribe = session.subscribe((event) => {
	// agent_settled 一次运行落定一次，用来数运行次数
	if (event.type === "agent_settled") {
		runs.push({ aborted: event.aborted });
	}
});

try {
	// 同一个会话连跑三次，监听器不重复注册，事件也不重复
	await session.prompt("第一件事");
	await session.prompt("第二件事");
	await session.prompt("第三件事");
	// runs.length === 3，一次运行一次落定
} finally {
	// 退订和释放都放在 finally，成功失败都要收尾
	unsubscribe();
	session.dispose();
}

监听器列表属于会话。第二次 prompt() 复用同一个监听器，收到的是第二次运行的事件，不是第一次的重播。所以"跑三次"收到的是三组事件，不是九组。

错误的写法：每次运行前都订阅一次，从不退订。

typescript

// 反例：监听器不断累积
for (const task of ["第一件事", "第二件事", "第三件事"]) {
	session.subscribe((event) => {
		if (event.type === "agent_settled") {
			console.log("完成一个任务");
		}
	});
	await session.prompt(task);
}
// 第一次运行：1 个监听器 → 收到 1 次
// 第二次运行：2 个监听器 → 收到 2 次
// 第三次运行：3 个监听器 → 收到 3 次

这个反例的输出是 1、2、3 次"完成一个任务"。看起来像"每次运行收到了多次事件"，其实是监听器在累积。 而且退订函数只解决最后一个订阅——前面几次订阅的退订函数被丢掉了，谁也摘不掉它们。

还有一个更隐蔽的变体：把订阅放在循环里但把退订函数收集起来是对的，前提是退订和订阅一一对应。如果订阅了三次却只退订一次（因为只保存了最后一个返回值），剩下两个监听器会一直留着。

第三条规则：dispose() 会清空整个监听器列表。所以会话用完之后直接 dispose()，不需要逐个退订。但在会话存活期间，只想摘掉自己那一个，就用 unsubscribe()。

验证方法很简单：订阅一次，跑三次，断言落定事件的条数等于三。

谁负责什么

职责	由谁承担	说明
产生事件	Pi	agent 循环与会话层
决定订阅哪些事件	宿主	监听器里按 `event.type` 分支
判断运行是否真的结束	宿主	依据 `agent_settled`，不是 `agent_end`
保证不漏事件	宿主	订阅早于动作
保证不重复收到	宿主	订阅一次，退订对应
释放监听与会话	宿主	`unsubscribe()` + `dispose()`

你现在能做什么

写出一份订阅模板：subscribe() 在 try 之前，unsubscribe() 和 dispose() 都在 finally 里。
用 agent_settled 判断"Pi 不会自动继续了"，并单独监听 agent_end.willRetry 知道后面还有没有重试。
用 message_end 的消息作为最终内容，而不是累加 message_update 的增量。
让同一个会话连跑三次，断言落定事件恰好收到三次。
在监听器里做一件 await 的收尾工作，验证它被计入这次运行（prompt() 的 resolve 会等它完成）。

排障速查

现象	检查点	处理
一次事件都没收到	订阅时机	`subscribe()` 必须在 `prompt()` 之前。快速完成的运行不会等你订阅
收到的消息是残缺的	该用哪个事件	`message_update` 是增量，`message_end` 才是完整消息。要最终内容用 `message_end`
收到 `agent_end` 后任务还没完	事件语义	`agent_end` 之后可能还有自动重试或排队工作。用 `agent_settled` 判断真的结束，或看 `agent_end.willRetry`
同一件事收到多次	监听器累积	检查是不是在循环里反复 `subscribe()`。订阅一次，用 `unsubscribe()` 摘掉
换了会话后收不到事件	重新订阅	运行时替换会话后，旧会话上的订阅属于旧对象，必须重新绑定
界面卡顿，模型输出变慢	监听器耗时	监听器是 await 的，重活会拖慢整个运行
释放后仍在打日志	释放顺序	`dispose()` 之后监听器被移除而不是被调用。确认没有别的代码路径在写
想在运行中追加消息	队列语义	流式输出时 `prompt()` 必须指定 `streamingBehavior`，否则抛错。用 `steer()` / `followUp()`
退订没摘干净	退订与订阅配对	每次 `subscribe()` 都要保留对应的 `unsubscribe()`，别只留最后一个

Agent 入门知识10

智能体开发94

开发入门7

模型接入6

结构化输出5

运行机制与控制14

工具与外部接入11

会话与上下文8

工作流与编排6

Skills 开发6

扩展与插件开发10

多智能体协作5

调试与可观测性8

评估与调优8

桌面 Agent 项目实践176

订阅事件与释放会话

订阅事件与释放会话

最终要跑通什么

输入、输出与责任边界

事件类型

一次运行的典型序列

三个"结束"的区别

监听器是 await 的

订阅要早于 prompt

完整的订阅与释放模板

重复运行不会重复收到事件

谁负责什么

你现在能做什么

排障速查

参考资料