← Создание агентов на Cloudflare / agents / runtime / execution

Устойчивое выполнение с помощью fiber

Запускайте задачи, которые переживают вытеснение 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 вытесняются из памяти по трём причинам:

1. Тайм-аут бездействия : примерно от 70 до 140 секунд без входящих запросов и открытых WebSocket-соединений
2. Обновления кода / перезапуски runtime : недетерминированно, от 1 до 2 раз в день
3. Тайм-аут обработчика будильника : 15 минут

Если вытеснение происходит посреди работы, upstream HTTP-соединение (с провайдером LLM, API или базой данных) обрывается безвозвратно. Состояние в памяти, то есть буферы потоковой передачи, частичные ответы и счётчики циклов, теряется. Многоходовые агентные циклы полностью теряют своё положение.

keepAlive() снижает вероятность вытеснения. runFiber() позволяет пережить вытеснение.

Для работы, которая должна выполняться независимо от агента с повторами на каждом шаге и многошаговой оркестрацией, используйте Workflows вместо этого. Fiber предназначены для работы, которая является частью собственного выполнения агента. См. Долгоживущие агенты: Workflows против внутренних паттернов агента для сравнения.

Предотвращает вытеснение при простое, создавая heartbeat на основе 30-секундного будильника, который сбрасывает таймер неактивности.

class Agent {
	keepAlive(): Promise<() => void>;
	keepAliveWhile<T>(fn: () => Promise<T>): Promise<T>;
}

keepAliveWhile() это рекомендуемый подход: он запускает асинхронную функцию и автоматически останавливает heartbeat по её завершении или при исключении:

const result = await this.keepAliveWhile(async () => {
	return await slowAPICall();
});

Для ручного контроля keepAlive() возвращает disposer. Всегда вызывайте его по завершении, иначе heartbeat будет продолжаться бесконечно:

const dispose = await this.keepAlive();
try {
	await longWork();
} finally {
	dispose();
}

Пока любой keepAlive ref удерживается, alarm срабатывает каждые 30 секунд и сбрасывает таймер бездействия. Когда вызваны все disposer, alarm останавливается, и DO может естественным образом перейти в состояние простоя.

Heartbeat невидим для listSchedules() : не создаются строки расписания. Это не конфликтует с вашими собственными расписаниями: система alarm мультиплексирует все расписания и keepAlive heartbeat через единственный слот alarm.

По умолчанию: 30 секунд. Таймаут бездействия составляет примерно от 70 до 140 секунд, поэтому 30 секунд дают достаточный запас. Переопределите его через статические параметры:

class MyAgent extends Agent {
	static options = { keepAliveIntervalMs: 2_000 };
}

keepAlive предотвращает вытеснение, но никак не решает восстановление. Если агент является вытесняется несмотря на heartbeat (обновление кода, истечение времени будильника, лимит ресурсов), вся незавершённая работа теряется.

runFiber вызывает keepAlive внутренне и сохраняет работу в SQLite, чтобы её можно было восстановить. Используйте keepAlive отдельно, если работу дёшево повторить или она не требует контрольных точек. Используйте runFiber когда работа затратна и нужно возобновить её с того места, где она была прервана.

Устойчивое выполнение с созданием контрольных точек и восстановлением.

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

Оба пути восстановления вызывают один и тот же хук. Путь через alarm критически важен для фоновых агентов, у которых нет входящих клиентских подключений: сохранённый alarm сам будит агента.

Fiber работают и внутри дочерних агентов. Строка fiber и снимки хранятся в собственной базе данных SQLite дочернего агента, и onFiberRecovered() выполняется с sub-agent в качестве this.

У дочерних агентов нет собственных слотов alarm, поэтому физический heartbeat принадлежит родительскому агенту верхнего уровня. Когда дочерний агент запускает fiber, родитель хранит достаточно метаданных, чтобы направлять проверки восстановления обратно к владеющему дочернему агенту, даже если у потомка нет клиентского подключения или входящих RPC.

Благодаря этому восстановление остаётся локальным для дочернего агента, а единственный физический слот alarm, принадлежащий родителю, сохраняется. Восстановленное продолжение может использовать 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();
});

Если DO вытесняется во время встроенного await, вызывающая сторона уже недоступна. При восстановлении onFiberRecovered срабатывает, но не может вернуть результат исходному вызывающему коду. Это неотъемлемое ограничение долговечного выполнения на границах процессов. Для длительных задач, которые могут пережить время жизни одного DO, используйте startFiber() когда вызывающему коду нужны сохранённая запись статуса, идемпотентный приём или отмена.

