作业和 alarm

一个 Durable Object 只有一个 alarm。Lifecycle 拥有该 alarm，并根据持久作业队列设置它。该队列是对象 SQLite 数据库中的 cf_agents_jobs 表。Scheduler 和 PiHarness 将未来的工作存储为作业，因此它们可以共享同一个 alarm 而不会互相覆盖。

一个作业会指定函数、到期时间和可选的有效负载。作业到期时，alarm 会唤醒对象，Lifecycle 随后调用作业所有者的 onJob() 钩子。所有者可以是宿主，也可以是推送该作业的能力。

从宿主推送作业

通过 lifecycle.jobs 推送作业，并在宿主的 onJob() 钩子中处理：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

export class CleanupAgent extends DurableObject {
	lifecycle = Lifecycle.install(this);

	onStart() {
		this.ctx.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)",
		);
	}

	async onRequest() {
		await this.lifecycle.jobs.push({
			id: "cleanup",
			fn: "deleteExpired",
			time: Date.now() + 60_000,
			retry: { maxAttempts: 5 },
		});
		return new Response("Cleanup scheduled", { status: 202 });
	}

	onJob({ job }) {
		if (job.fn !== "deleteExpired") return;
		this.ctx.storage.sql.exec(
			"DELETE FROM sessions WHERE expires_at < ?",
			Date.now(),
		);
	}
}

import { DurableObject } from "cloudflare:workers";
import {
	Lifecycle,
	type LifecycleJobContext,
	type LifecycleJobOutcome,
} from "agents/lifecycle";

export class CleanupAgent extends DurableObject<Env> {
	readonly lifecycle = Lifecycle.install(this);

	onStart(): void {
		this.ctx.storage.sql.exec(
			"CREATE TABLE IF NOT EXISTS sessions (id TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)",
		);
	}

	async onRequest(): Promise<Response> {
		await this.lifecycle.jobs.push({
			id: "cleanup",
			fn: "deleteExpired",
			time: Date.now() + 60_000,
			retry: { maxAttempts: 5 },
		});
		return new Response("Cleanup scheduled", { status: 202 });
	}

	onJob({ job }: LifecycleJobContext): LifecycleJobOutcome {
		if (job.fn !== "deleteExpired") return;
		this.ctx.storage.sql.exec(
			"DELETE FROM sessions WHERE expires_at < ?",
			Date.now(),
		);
	}
}

push() 解析完成后，作业即已持久化。如果对象被驱逐，作业到期时 alarm 会将其唤醒。能力以相同方式通过 this.lifecycle.jobs 推送作业。有关详细信息，请参阅构建能力。

推送选项

push() 接受以下选项，并解析为已存储的 LifecycleJob：

选项	类型	说明
`fn`	`string`	必填。你的 `onJob()` 据以分派的函数名称。
`time`	`number`	必填。以 Unix 毫秒表示的到期时间。过去的时间会在下一次 alarm 时运行。
`payload`	`unknown`	可 JSON 序列化的值，作为 `job.payload` 传递给 `onJob()`。
`id`	`string`	稳定的作业 ID。推送现有 ID 会替换该作业。省略 ID 时，Lifecycle 会生成一个 ID。
`retry`	`RetryOptions`	失败运行的重试策略。有关详细信息，请参阅重试和失败。
`singleflight`	`boolean`	当该作业之前的一次运行仍在进行时跳过此作业。
`hungTimeoutSeconds`	`number`	正在进行的单次运行在多少秒后视为挂起，以及缓慢运行在多少秒后记录警告。默认为 `30`。
`exclusive`	`boolean`	当此作业处于待处理状态时，只为它设置 alarm，防止其他作业唤醒对象。`Agent` 类将此选项用于延迟销毁。
`recoveryLoop`	`boolean`	将作业标记为可能耗尽内存的恢复循环的一部分。有关详细信息，请参阅内存限制保护。

