Lifecycle及其功能会通过诊断渠道发布重要操作的结构化事件，例如 RPC 调用、计划任务执行、作业队列警告和 MCP 连接。Agent 类还会发布状态变更、聊天、工作流和电子邮件相关事件。没有监听者时，发布不会产生开销。
在安装了 Lifecycle 的任何 Durable Object 上，以及 Agent 类中，事件的工作方式都相同。
每个事件包含以下字段：
{
type: "rpc", // what happened
source: "websockets", // which capability emitted it
agent: "MyAgent", // which Durable Object class emitted it
name: "user-123", // which instance (Durable Object name)
payload: { method: "getWeather" }, // details
timestamp: 1758005142787 // when (ms since epoch)
}agent是 Durable Object 类名，name是实例名称。source是发出事件的功能 ID；如果事件来自 Lifecycle 作业队列，则值为lifecycle。此字段会出现在 Lifecycle 直接发布的事件上。通过Agent类的observability属性发出的事件不包含source。
每个事件都会发送到一个渠道，该渠道由其 type 前缀决定：
| 通道 | 事件类型 | 描述 |
|---|---|---|
agents:state |
state:* |
状态同步事件 |
agents:rpc |
rpc, rpc:* |
RPC 方法调用与失败 |
agents:message |
message:*, tool:*, submission:*, action:* |
聊天消息、工具、Think 提交和 Think 操作的生命周期事件 |
agents:chat |
chat:*，但不包括 chat:transcript:* |
聊天轮次、请求、恢复、流停滞和上下文处理 |
agents:transcript |
transcript:*, chat:transcript:* |
转录修复事件 |
agents:fiber |
fiber:* |
持久 fiber 生命周期 |
agents:agent_tool |
agent_tool:* |
父代理和子代理工具的运行 |
agents:schedule |
schedule:*, queue:* |
已安排和已排队的回调 |
agents:workflow |
workflow:* |
Workflow 状态转换 |
agents:mcp |
mcp:* |
MCP 客户端操作 |
agents:email |
email:* |
邮件处理 |
agents:task |
task:* |
来自实验性 Tasks 功能的事件。本页面不涵盖 |
agents:stream |
stream:* |
来自实验性 Streams 功能的事件。本页面不涵盖 |
agents:channel |
channel:*, notice:* |
来自实验性 channels 模块的事件。本页面不涵盖 |
agents:lifecycle |
connect、disconnect、destroy、job:*、alarm:*，以及所有不匹配其他前缀的类型 |
连接、拆除、作业队列和自定义功能事件 |
agents:lifecycle 是回退渠道。类型与以上任何前缀均不匹配的事件（例如来自你自己的功能的事件）会发送到该渠道。
agents/observability 中的 subscribe() 提供对特定通道上事件的类型安全访问：
import { subscribe } from "agents/observability";
const unsub = subscribe("rpc", (event) => {
if (event.type === "rpc") {
console.log(`RPC call: ${event.payload.method}`);
}
if (event.type === "rpc:error") {
console.error(
`RPC failed: ${event.payload.method} — ${event.payload.error}`,
);
}
});
// Clean up when done
unsub();import { subscribe } from "agents/observability";
const unsub = subscribe("rpc", (event) => {
if (event.type === "rpc") {
console.log(`RPC call: ${event.payload.method}`);
}
if (event.type === "rpc:error") {
console.error(
`RPC failed: ${event.payload.method} — ${event.payload.error}`,
);
}
});
// Clean up when done
unsub();回调带有类型定义：event 会收窄为该渠道的事件参考中列出的事件类型。你自己的功能所发出的事件不属于这些类型。请使用原始 API 读取它们。
类型化辅助函数使用 camelCase 键，因此代理工具事件应使用 subscribe("agentTool", ...) 订阅。原始诊断渠道订阅者则使用发出的渠道名称 agents:agent_tool。
也可直接使用 Node.js API 订阅：
import { subscribe } from "node:diagnostics_channel";
subscribe("agents:schedule", (event) => {
console.log(event);
});import { subscribe } from "node:diagnostics_channel";
subscribe("agents:schedule", (event) => {
console.log(event);
});扩展 LifecycleCapability 的功能可以通过 this.lifecycle.events.emit(type, payload) 发布事件。Lifecycle 会添加 source、agent、name 和 timestamp 字段，并根据事件类型的前缀路由事件。前缀未知的类型会发送到 agents:lifecycle：
import { DurableObject } from "cloudflare:workers";
import { Lifecycle, LifecycleCapability } from "agents/lifecycle";
class Orders extends LifecycleCapability {
constructor() {
super("orders");
}
place(orderId) {
// Unknown prefix, so this lands on agents:lifecycle
this.lifecycle.events.emit("order:placed", { orderId });
}
}
export class Shop extends DurableObject {
orders = new Orders();
lifecycle = Lifecycle.install(this).use(this.orders);
}import { DurableObject } from "cloudflare:workers";
import { Lifecycle, LifecycleCapability } from "agents/lifecycle";
class Orders extends LifecycleCapability {
constructor() {
super("orders");
}
place(orderId: string) {
// Unknown prefix, so this lands on agents:lifecycle
this.lifecycle.events.emit("order:placed", { orderId });
}
}
export class Shop extends DurableObject<Env> {
readonly orders = new Orders();
readonly lifecycle = Lifecycle.install(this).use(this.orders);
}在原始渠道上按类型读取这些事件：
import { subscribe } from "node:diagnostics_channel";
subscribe("agents:lifecycle", (message) => {
const event = message;
if (event.type === "order:placed") {
console.log(event.source, event.payload);
}
});import { subscribe } from "node:diagnostics_channel";
subscribe("agents:lifecycle", (message) => {
const event = message as { type: string; source?: string; payload: unknown };
if (event.type === "order:placed") {
console.log(event.source, event.payload);
}
});事件采用尽力而为的方式处理。Lifecycle 启动期间发出的事件会暂存，并在启动完成后发布；如果启动失败，则会丢弃这些事件。有关功能 API 的其余内容，请参阅构建功能。
在生产环境中，所有诊断渠道消息都会自动转发到 Tail Workers。代理本身无需编写订阅代码。关联一个 Tail Worker，并从 event.diagnosticsChannelEvents 读取事件：
export default {
async tail(events) {
for (const event of events) {
for (const msg of event.diagnosticsChannelEvents) {
// msg.channel is "agents:rpc", "agents:lifecycle", etc.
// msg.message is the event
console.log(msg.timestamp, msg.channel, msg.message);
}
}
},
};export default {
async tail(events) {
for (const event of events) {
for (const msg of event.diagnosticsChannelEvents) {
// msg.channel is "agents:rpc", "agents:lifecycle", etc.
// msg.message is the event
console.log(msg.timestamp, msg.channel, msg.message);
}
}
},
};这样，你就可以在生产环境中获得结构化且可筛选的可观测性，同时不会增加代理热路径的开销。
Agent 类通过 observability 属性发送事件。默认实现 agents/observability 中的 genericObservability 会将每个事件发布到上述渠道。直接安装 Lifecycle 的 Durable Object 没有 observability 属性。其事件始终会发送到各渠道，你可以在订阅者中筛选。
若要替换 Agent 上的默认实现，请提供自己的 Observability 实现：
import { Agent } from "agents";
const myObservability = {
emit(event) {
// Send to your logging service, filter events, etc.
if (event.type === "rpc:error") {
console.error(event.payload.method, event.payload.error);
}
},
};
class MyAgent extends Agent {
observability = myObservability;
}import { Agent } from "agents";
import type { Observability } from "agents/observability";
const myObservability: Observability = {
emit(event) {
// Send to your logging service, filter events, etc.
if (event.type === "rpc:error") {
console.error(event.payload.method, event.payload.error);
}
},
};
class MyAgent extends Agent {
override observability = myObservability;
}将 observability 设为 undefined 可禁用所有事件发出：
import { Agent } from "agents";
class MyAgent extends Agent {
observability = undefined;
}import { Agent } from "agents";
class MyAgent extends Agent {
override observability = undefined;
}| 类型 | 负载 | 触发时机 |
|---|---|---|
rpc |
{ method, streaming? } |
客户端调用 WebSockets 功能 callables 上的方法，或调用某个 @callable 方法 |
rpc:error |
{ method, error } |
该方法抛出异常 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
state:update |
{} |
在 Agent 上调用 setState() |
这些事件跟踪聊天消息生命周期、客户端工具交互与 Think 的持久提交。
| 类型 | 负载 | 触发时机 |
|---|---|---|
message:request |
{} |
收到聊天消息 |
message:response |
{} |
聊天响应流完成 |
message:clear |
{} |
聊天历史被清除 |
message:cancel |
{ requestId } |
流式请求被取消 |
message:error |
{ error } |
聊天流失败 |
tool:result |
{ toolCallId, toolName } |
收到客户端工具结果 |
tool:approval |
{ toolCallId, approved } |
工具调用被批准或拒绝 |
submission:create |
{ submissionId } |
Think 提交被接受 |
submission:status |
{ submissionId, status } |
Think 提交状态变化 |
submission:error |
{ submissionId, error } |
Think 提交失败 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
chat:turn:start |
{ requestId, trigger, admission, continuation?, generation? } |
聊天轮次开始 |
chat:turn:finish |
{ requestId, trigger, admission, continuation?, generation?, status, durationMs, error? } |
聊天轮次结束 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
chat:request:failed |
{ requestId?, stage, messagesPersisted?, error } |
Think 聊天请求在解析、持久化、运行或流式传输时失败 |
chat:recovery:detected |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind } |
首次观察到被中断的聊天 fiber |
chat:recovery:attempt |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind } |
框架开始恢复尝试 |
chat:recovery:scheduled |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind } |
调度重试或续传回调 |
chat:recovery:completed |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind } |
恢复成功完成 |
chat:recovery:skipped |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } |
因对话已变化或不再可恢复而跳过恢复 |
chat:recovery:failed |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } |
恢复运行但失败 |
chat:recovery:exhausted |
{ incidentId, requestId, attempt, maxAttempts, recoveryKind, reason } |
恢复超过配置的尝试预算 |
chat:stream:stalled |
{ requestId, timeoutMs } |
不活动看门狗触发，因为在 chatStreamStallTimeoutMs 内没有到达流分块。该轮次会进入持久恢复 |
recoveryKind 为 "retry" 表示恢复重放未应答的用户轮次；为 "continue" 表示继续部分 assistant 轮次。
| 类型 | 负载 | 触发时机 |
|---|---|---|
chat:context:compacted |
{ reason, shortened, requestId?, attempt? } |
Think 压缩会话以处理上下文窗口溢出。reason 为 "proactive"（步骤前 contextOverflow.proactive 防护触发）或 "reactive"（溢出后 contextOverflow.reactive 触发）。shortened 表示压缩是否实际缩短历史——false 表示重试仍会溢出。见 上下文窗口溢出恢复。 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
chat:transcript:repaired |
{ requestId?, removedToolCalls, normalizedInputs, toolCallIds? } |
Think 在发送给提供商前修复已持久化转录。removedToolCalls 统计已修复的孤立工具调用；normalizedInputs 统计已修复的字符串化或缺失工具输入 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
fiber:run:started |
{ fiberId, fiberName, managed? } |
持久 fiber 启动 |
fiber:run:completed |
{ fiberId, fiberName, managed?, elapsedMs? } |
持久 fiber 完成 |
fiber:run:failed |
{ fiberId, fiberName, managed?, error, elapsedMs? } |
持久 fiber 抛出 |
fiber:run:interrupted |
{ fiberId, fiberName, managed?, recoveryReason, elapsedMs? } |
启动时发现中断的 fiber |
fiber:recovery:detected |
{ fiberId, fiberName, managed?, recoveryReason, elapsedMs? } |
恢复发现中断的 fiber |
fiber:recovery:attempt |
{ fiberId, fiberName, managed?, recoveryReason } |
恢复钩子启动 |
fiber:recovery:handled |
{ fiberId, fiberName, managed?, recoveryReason, status, elapsedMs? } |
恢复处理完成 |
fiber:recovery:skipped |
{ fiberId, fiberName, managed?, reason, elapsedMs? } |
恢复扫描跳过剩余工作 |
fiber:recovery:failed |
{ fiberId, fiberName, managed?, error, reason?, elapsedMs? } |
恢复钩子失败 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
agent_tool:recovery:begin |
{ runCount, totalTimeoutMs? } |
父恢复开始扫描过期的 agent-tool 运行 |
agent_tool:recovery:row |
{ runId, agentType, status, reason?, elapsedMs? } |
协调一条过期运行 |
agent_tool:recovery:deadline |
{ runId, agentType, elapsedMs? } |
检查行前总恢复截止时间已耗尽 |
agent_tool:recovery:complete |
{ runCount, elapsedMs? } |
父恢复完成扫描行 |
agent_tool:recovery:failed |
{ error } |
父恢复意外失败 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
schedule:create |
{ callback, id } |
创建调度 |
schedule:execute |
{ callback, id } |
已调度回调启动 |
schedule:cancel |
{ callback, id } |
取消调度 |
schedule:retry |
{ callback, id, attempt, maxAttempts } |
已调度回调重试 |
schedule:error |
{ callback, id, error, attempts } |
已调度回调耗尽重试后失败 |
schedule:duplicate_warning |
{ callback, count, type } |
非幂等调度可能重复工作 |
queue:create |
{ callback, id } |
任务入队 |
queue:retry |
{ callback, id, attempt, maxAttempts } |
已入队回调重试 |
queue:error |
{ callback, id, error, attempts } |
已入队回调耗尽重试后失败 |
这些事件来自 Agent 类：
| 类型 | 负载 | 触发时机 |
|---|---|---|
connect |
{ connectionId } |
建立 WebSocket 连接 |
disconnect |
{ connectionId, code, reason } |
WebSocket 连接关闭 |
destroy |
{} |
agent 被销毁 |
这些事件来自安装了 Lifecycle 的任意 Durable Object 的 Lifecycle 作业队列。它们的 source 为 lifecycle：
| 类型 | 负载 | 触发时机 |
|---|---|---|
job:slow_dispatch |
{ capability, fn, id, thresholdMs } |
作业的 onJob 分发运行时间超过其阈值。长时间分发会延迟对象上的其他所有作业 |
job:backlog_warning |
{ capability, count } |
某项功能在单个闹钟周期内有许多到期作业，通常是因为推送的一次性作业没有稳定 ID |
alarm:memory_limit_reset |
{ strikes, limit, sealed, error } |
闹钟触及内存限制。达到触发次数上限后，sealed 为 true |
| 类型 | 负载 | 触发时机 |
|---|---|---|
workflow:start |
{ workflowId, workflowName? } |
启动 Workflow 实例 |
workflow:event |
{ workflowId, eventType? } |
向 Workflow 发送事件 |
workflow:approved |
{ workflowId, reason? } |
批准 Workflow |
workflow:rejected |
{ workflowId, reason? } |
拒绝 Workflow |
workflow:terminated |
{ workflowId, workflowName? } |
终止 Workflow |
workflow:paused |
{ workflowId, workflowName? } |
暂停 Workflow |
workflow:resumed |
{ workflowId, workflowName? } |
恢复 Workflow |
workflow:restarted |
{ workflowId, workflowName? } |
重启 Workflow |
| 类型 | 负载 | 触发时机 |
|---|---|---|
mcp:client:preconnect |
{ serverId } |
连接 MCP 服务器之前 |
mcp:client:connect |
{ url, transport, state, error? } |
MCP 连接尝试完成或失败 |
mcp:client:authorize |
{ serverId, authUrl, clientId? } |
MCP OAuth 流程开始 |
mcp:client:discover |
{ url?, state?, error?, capability? } |
MCP 能力发现成功或失败 |
mcp:client:close |
{ url, transport?, state, error?, phase? } |
MCP 连接已关闭 |
| 类型 | 负载 | 触发时机 |
|---|---|---|
email:receive |
{ from, to, subject? } |
收到邮件 |
email:reply |
{ from, to, subject? } |
发送回复邮件 |
email:send |
{ from, to, subject? } |
发送邮件 |
追踪
使用 Workers traces 追踪模型调用、工具运行与审批。
Tail Workers
将 diagnostics channel 事件转发到 Tail Worker 以进行生产监控。