Используйте startFiber() когда вызывающему коду нужно устойчиво принять фоновую работу, быстро вернуть управление и безопасно устранять дублирование повторных попыток. Он сохраняет запись о fiber перед выполнением колбэка, а затем запускает колбэк в фоновом режиме, используя тот же механизм keep-alive и восстановления, что и runFiber().

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() возвращается после того, как работа надёжно принята. Передайте waitForCompletion: true когда вызывающая сторона должна оставаться открытой, пока принятый fiber не достигнет конечного статуса. Повторные вызовы с тем же ключом идемпотентности по возможности присоединяются к активному выполнению в памяти, а затем возвращают сохранённый статус с 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 rows. Это не удаляет interrupted rows, если вы явно не передадите этот статус, поскольку прерванные rows часто требуют проверки или ручного разрешения.

Отмена является кооперативной. cancelFiber() записывает прерванное конечное состояние и прерывает ctx.signal если файбер выполняется в текущем изоляте. Ваш callback должен проверить ctx.signal.aborted вокруг затратных операций и перед видимыми побочными эффектами. Вызывающий код, использующий waitForCompletion: true возвращается, когда ledger достигает aborted, даже если некооперативный колбэк продолжает выполняться в текущем isolate.

Если Durable Object вытесняется в процессе выполнения fiber, сохранённая запись помечается как interrupted и onFiberRecovered() получает последнюю контрольную точку. Исходное closure нельзя воспроизвести автоматически, используйте ctx.name, ctx.snapshot, и метаданные, чтобы решить, возобновлять работу, компенсировать её или оставить запись для проверки.

Возвращает FiberRecoveryResult от onFiberRecovered() чтобы зафиксировать решение политики:

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 и записывает ошибку восстановления для последующего анализа. Завершённые управляемые fiber, такие как aborted повторно не восстанавливаются, если остаётся устаревшая строка выполнения.

Если восстановление запускается более поздним дублирующим вебхуком вместо onFiberRecovered(), используйте resolveFiber() с той же формой результата после успешного восстановления на уровне приложения. resolveFiber() обновляет только те управляемые fiber, которые в данный момент interrupted; он возвращает 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() использует прямой closure над ID fiber. this.stash() использует AsyncLocalStorage чтобы найти текущий выполняющийся fiber: это работает корректно даже при параллельных fiber, поскольку контекст ALS у каждого fiber независим.

this.stash() удобен при вызове из вложенных функций, у которых нет доступа к ctx. Он выбрасывает исключение при вызове за пределами runFiber колбэк.

Переопределить onFiberRecovered чтобы обрабатывать прерванные fiber. Реализация по умолчанию записывает предупреждение в лог и удаляет строку.

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. Lambda нельзя сериализовать: логика восстановления должна находиться в hook.
• Неуправляемый runFiber() rows удаляются после успешного завершения hook. Если вы хотите продолжить неуправляемую работу, вызовите runFiber() снова внутри хука: это создаёт новую строку.
• Managed startFiber() rows сохраняются. Верните FiberRecoveryResult чтобы пометить прерванный управляемый fiber как completed, error, aborted, либо всё ещё interrupted.
• Вы сами определяете, что означает восстановление. Повторить с начала, продолжить с контрольной точки, пропустить с уведомлением пользователя или ничего не делать. Фреймворк не навязывает конкретную стратегию.
• Если хук выбрасывает исключение, строка сохраняется (до заданного предела). Последующий запуск или проверка по alarm повторяют восстановление, что защищает от временных сбоев хранилища или планирования. Перехватывайте ошибки уровня приложения самостоятельно, если хотите пометить работу как завершённую вместо повторной попытки. Hook, который всегда выбрасывает исключение, повторяется по расписанию с нарастающей отсрочкой (recovery alarm использует экспоненциальную задержку с ограничением в 5 минут, поэтому это не цикл активного опроса) до тех пор, пока строка не превысит fiberRecoveryMaxAgeMs (по умолчанию 24 ч), после чего он отбрасывается с fiber:recovery:skipped (reason: "max_age_exceeded") событие. Настройка fiberRecoveryMaxAgeMs: 0 бесконечно сохраняет такие rows: recovery продолжает повторные попытки с ограниченной задержкой, и Durable Object никогда не вытесняется из памяти по бездействию, пока существует невосстановимая row. Поэтому лучше задавать конечный возраст, если только вы не планируете сами просматривать или очищать эти rows. Для managed работы сохранённая row остаётся interrupted и записывает ошибку восстановления для последующего анализа.

AIChatAgent и Think построены на fiber для восстановления потоковой передачи LLM. Каждый ход чата автоматически оборачивается в fiber. Фреймворк обрабатывает внутренний путь восстановления и предоставляет onChatRecovery для стратегий, специфичных для провайдера. См. Долгоживущие агенты: восстановление прерванных LLM-потоков для получения подробностей.

Несколько fiber могут работать одновременно. У каждого fiber есть своя строка в SQLite со своим снимком, и каждый из них вызывает keepAlive() независимо (с подсчётом ссылок, поэтому DO остаётся активным, пока не завершатся все файберы).