如果多次推送同一作业，请为其提供稳定的 id，这样新的推送会替换待处理作业，而不是再添加一个。当同一所有者的 10 个或更多到期作业在一次 alarm 中运行时，Lifecycle 会记录警告并发出 job:backlog_warning 事件，因为这通常意味着作业在没有稳定 ID 的情况下被重复推送。

作业 ID 共用一个表。每个所有者只能读取、替换或取消自己的作业。推送已由其他所有者使用的 ID 会抛出错误。

管理作业

lifecycle.jobs 是一个 LifecycleJobs 对象。能力的 this.lifecycle.jobs 具有相同的方法，但作用域限定为该能力：

方法	返回值	说明
`push(options)`	`Promise<LifecycleJob>`	推送或替换一个作业。
`cancel(id)`	`Promise<boolean>`	删除一个作业。没有匹配作业时解析为 `false`。
`reschedule(id, time)`	`Promise<boolean>`	更改一个作业的到期时间。没有匹配作业时解析为 `false`。
`get(id)`	`LifecycleJob \| undefined`	读取一个作业。
`list()`	`LifecycleJob[]`	列出所有者的作业，并按到期时间排序。
`rearm()`	`Promise<void>`	从队列重新设置 alarm，而不更改队列。

对队列的每次更改都会重新设置 alarm，因此你很少需要调用 rearm()。可使用它为已存在的作业恢复丢失的 alarm。例如，Scheduler 会在幂等的 set() 找到现有计划且未推送任何内容后调用它。

`LifecycleJob` 的结构

onJob() 以 context.job 的形式接收作业：

字段	类型	说明
`id`	`string`	作业 ID。在重新调度时保持不变。
`capability`	`string`	所属能力的 ID；宿主作业则为 `"host"`。
`fn`	`string`	函数名称。
`time`	`number`	以 Unix 毫秒表示的到期时间。
`payload`	`unknown`	经过一次 JSON 往返后的有效负载。
`retry`	`RetryOptions \| undefined`	推送时提供的重试策略。
`singleflight`	`boolean`	前一次运行仍在进行时是否跳过该作业。
`exclusive`	`boolean`	当作业处于待处理状态时，是否只为该作业设置 alarm。
`recoveryLoop`	`boolean`	是否由内存限制断路器管理该作业。
`createdAt`	`number`	以 Unix 秒表示的创建时间。

context.attempt 是当前 alarm 中的尝试次数，从 1 开始。

控制作业运行后的行为

onJob() 返回的值（即其 LifecycleJobOutcome）决定了作业接下来会发生什么：

返回值	结果
无	作业完成，Lifecycle 将其删除。
`{ rescheduleAt: time }`	作业保留在队列中，直到 `time`。
`"yield"`	作业保持到期状态，对象会立即再次唤醒。

对于重复工作（例如每五分钟轮询一次），返回 { rescheduleAt }。返回 "yield" 可将长作业拆分为更小的部分。

如果代码在作业运行期间推送或重新调度同一作业 ID，较新的更改优先。Lifecycle 会丢弃正在进行的那次运行的返回值。

重试和失败

当 onJob() 抛出错误时，Lifecycle 会在同一次 alarm 中使用指数退避进行重试，最多重试到 retry.maxAttempts。RetryOptions 包含以下字段：

字段	默认值	说明
`maxAttempts`	`3`	尝试次数，包括第一次
`baseDelayMs`	`100`	退避的基础延迟
`maxDelayMs`	`3000`	两次尝试之间的最大延迟

最后一次尝试失败后：

能力的 onJobError(context, error) 钩子决定接下来的行为。其返回值的作用与 onJob() 结果相同。如果没有此钩子，作业将完成。
宿主作业会完成。宿主没有 onJobError() 钩子，因此应从持久状态派生宿主作业，并在需要时重新推送。

平台故障不会消耗尝试次数。当部署替换隔离环境或对象达到内存限制时，Lifecycle 会保留该作业，并让新的调用运行它。

如果作业的所有者不再安装，或其所有者没有 onJob() 钩子，该作业会被记录并删除。

Alarm 的运行方式

Alarm 触发时，Lifecycle 会：

