Durable Object のエビクションを乗り越える作業を実行します。runFiber() はタスクを SQLite に登録し、実行中はエージェントを生存させ、stash() で中間状態をチェックポイントし、タスク途中でエージェントがエビクションされた場合は次回起動時に onFiberRecovered() を呼びます。
呼び出し側がバックグラウンド作業を耐久的に受け付け、すばやく返し、再試行を安全に重複排除し、後から状態を確認し、実行中のジョブをキャンセルする必要があるときは startFiber() を使います。
import { Agent } from "agents";
import type { FiberRecoveryContext } from "agents";
class MyAgent extends Agent {
async doWork() {
await this.runFiber("my-task", async (ctx) => {
const step1 = await expensiveOperation();
ctx.stash({ step1 });
const step2 = await anotherExpensiveOperation(step1);
this.setState({ ...this.state, result: step2 });
});
}
async onFiberRecovered(ctx: FiberRecoveryContext) {
if (ctx.name !== "my-task") return;
const snapshot = ctx.snapshot as { step1: unknown } | null;
if (snapshot) {
const step2 = await anotherExpensiveOperation(snapshot.step1);
this.setState({ ...this.state, result: step2 });
}
}
}Durable Objects がエビクションされる理由は 3 つあります。
- 非アクティブタイムアウト — 受信リクエストも開いている WebSocket もない状態が約 70〜140 秒続く
- コード更新 / ランタイム再起動 — 非決定的で、1 日あたり 1〜2 回
- アラームハンドラのタイムアウト — 15 分
作業の途中でエビクションが起きると、上流の HTTP 接続（LLM プロバイダー、API、データベース）は永続的に切れます。メモリ上の状態（ストリーミングバッファ、部分応答、ループカウンタ）は失われます。複数ターンのエージェントループは、進行位置を完全に失います。
keepAlive() はエビクションの可能性を下げます。runFiber() はエビクションを生き延びられるようにします。
エージェントから独立して動かし、ステップごとの再試行と複数ステップのオーケストレーションが必要な作業には、代わりに Workflows を使います。fiber は、エージェント自身の実行の一部である作業向けです。比較は 長時間稼働するエージェント: Workflows とエージェント内部パターン を参照してください。
30 秒ごとのアラームハートビートを作り、非アクティブタイマーをリセットしてアイドルエビクションを防ぎます。
class Agent {
keepAlive(): Promise<() => void>;
keepAliveWhile<T>(fn: () => Promise<T>): Promise<T>;
}推奨は keepAliveWhile() です。非同期関数を実行し、完了または例外時にハートビートを自動で片付けます。
const result = await this.keepAliveWhile(async () => {
return await slowAPICall();
});手動制御では、keepAlive() は disposer を返します。終わったら必ず呼んでください。呼ばないとハートビートが無期限に続きます。
const dispose = await this.keepAlive();
try {
await longWork();
} finally {
dispose();
}keepAlive の参照が 1 つでも残っているあいだ、30 秒ごとにアラームが発火し、非アクティブタイマーをリセットします。すべての disposer が呼ばれるとアラームは止まり、DO は自然にアイドルになれます。
ハートビートは listSchedules() からは見えません。スケジュール行は作られません。独自のスケジュールとも衝突しません。アラームシステムは、すべてのスケジュールと keepAlive ハートビートを 1 つのアラーム枠で多重化します。
デフォルトは 30 秒です。非アクティブタイムアウトは約 70〜140 秒なので、30 秒なら余裕があります。静的オプションで上書きできます。
class MyAgent extends Agent {
static options = { keepAliveIntervalMs: 2_000 };
}keepAlive はエビクションを防ぎますが、復旧はしません。ハートビートがあってもエージェントがエビクションされた場合（コード更新、アラームタイムアウト、リソース上限）、進行中の作業は失われます。
runFiber は内部で keepAlive を呼び、作業を SQLite に永続化して復旧できるようにします。やり直しが安価、またはチェックポイントが不要なときは keepAlive だけを使います。作業が高価で、中断した位置から再開したいときは runFiber を使います。
| シナリオ | 使うもの |
|---|---|
| 遅い API 呼び出しを待つ | keepAlive() |
LLM 応答のストリーミング（AIChatAgent 経由） |
自動（組み込み） |
| 中間結果がある複数ステップの計算 | runFiber() |
| 10 分以上かかるバックグラウンド調査ループ | stash() 付きの runFiber() |
| ちょうど 1 回だけ受け付ける必要のある webhook ジョブ | startFiber() |
チェックポイントと復旧付きの耐久実行です。
class Agent {
runFiber<T>(name: string, fn: (ctx: FiberContext) => Promise<T>): Promise<T>;
startFiber(
name: string,
fn: (ctx: FiberContext) => Promise<void>,
options?: StartFiberOptions,
): Promise<StartFiberResult>;
inspectFiber(fiberId: string): Promise<FiberInspection | null>;
inspectFiberByKey(idempotencyKey: string): Promise<FiberInspection | null>;
listFibers(options?: ListFibersOptions): Promise<FiberInspection[]>;
cancelFiber(fiberId: string, reason?: string): Promise<boolean>;
cancelFiberByKey(idempotencyKey: string, reason?: string): Promise<boolean>;
deleteFibers(options?: DeleteFibersOptions): Promise<number>;
resolveFiber(fiberId: string, result: FiberRecoveryResult): Promise<boolean>;
stash(data: unknown): void;
onFiberRecovered(
ctx: FiberRecoveryContext,
): Promise<void | FiberRecoveryResult>;
}
type FiberContext = {
id: string;
signal: AbortSignal;
stash(data: unknown): void;
snapshot: unknown | null;
};
type FiberStatus =
| "pending"
| "running"
| "completed"
| "aborted"
| "interrupted"
| "error";
type FiberRecoveryContext = {
id: string;
name: string;
status?: FiberStatus;
idempotencyKey?: string;
metadata?: Record<string, unknown> | null;
snapshot: unknown | null;
createdAt: number;
recoveryReason: "interrupted";
};runFiber("work", fn)
├─ Persist recovery metadata
├─ keepAlive() — heartbeat starts
├─ Execute fn(ctx)
│ ├─ ctx.stash(data) → persist snapshot
│ ├─ ctx.stash(data) → persist snapshot
│ └─ return result
├─ Delete recovery metadata
├─ keepAlive dispose — heartbeat stops
└─ Return result to caller[DO evicted — all in-memory state lost]
On next activation:
├─ Request/connection → onStart() → check for orphaned fibers [primary path]
│ OR
├─ Persisted alarm fires → housekeeping check [fallback path]
Recovery:
├─ Load interrupted fibers from storage
├─ For each interrupted fiber:
│ ├─ Parse snapshot from JSON
│ ├─ Call onFiberRecovered(ctx)
│ └─ Delete recovery metadata after successful recovery
└─ If onFiberRecovered calls runFiber() again → new fiber, normal executionどちらの復旧パスも同じフックを呼びます。アラーム経路は、受信クライアント接続がないバックグラウンドエージェントに重要です。永続化されたアラームがエージェント自身を起こします。
fiber はサブエージェント内でも動きます。fiber 行とスナップショットはサブエージェント自身の SQLite データベースに保存され、onFiberRecovered() はサブエージェントを this として実行されます。
サブエージェントには独立したアラーム枠がないため、物理ハートビートはトップレベルの親が所有します。サブエージェントが fiber を始めると、親は復旧チェックを所有側のサブエージェントへ戻すためのメタデータを十分に追跡します。子にクライアント接続や受信 RPC がなくても同様です。
これにより復旧は子のローカルに保ちつつ、親が所有する単一の物理アラーム枠を維持します。復旧後の継続処理は facet 内から schedule() を使えます。親が物理アラームを所有し、コールバックを子へルーティングします。
fn(ctx) throws Error
├─ DELETE row from cf_agents_runs
├─ keepAlive dispose
└─ Error propagates to caller (or logged if fire-and-forget)自動再試行はありません。復旧ロジックは onFiberRecovered に置きます。スナップショットと、何が失敗したかの十分なコンテキストがそこにあります。
runFiber() は両方のパターンをサポートします。
// Inline — await the result
const result = await this.runFiber("work", async (ctx) => {
return computeExpensiveThing();
});
// Fire-and-forget — caller does not wait
void this.runFiber("background", async (ctx) => {
await longRunningProcess();
});インラインの await 中に DO がエビクションされると、呼び出し側はいなくなります。復旧時は onFiberRecovered が発火しますが、元の呼び出し側へ結果を返せません。これはプロセス境界をまたぐ耐久実行の本質的な制約です。1 回の DO 寿命を超える可能性が高い長時間作業で、呼び出し側が状態レコードの保持、冪等な受付、キャンセルを必要とする場合は startFiber() を使います。
呼び出し側がバックグラウンド作業を耐久的に受け付け、すばやく返し、再試行を安全に重複排除する必要があるときは startFiber() を使います。コールバック実行前に保持される fiber レコードを保存し、runFiber() と同じ keep-alive と復旧の仕組みでコールバックをバックグラウンド開始します。
const receipt = await this.startFiber(
"reply-to-webhook",
async (ctx) => {
ctx.stash({ webhookId, threadId });
await postReply(threadId);
},
{
idempotencyKey: `webhook:${webhookId}`,
metadata: { threadId },
},
);
if (!receipt.accepted) {
// This webhook was already accepted by an earlier delivery.
}デフォルトでは、startFiber() は作業が耐久的に受け付けられたあとに返します。受け付け済み fiber が終端状態になるまで呼び出し側を開いたままにする場合は waitForCompletion: true を渡します。同じ冪等キーでの重複呼び出しは、可能なときはアクティブなメモリ上の実行に合流し、accepted: false 付きの保持状態を返します。
const result = await this.startFiber("reply-to-webhook", reply, {
idempotencyKey: `webhook:${webhookId}`,
waitForCompletion: true,
});
if (result.status === "error") {
console.error(result.error);
}startFiber() は耐久受付 API であり、値を返す API ではありません。管理対象 fiber の状態は返しますが、コールバックの結果は返しません。後から状態を見るには inspectFiber() または inspectFiberByKey() を使います。
const current = await this.inspectFiberByKey(`webhook:${webhookId}`);
if (current) {
await this.cancelFiber(current.fiberId, "No longer needed");
}
await this.deleteFibers({
status: ["completed", "error", "aborted"],
settledBefore: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000),
});デフォルトでは、deleteFibers() は確定した completed、error、aborted 行を削除します。interrupted 行は、その状態を明示しない限り削除しません。中断行は検査や手動解決が必要なことが多いためです。
キャンセルは協調的です。cancelFiber() は中止の終端状態を記録し、fiber が現在の isolate で実行中なら ctx.signal を中止します。コールバックは高コストな作業の前後と、目に見える副作用の前に ctx.signal.aborted を確認してください。waitForCompletion: true の呼び出し側は、台帳が aborted になると返します。非協調的なコールバックが現在の isolate で動き続けていても同様です。
fiber の途中で Durable Object がエビクションされると、保持レコードは interrupted になり、onFiberRecovered() は最後のチェックポイントを受け取ります。元のクロージャは自動では再生できません。再開するか、補償するか、検査のために残すかは、ctx.name、ctx.snapshot、メタデータで判断します。
方針の決定を記録するには、onFiberRecovered() から FiberRecoveryResult を返します。
async onFiberRecovered(ctx: FiberRecoveryContext) {
if (ctx.name !== "reply-to-webhook") return;
const snapshot = ctx.snapshot as { webhookId: string; threadId: string };
await postRecoveryMessage(snapshot.threadId);
return {
status: "completed",
snapshot: { ...snapshot, recovered: true },
};
}undefined を返すと、管理対象 fiber は interrupted のままです。例外を投げると interrupted のままで、検査用に復旧エラーを記録します。aborted などの終端済み管理対象 fiber は、古い実行行が残っていても再復旧しません。
復旧が onFiberRecovered() ではなく、後続の重複 webhook で起きた場合は、アプリケーション層の復旧が成功したあとに同じ結果形で resolveFiber() を使います。resolveFiber() が更新するのは、現在 interrupted の管理対象 fiber だけです。pending、running、またはすでに終端の行には false を返します。
ctx.stash(data) は SQLite へ同期的に書き込みます。「保存すると決めた」と「保存された」のあいだに非同期の隙間はありません。stash() が返ったあとにエビクションが起きても、データは SQLite にあることが保証されます。
各呼び出しは前回のスナップショットを完全に置き換えます。マージではありません。復旧に必要な状態をすべて書いてください。
await this.runFiber("research", async (ctx) => {
const steps = ["search", "analyze", "synthesize"];
const completed: string[] = [];
const results: Record<string, unknown> = {};
for (const step of steps) {
results[step] = await executeStep(step);
completed.push(step);
ctx.stash({
completed,
results,
pendingSteps: steps.slice(completed.length),
});
}
});どちらも同じことをします。ctx.stash() は fiber ID への直接クロージャを使います。this.stash() は AsyncLocalStorage で現在実行中の fiber を探します。各 fiber の ALS コンテキストは独立なので、並行 fiber でも正しく動きます。
this.stash() は、ctx にアクセスできないネストした関数から呼ぶときに便利です。runFiber コールバックの外で呼ぶと例外を投げます。
中断した fiber を扱うには onFiberRecovered をオーバーライドします。デフォルト実装は警告をログし、行を削除します。
class ResearchAgent extends Agent {
async onFiberRecovered(ctx: FiberRecoveryContext) {
if (ctx.name !== "research") return;
const snapshot = ctx.snapshot as {
completed: string[];
results: Record<string, unknown>;
pendingSteps: string[];
} | null;
if (snapshot && snapshot.pendingSteps.length > 0) {
void this.runFiber("research", async (fiberCtx) => {
const { completed, results, pendingSteps } = snapshot;
for (const step of pendingSteps) {
results[step] = await this.executeStep(step);
completed.push(step);
fiberCtx.stash({
completed,
results,
pendingSteps: pendingSteps.slice(pendingSteps.indexOf(step) + 1),
});
}
});
}
}
}要点は次のとおりです。
- 元のラムダはもうありません。 復旧時に手元にあるのは
nameとsnapshotだけです。ラムダはシリアライズできないので、復旧ロジックはフックに置く必要があります。 - 管理対象外の
runFiber()行は、フックが正常に返ったあとに削除されます。 管理対象外の作業を続けたい場合は、フック内で再度runFiber()を呼びます。新しい行が作られます。 - 管理対象の
startFiber()行は保持されます。 中断した管理対象 fiber をcompleted、error、aborted、または引き続きinterruptedにするにはFiberRecoveryResultを返します。 - 復旧の意味は自分で決めます。 最初から再試行する、チェックポイントから再開する、スキップしてユーザーに通知する、何もしない、のいずれでも構いません。フレームワークは戦略を強制しません。
- フックが例外を投げると、行は（上限まで）残ります。 後続の起動またはアラーム走査が復旧を再試行します。一時的なストレージやスケジュールの失敗から守れます。作業を終端にして再試行したくないときは、アプリケーション層のエラーを自分で捕捉してください。常に例外を投げるフックは、バックオフ付きスケジュールで再試行されます（復旧アラームは指数遅延で上限 5 分なので、ビジーループにはなりません）。行が
fiberRecoveryMaxAgeMs（デフォルト 24 時間）を超えると、fiber:recovery:skipped（reason: "max_age_exceeded"）イベント付きで破棄されます。fiberRecoveryMaxAgeMs: 0にすると、そのような行を無期限に保持します。復旧は上限付きバックオフで再試行し続け、復旧不能な行があるあいだ Durable Object はアイドルエビクションしません。自分で検査または削除するつもりがない限り、有限の寿命を推奨します。管理対象の作業では、保持行はinterruptedのままで、検査用に復旧エラーを記録します。
AIChatAgent と Think は、LLM ストリーミング復旧に fiber を使います。すべてのチャットターンは自動で fiber に包まれます。フレームワークが内部の復旧経路を扱い、プロバイダー固有の戦略向けに onChatRecovery を公開します。詳細は 長時間稼働するエージェント: 中断した LLM ストリームの復旧 を参照してください。
複数の fiber を同時に実行できます。それぞれが SQLite に独自の行とスナップショットを持ち、それぞれが独立して keepAlive() を呼びます（参照カウントなので、すべての fiber が完了するまで DO は生存します）。
void this.runFiber("fetch-data", async (ctx) => {
/* ... */
});
void this.runFiber("process-queue", async (ctx) => {
/* ... */
});復旧時は、孤立したすべての行を走査し、それぞれに対して onFiberRecovered を呼びます。復旧フックでは ctx.name で fiber の種類を区別します。
wrangler dev では、fiber 復旧は本番と同じように動きます。SQLite とアラーム状態は、再起動のあいだディスクに残ります。
- エージェントを起動し、fiber（
runFiber）を開始します - wrangler プロセスを終了します（Ctrl-C または SIGKILL）
- wrangler を再起動します
- 復旧は自動で発火します。リクエストが来れば
onStart()経由、クライアントが接続しなければ永続化されたアラーム経由です
耐久 fiber を実行します。fiber は fn の実行前に SQLite へ登録され、完了（または例外）後に削除されます。そのあいだ keepAlive() が保持されます。
name— fiber の識別子。onFiberRecoveredで種類を区別します。一意ではありません。複数の fiber が同じ名前を共有できます。fn—FiberContextを受け取る非同期関数。クロージャは自然に使えます（thisとローカル変数がキャプチャされます）。- 戻り値 —
fnが返した値。完了前に DO がエビクションされると戻り値は失われ、復旧はフック経由で行われます。
保持されるバックグラウンド fiber を耐久的に受け付けます。返される StartFiberResult には、生成された fiberId、現在の status、任意の metadata、および accepted が含まれます。同じ冪等キーに既存 fiber が一致した場合、accepted は false です。
name— 管理対象 fiber の識別子。検査と復旧で使います。fn—FiberContextを受け取る非同期関数。関数の結果は保存されません。options.idempotencyKey— 再試行の重複排除に使う、安定した外部キーです。options.metadata— 保持行と一緒に保存する、JSON シリアライズ可能なデータです。options.waitForCompletion— 返す前に終端状態を待ちます。
管理対象 fiber の保持状態行を返します。行がなければ null です。
保持されている管理対象 fiber を一覧します。status または name で絞り込み、limit で件数を上限できます。
管理対象 fiber を aborted にし、現在の isolate で実行中ならメモリ上の ctx.signal を中止します。fiber が存在しない、またはすでに終端なら false を返します。
アプリケーション層の復旧が成功したあと、interrupted の管理対象 fiber を解決します。pending、running、またはすでに終端の行には false を返します。
保持されている管理対象 fiber 行を削除します。デフォルトでは、確定した completed、error、aborted 行が対象です。クリーンアップを絞るには status、settledBefore、または limit を渡します。
現在の fiber の状態をチェックポイントします。SQLite へ同期書き込みします。各呼び出しは前回のスナップショットを完全に置き換えます。data は JSON シリアライズ可能である必要があります。
エージェント再起動時に、孤立した fiber 行ごとに 1 回呼ばれます。復旧を実装するにはオーバーライドします。管理対象外の runFiber() 行は、このフックが正常に返ったあとに削除されます。復旧が例外を投げると、行は後続の走査のために残ります。一時障害で復旧ハンドルを失わないためです。管理対象の startFiber() 行は保持され、FiberRecoveryResult を返すことで解決できます。
ctx.id— 一意の fiber IDctx.name—runFiber()に渡した名前ctx.status— 管理対象 fiber の保持状態ctx.idempotencyKey— 管理対象 fiber の冪等キー（指定した場合）ctx.metadata— 管理対象 fiber のメタデータ（指定した場合）ctx.snapshot— 最後のstash()データ。stash()を一度も呼んでいなければnullctx.createdAt—runFiber()開始時のエポックミリ秒。安全に再生するには古すぎる復旧をスキップするためにDate.now()と比較します。ctx.recoveryReason— 復旧が動いている理由。現在はエビクションまたは再起動復旧のため、常に"interrupted"です。
30 秒のアラームハートビートを作ります。disposer 関数を返します。冪等です。disposer を複数回呼んでも安全です。
DO を生存させたまま非同期関数を実行します。ハートビートは fn の前に始まり、完了または例外時に止まります。fn が返した値を返します。
- 長時間稼働するエージェント — fiber がスケジュール、プラン、非同期操作とどう組み合わさるか
- タスクのスケジュール —
keepAliveの詳細とアラームシステム - サブエージェント — サブエージェント内の耐久実行とスケジュール
- Workflows — エージェント外の耐久的な複数ステップ実行
- チャットエージェント —
chatRecoveryとonChatRecovery