void this.runFiber("fetch-data", async (ctx) => {
	/* ... */
});
void this.runFiber("process-queue", async (ctx) => {
	/* ... */
});

При восстановлении перебираются все осиротевшие строки, и onFiberRecovered вызывается для каждого. Используйте ctx.name чтобы различать типы fiber в хуке восстановления.

В wrangler dev, восстановление fiber работает так же, как в продакшене. Состояние SQLite и alarm сохраняется на диске между перезапусками.

1. Запустите своего агента и инициируйте fiber (runFiber)
2. Завершите процесс Wrangler (Ctrl-C или SIGKILL)
3. Перезапустить wrangler
4. Восстановление запускается автоматически: через onStart() при поступлении запроса, либо через сохранённый будильник, если не подключится ни один клиент

Выполнять устойчивый fiber. Fiber регистрируется в SQLite до того, как fn выполняется и удаляется после завершения (или возникновения ошибки). keepAlive() удерживается на всё время выполнения.

• name : идентификатор fiber, используемый в onFiberRecovered чтобы различать типы fiber. Значение не уникально: несколько fiber могут иметь одно и то же имя.
• fn : асинхронная функция, принимающая FiberContext. Замыкания работают естественным образом (this и локальные переменные захватываются).
• Возвращает : значение, возвращённое fn. Если DO вытесняется до завершения, возвращаемое значение теряется; восстановление происходит через hook.

Устойчиво примите сохранённый фоновый fiber. Возвращённый StartFiberResult включает сгенерированный fiberId, текущий status, необязательный metadata, а также accepted, который false когда существующий fiber совпал по тому же ключу идемпотентности.

• name : идентификатор управляемого fiber, используемый при проверке и восстановлении.
• fn : асинхронная функция, принимающая FiberContext. Результат функции не сохраняется.
• options.idempotencyKey : стабильный внешний ключ, используемый для устранения дублирования повторных попыток.
• options.metadata : данные, сериализуемые в JSON, которые хранятся вместе с сохранённой строкой.
• options.waitForCompletion : ждёт конечного статуса перед возвратом.

Верните сохранённую строку статуса для управляемого fiber или null если строка не существует.

Выводит список сохранённых управляемых fiber. Фильтрация по status или name, и используйте limit чтобы ограничить набор результатов.

Пометить управляемый fiber как aborted и прерывает своё хранящееся в памяти ctx.signal когда он выполняется в текущем isolate. Возвращает false если файбер не существует или уже завершён.

Разрешить interrupted управляемый fiber после успешного восстановления на уровне приложения. Возвращает false для строк в состоянии ожидания, выполнения или уже в конечном состоянии.

Удаляет сохранённые строки managed fiber. По умолчанию завершённые completed, error, а также aborted rows подходят. Передайте status, settledBefore, или limit чтобы сузить область очистки.

Создаёт контрольную точку текущего состояния fiber. Синхронно записывает в SQLite. Каждый вызов полностью заменяет предыдущий снимок. data должен сериализоваться в JSON.

Вызывается один раз для каждой осиротевшей строки fiber при перезапуске агента. Переопределите для реализации восстановления. Unmanaged runFiber() rows удаляются после успешного завершения этого hook; если recovery выбрасывает ошибку, row остаётся для последующего сканирования, чтобы временные сбои не приводили к потере recovery handle. Managed startFiber() rows остаются сохранёнными и могут быть разрешены путём возврата FiberRecoveryResult.

• ctx.id : уникальный ID fiber
• ctx.name : имя, переданное в runFiber()
• ctx.status : сохранённый статус для управляемых fiber
• ctx.idempotencyKey : ключ идемпотентности для управляемых fiber, если он указан
• ctx.metadata : метаданные для управляемых fiber, если они указаны
• ctx.snapshot : последний stash() данные, или null если stash() ни разу не был вызван
• ctx.createdAt : миллисекунды эпохи, когда runFiber() запущен. Сравните с Date.now() чтобы пропустить восстановления, которые слишком стары для безопасного повторного воспроизведения.
• ctx.recoveryReason : почему выполняется восстановление. В настоящий момент всегда "interrupted" для вытеснения или восстановления после перезапуска.

Создайте alarm-heartbeat с интервалом 30 секунд. Возвращает функцию отмены. Идемпотентно: повторный вызов этой функции безопасен.

Запустите асинхронную функцию, сохраняя DO активным. Heartbeat запускается до того, как fn и останавливается при завершении или возникновении исключения. Возвращает значение, возвращённое fn.

• Долгоживущие агенты : как fiber взаимодействуют с расписаниями, планами и асинхронными операциями
• Планирование задач, keepAlive сведения и систему alarm
• Дочерние агенты : устойчивое выполнение и расписания внутри дочерних агентов
• Workflows : устойчивое многошаговое выполнение за пределами агента
• Чат-агенты, chatRecovery и onChatRecovery