当队列中仍有作业时，会设置一个备用闹钟，时间为下一个作业的执行时间，但至少延后 30 秒。如果启动过程抛出异常或隔离环境在启动期间退出，此闹钟会重试启动。
如果对象处于冷启动状态，则运行启动流程。
如果有到期作业，则设置一个 30 秒后触发的备用闹钟。如果隔离环境在作业运行期间退出，此闹钟会唤醒对象以继续处理。
按到期时间顺序逐个运行所有到期作业。
运行主机的 onAlarm()。
为下一个到期作业设置闹钟。

如果启动流程推送或重新安排作业后抛出异常，无论异常发生在闹钟还是请求期间，也会设置备用闹钟，以确保这些作业仍能唤醒对象。

请勿在使用 Lifecycle 的对象上调用 ctx.storage.setAlarm()，否则会替换 Lifecycle 管理的闹钟。请改为推送作业。lifecycle.rearmAlarm() 会按需根据队列设置闹钟，而 lifecycle.disableAlarms() 会在拆除期间永久关闭闹钟。

运行长时间任务

Lifecycle 会等待每个作业完成后再启动下一个作业。缓慢的 onJob() 会延迟该对象上的所有其他作业。如果一次运行时间超过 hungTimeoutSeconds，Lifecycle 会记录警告并发出 job:slow_dispatch 事件。

对于长时间任务，请使用心跳机制。启动任务但不要等待其完成，将其交给 trackAlarmWork()，然后返回 { rescheduleAt }，让作业很快再次运行。每次心跳都会使任务继续运行。对象被逐出后，下次心跳会发现没有运行中的任务，并从存储中重新启动任务：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

const HEARTBEAT_MS = 30_000;

export class ImportAgent extends DurableObject {
	lifecycle = Lifecycle.install(this);
	#running;

	async startImport() {
		await this.lifecycle.start();
		await this.ctx.storage.put("import:pending", true);
		await this.lifecycle.jobs.push({
			id: "import",
			fn: "import",
			time: Date.now(),
			recoveryLoop: true,
		});
	}

	async onJob({ job }) {
		if (job.fn !== "import") return;
		if (!(await this.ctx.storage.get("import:pending"))) return;

		// Start the loop if this wake has none, for example after an eviction.
		this.#running ??= this.#runImport().finally(() => {
			this.#running = undefined;
		});
		this.lifecycle.trackAlarmWork(this.#running);
		return { rescheduleAt: Date.now() + HEARTBEAT_MS };
	}

	async #runImport() {
		// Resume from the last checkpoint in storage, then clear the flag.
		await this.ctx.storage.delete("import:pending");
	}
}

import { DurableObject } from "cloudflare:workers";
import {
	Lifecycle,
	type LifecycleJobContext,
	type LifecycleJobOutcome,
} from "agents/lifecycle";

const HEARTBEAT_MS = 30_000;

export class ImportAgent extends DurableObject<Env> {
	readonly lifecycle = Lifecycle.install(this);
	#running: Promise<void> | undefined;

	async startImport(): Promise<void> {
		await this.lifecycle.start();
		await this.ctx.storage.put("import:pending", true);
		await this.lifecycle.jobs.push({
			id: "import",
			fn: "import",
			time: Date.now(),
			recoveryLoop: true,
		});
	}

	async onJob({ job }: LifecycleJobContext): Promise<LifecycleJobOutcome> {
		if (job.fn !== "import") return;
		if (!(await this.ctx.storage.get("import:pending"))) return;

		// Start the loop if this wake has none, for example after an eviction.
		this.#running ??= this.#runImport().finally(() => {
			this.#running = undefined;
		});
		this.lifecycle.trackAlarmWork(this.#running);
		return { rescheduleAt: Date.now() + HEARTBEAT_MS };
	}

	async #runImport(): Promise<void> {
		// Resume from the last checkpoint in storage, then clear the flag.
		await this.ctx.storage.delete("import:pending");
	}
}

trackAlarmWork(promise) 会将交接的任务保留在当前闹钟的内存限制保护范围内。如果该任务之后触及内存限制，Lifecycle 会将其计入启动该任务的作业。闹钟运行期间调用 trackAlarmWork() 时返回 true；其他情况下返回 false，且不会跟踪任何任务。功能通过 this.lifecycle.trackAlarmWork() 调用此方法。PiHarness 以这种方式运行其代理循环。

内存限制保护

耗尽隔离环境内存的任务会重置对象。随后平台会重试闹钟，而闹钟可能再次运行相同任务并再次重置对象，如此无限循环。Lifecycle 使用断路器打破这个循环：

它会统计连续导致内存限制重置的闹钟次数。闹钟成功完成后，计数会重置。
每次未达到上限的重置都会延迟正在运行的作业，以及所有以 recoveryLoop: true 推送的待处理作业。每次重置延迟 30 秒，最多延迟五分钟。
达到上限的那次重置会封闭循环：删除正在运行的作业和所有 recoveryLoop 作业。其他作业不受影响。
它会调用每项功能的 onMemoryLimit() 钩子，然后调用主机的 onAlarmMemoryLimit() 钩子，并发出 alarm:memory_limit_reset 事件。

当作业恢复可能再次耗尽内存的任务时，请为作业标记 recoveryLoop，例如代理循环重放大型对话记录时。Scheduler 和 PiHarness 会以这种方式标记其恢复作业。

默认上限为三次重置。使用 maxAlarmMemoryLimitStrikes 选项进行设置，并实现 onAlarmMemoryLimit()，将失败记录到自己的状态中：

import { DurableObject } from "cloudflare:workers";
import { Lifecycle } from "agents/lifecycle";

export class MyAgent extends DurableObject {
	lifecycle = Lifecycle.install(this, {
		maxAlarmMemoryLimitStrikes: 5,
	});

	async onAlarmMemoryLimit({ sealed, executing }) {
		if (sealed && executing) {
			await this.ctx.storage.put(`failed:${executing.id}`, Date.now());
		}
	}
}

import { DurableObject } from "cloudflare:workers";
import { Lifecycle, type MemoryLimitContext } from "agents/lifecycle";

export class MyAgent extends DurableObject<Env> {
	readonly lifecycle = Lifecycle.install(this, {
		maxAlarmMemoryLimitStrikes: 5,
	});

	async onAlarmMemoryLimit({
		sealed,
		executing,
	}: MemoryLimitContext): Promise<void> {
		if (sealed && executing) {
			await this.ctx.storage.put(`failed:${executing.id}`, Date.now());
		}
	}
}

两个钩子都会在 Lifecycle 延迟或删除作业之后、隔离环境重置之前运行。请保持写入内容精简。MemoryLimitContext 包含以下字段：

字段	类型	说明
`sealed`	`boolean`	此重置是否达到上限并删除了恢复作业
`nextTime`	`number \| undefined`	延迟的作业再次运行的时间（适用于未达到上限的重置）
`executing`	`LifecycleJob \| undefined`	如果重置发生在作业期间，则为当时正在运行的作业
`purgedRecoveryLoopJobs`	`ReadonlyArray<LifecycleJob> \| undefined`	此次重置封闭循环时删除的 `recoveryLoop` 作业

事件

作业队列会在 agents:lifecycle 诊断渠道上尽力发出以下事件：

事件	触发时机
`job:backlog_warning`	某个所有者在单次闹钟中有 10 个或更多到期作业
`job:slow_dispatch`	一次 `onJob()` 调用的运行时间超过该作业的 `hungTimeoutSeconds`
`alarm:memory_limit_reset`	闹钟运行以一次内存限制重置结束

在 `Agent` 类上使用

Agent 类使用相同的作业队列。其 schedule() 和 scheduleEvery() 方法会通过 Scheduler 功能推送作业，延迟执行的 destroy() 则会推送一个 exclusive 作业。通过 Agent 的静态 options 设置 maxAlarmMemoryLimitStrikes。更多信息请参阅安排任务。