← Vytvářejte agenty na Cloudflare / agents / runtime / execution

Trvalé provádění pomocí fiberů

Spouštějte úlohy, které přežijí vyřazení Durable Object z paměti. runFiber() zaregistruje úlohu v SQLite, po dobu provádění udržuje agenta naživu a umožňuje ukládat mezistav do kontrolních bodů pomocí stash(), a volá onFiberRecovered() při další aktivaci, pokud byl agent v polovině úlohy odstraněn z paměti.

Použijte startFiber() když volající potřebuje trvale přijmout práci na pozadí, rychle vrátit odpověď, bezpečně deduplikovat opakované pokusy, později zkontrolovat stav nebo zrušit běžící úlohu.

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 se vyřazují z paměti ze tří důvodů:

1. Časový limit neaktivity : přibližně 70 až 140 sekund bez příchozích požadavků nebo otevřených WebSocketů
2. Aktualizace kódu / restarty runtime : nedeterministicky, 1 až 2krát denně
3. Časový limit handleru alarmu : 15 minut

Když dojde k vyřazení z paměti uprostřed práce, nadřazené připojení HTTP (k poskytovateli LLM, API nebo databázi) se natrvalo přeruší. Stav v paměti, tedy vyrovnávací paměti streamu, částečné odpovědi a čítače smyček, se ztratí. Vícekolové smyčky agenta zcela ztratí svou pozici.

keepAlive() snižuje pravděpodobnost vyřazení z paměti. runFiber() umožňuje přežít vyřazení z paměti.

Pro práci, která má běžet nezávisle na agentovi s opakováním jednotlivých kroků a vícekrokovou orchestrací, použijte Workflows místo toho. Objekty fiber slouží pro práci, která je součástí vlastního běhu agenta. Viz Dlouhodobě běžící agenti: Workflows oproti vzorům uvnitř agenta pro porovnání.

Zabraňuje vyřazení z paměti při nečinnosti vytvořením 30sekundového heartbeatu vázaného na alarm, který resetuje časovač nečinnosti.

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

keepAliveWhile() je doporučený postup: spustí asynchronní funkci a po jejím dokončení nebo vyvolání výjimky automaticky ukončí heartbeat:

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

Pro ruční kontrolu keepAlive() vrátí disposer. Po dokončení ho vždy zavolejte, jinak bude heartbeat pokračovat donekonečna:

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

Zatímco jakákoli keepAlive ref drží, spouští se alarm každých 30 sekund, který resetuje časovač nečinnosti. Jakmile jsou zavolány všechny disposery, alarmy se zastaví a DO může přirozeně přejít do nečinnosti.

Heartbeat je neviditelný pro listSchedules() : nevytvářejí se žádné řádky plánu. Nekoliduje to s vašimi vlastními plány, systém alarmů multiplexuje všechny plány i heartbeat keepAlive přes jediný alarmový slot.

Výchozí hodnota: 30 sekund. Časový limit neaktivity je přibližně 70 až 140 sekund, takže 30 sekund poskytuje dostatečnou rezervu. Přepsat lze pomocí statických možností:

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

keepAlive brání vyřazení z paměti, ale nijak neřeší obnovení. Pokud agent je odebrán z paměti i přes heartbeat (aktualizace kódu, vypršení časového limitu alarmu, limit prostředků), veškerá rozpracovaná práce se ztratí.

runFiber volání keepAlive interně a uloží práci do SQLite, aby ji bylo možné obnovit. Použijte keepAlive samotné, pokud je opakování práce levné nebo nevyžaduje vytváření kontrolních bodů. Použijte runFiber když je práce nákladná a potřebujete pokračovat tam, kde jste skončili.

Trvalé provádění s vytvářením kontrolních bodů a obnovením.

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

Obě cesty obnovy volají stejný hook. Cesta přes alarm je klíčová pro agenty na pozadí, kteří nemají žádná příchozí klientská připojení: uložený alarm probudí agenta sám o sobě.

Fibery fungují i uvnitř subagentů. Řádek fiberu a snímky se ukládají do vlastní databáze SQLite subagenta a onFiberRecovered() se spustí s podřízeným agentem jako this.

Podřízení agenti nemají nezávislé sloty pro alarmy, takže fyzický heartbeat vlastní nadřazený agent nejvyšší úrovně. Když podřízený agent spustí fiber, nadřazený agent eviduje dostatek metadat, aby mohl směrovat kontroly obnovy zpět k vlastnícímu podřízenému agentovi, i když potomek nemá žádné klientské připojení ani příchozí RPC.

Tím zůstává obnovení lokální u potomka, zatímco se zachovává jediný fyzický alarm slot patřící nadřazenému. Obnovené pokračování může použít schedule() zevnitř facety; nadřazený prvek vlastní fyzický alarm a směruje callback zpět k potomkovi.

fn(ctx) throws Error
  ├─ DELETE row from cf_agents_runs
  ├─ keepAlive dispose
  └─ Error propagates to caller (or logged if fire-and-forget)

Žádné automatické opakování. Logika obnovení patří do onFiberRecovered, kde máte snapshot a úplný kontext o tom, co se pokazilo.

runFiber() podporuje oba vzory:

// 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();
});

Pokud je DO vyřazen z paměti během vloženého await, volající už neexistuje. Při obnovení onFiberRecovered se spustí, nemůže ale vrátit výsledek původnímu volajícímu. Jde o vlastní omezení trvalého provádění napříč hranicemi procesů. Pro dlouhotrvající úlohy, které pravděpodobně přežijí životnost jedné instance DO, použijte startFiber() když volající potřebují zachovaný záznam stavu, idempotentní přijetí nebo zrušení.

Použijte startFiber() když volající potřebuje trvale přijmout práci na pozadí, rychle vrátit odpověď a bezpečně deduplikovat opakované pokusy. Před spuštěním callbacku uloží záznam zachovaného fiberu, poté callback spustí na pozadí pomocí stejného mechanismu keep-alive a obnovy jako 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.
}

Standardně startFiber() se vrátí poté, co je práce trvale přijata. Předejte waitForCompletion: true když volání má zůstat otevřené, dokud přijatý fiber nedosáhne konečného stavu. Duplicitní volání se stejným idempotentním klíčem se pokud možno připojí k aktivnímu provádění v paměti a poté vrátí zachovaný stav s 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() je trvalé API pro přijetí požadavku, nikoli API pro vrácení hodnoty. Vrací spravovaný stav fiber, nikoli výsledek callbacku. Stav zjistíte později pomocí inspectFiber() nebo 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),
});

Standardně deleteFibers() odstraní vyřízené completed, error, a aborted řádky. Nemaže interrupted řádky, pokud tento stav explicitně nepředáte, protože přerušené řádky často vyžadují kontrolu nebo ruční vyřešení.

Zrušení je kooperativní. cancelFiber() zaznamená přerušený koncový stav a přeruší ctx.signal pokud fiber běží v aktuálním izolátu. Váš callback by měl zkontrolovat ctx.signal.aborted kolem nákladných operací a před viditelnými vedlejšími efekty. Volající, kteří používají waitForCompletion: true se vrátí, jakmile deník dosáhne aborted, i když nekooperativní callback dál běží v aktuálním izolátu.

Pokud je Durable Object vyřazen z paměti uprostřed fiberu, uchovaný záznam se označí jako interrupted a onFiberRecovered() obdrží poslední kontrolní bod. Původní closure nelze automaticky přehrát, použijte ctx.name, ctx.snapshot, a metadata, aby bylo možné rozhodnout, zda pokračovat, kompenzovat, nebo ponechat záznam ke kontrole.

Vrátí FiberRecoveryResult z onFiberRecovered() k zaznamenání rozhodnutí zásady:

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 },
	};
}

Vracení undefined udržuje spravovaný fiber interrupted. Vyhození výjimky ponechá interrupted a zaznamená chybu zotavení pro pozdější kontrolu. Ukončená vlákna managed fiber, jako například aborted se znovu neobnoví, pokud zůstane zastaralý řádek běhu.

Pokud obnovení spustí až pozdější duplicitní webhook místo onFiberRecovered(), použijte resolveFiber() se stejným tvarem výsledku poté, co uspěje obnovení na úrovni aplikace. resolveFiber() aktualizuje pouze spravovaná vlákna, která jsou právě interrupted; vrátí false pro čekající, běžící nebo již ukončené řádky.

ctx.stash(data) zapisuje do SQLite synchronně. Mezi okamžikem „rozhodl jsem se uložit“ a okamžikem „je uloženo“ neexistuje žádná asynchronní mezera. Pokud k vyřazení z paměti dojde po stash() vrátí, je zaručeno, že data jsou v SQLite.

Každé volání zcela nahrazuje předchozí snímek, nejde o sloučení. Zapište celý stav obnovy, který potřebujete:

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),
		});
	}
});

Dělají totéž. ctx.stash() používá přímé uzavření (closure) nad ID fiberu. this.stash() používá AsyncLocalStorage k nalezení právě spuštěného fiberu. Funguje správně i u souběžných fiberů, protože kontext ALS každého fiberu je nezávislý.

this.stash() se hodí při volání z vnořených funkcí, které nemají přístup k ctx. Vyvolá chybu, pokud je volán mimo runFiber callback.

Přepsat onFiberRecovered ke zpracování přerušených fiberů. Výchozí implementace zaznamená varování a smaže řádek.

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),
					});
				}
			});
		}
	}
}

Klíčové body:

• Původní lambda funkce zanikla. Při obnově máte k dispozici pouze name a snapshot. Lambda funkci nelze serializovat, logika obnovy proto musí být v hooku.
• Nespravované runFiber() řádky se po úspěšném návratu hooku smažou. Pokud chcete pokračovat v nespravované práci, volejte runFiber() znovu uvnitř hooku, čímž se vytvoří nový řádek.
• Spravovaný startFiber() řádky zůstávají zachovány. Vraťte FiberRecoveryResult k označení přerušeného vlákna managed fiber jako completed, error, aborted, nebo stále interrupted.
• Co obnovení znamená, určujete vy. Opakovat od začátku, pokračovat od kontrolního bodu, přeskočit a upozornit uživatele, nebo nedělat nic. Framework nevynucuje žádnou konkrétní strategii.
• Pokud hook vyvolá chybu, řádek se zachová (až do daného limitu). Následné spuštění nebo kontrola alarmem opakuje pokus o obnovu, což chrání proti přechodným chybám úložiště nebo plánování. Chyby na úrovni aplikace zachytávejte sami, pokud chcete práci označit jako ukončenou místo opakování. Hook, který vždy vyvolá výjimku, se opakuje podle plánu s prodlužujícím se intervalem (obnovovací alarm používá exponenciální zpoždění omezené na 5 minut, takže nejde o aktivní smyčku), dokud řádek nepřekročí fiberRecoveryMaxAgeMs (výchozí 24 h), po jejichž uplynutí je zahozen s fiber:recovery:skipped (reason: "max_age_exceeded") událost. Nastavení fiberRecoveryMaxAgeMs: 0 takové řádky uchovává neomezeně dlouho: obnovení dál opakuje pokusy s prodlením omezeným horní hranicí a Durable Object se nikdy nevyřadí z paměti kvůli nečinnosti, dokud existuje neobnovitelný řádek, proto raději nastavte konečné stáří, pokud tyto řádky nechcete kontrolovat nebo mazat ručně sami. U spravované práce zůstává uchovaný řádek interrupted a zaznamená chybu zotavení pro pozdější kontrolu.

AIChatAgent a Think staví na fiberech pro obnovu streamování LLM. Každý krok konverzace je automaticky zabalen do fiberu. Framework se stará o interní cestu obnovy a zpřístupňuje onChatRecovery pro strategie specifické pro daného poskytovatele. Viz Dlouhodobě běžící agenti: obnova přerušených streamů LLM s podrobnostmi.

Více fiberů může běžet současně. Každý má svůj vlastní řádek v SQLite s vlastním snapshotem a každý volá keepAlive() nezávisle (s počítáním referencí, takže DO zůstává aktivní, dokud neskončí všechny fibery).

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

Při obnově se projdou všechny osiřelé řádky a onFiberRecovered se volá pro každý z nich. Použijte ctx.name k rozlišení mezi typy fiberů ve vašem recovery hooku.

V wrangler dev, obnovení fiberu funguje stejně jako v produkci. Stav SQLite a alarmů zůstává mezi restarty uložen na disku.

1. Spusťte svého agenta a vyvolejte fiber (runFiber)
2. Ukončete proces wrangler (Ctrl-C nebo SIGKILL)
3. Restartujte wrangler
4. Obnovení se spouští automaticky: prostřednictvím onStart() při příchodu požadavku, nebo prostřednictvím trvalého alarmu, pokud se nepřipojí žádný klient

Spustí trvalý fiber. Fiber se zaregistruje v SQLite ještě před fn se spustí a po dokončení (nebo vyvolání výjimky) se smaže. keepAlive() se drží po celou dobu trvání.

• name : identifikátor fiberu, použitý v onFiberRecovered k rozlišení typů fiberů. Není jedinečný, více fiberů může sdílet stejný název.
• fn : Asynchronní funkce přijímající FiberContext. Closures fungují přirozeně (this a lokální proměnné jsou zachyceny).
• Vrací : hodnota vrácená fn. Pokud je DO vyřazen z paměti před dokončením, návratová hodnota se ztratí; obnovu zajišťuje hook.

Trvale přijměte zachovaný fiber na pozadí. Vrácený StartFiberResult obsahuje vygenerovaný fiberId, aktuální status, volitelné metadata, a accepted, což je false když existující fiber odpovídal stejnému idempotentnímu klíči.

• name : identifikátor spravovaného fiberu, používaný při kontrole a obnovení.
• fn : Asynchronní funkce přijímající FiberContext. Výsledek funkce se neukládá.
• options.idempotencyKey : stabilní externí klíč používaný k deduplikaci opakování.
• options.metadata : Data serializovatelná do formátu JSON uložená spolu se zachovaným řádkem.
• options.waitForCompletion : před návratem počká na finální stav.

Vraťte zachovaný řádek stavu pro spravovaný fiber, nebo null pokud neexistuje žádný řádek.

Vypsat zachované spravované fibery. Filtrovat podle status nebo name, a použijte limit pro omezení množiny výsledků.

Označit spravovaný fiber jako aborted a zrušit jeho stav v paměti ctx.signal když běží v aktuálním izolátu. Vrací false pokud fiber neexistuje nebo je již ukončený.

Určit interrupted spravovaný fiber poté, co uspěje obnovení na úrovni aplikace. Vrací false pro čekající, běžící nebo již ukončené řádky.

Smaže uchované řádky spravovaného fiberu. Ve výchozím nastavení dokončené completed, error, a aborted řádky jsou způsobilé. Předejte status, settledBefore, nebo limit k zúžení rozsahu úklidu.

Vytvoří kontrolní bod aktuálního stavu fiberu. Zapisuje synchronně do SQLite. Každé volání zcela nahradí předchozí snímek. data musí být serializovatelné do JSON.

Volá se jednou pro každý osiřelý řádek fiberu při restartu agenta. Přepište tuto metodu, chcete-li implementovat obnovení. Nespravované runFiber() řádky se po úspěšném návratu tohoto hooku smažou. Pokud obnovení vyvolá výjimku, řádek zůstane zachovaný pro pozdější kontrolu, aby přechodná selhání neztratila handle pro obnovení. Spravované startFiber() řádky zůstávají zachovány a lze je vyřešit vrácením FiberRecoveryResult.

• ctx.id : jedinečné ID fiberu
• ctx.name : název předaný do runFiber()
• ctx.status : zachovaný stav pro spravované fibery
• ctx.idempotencyKey : klíč idempotence pro spravované fibery, pokud je zadán
• ctx.metadata : metadata pro spravované fibery, pokud jsou zadána
• ctx.snapshot : poslední stash() data, nebo null v případě, že stash() nebyl nikdy zavolán
• ctx.createdAt : milisekundy epochy, kdy runFiber() spustilo. Porovnejte s Date.now() a přeskočit obnovení, která jsou příliš stará na to, aby je bylo možné bezpečně přehrát.
• ctx.recoveryReason : proč obnovení právě probíhá. V současnosti vždy "interrupted" pro vyřazení z paměti nebo zotavení po restartu.

Vytvořte 30sekundový heartbeat alarmu. Vrací funkci disposer. Je idempotentní: opakované volání disposeru je bezpečné.

Spusťte asynchronní funkci a přitom udržujte DO naživu. Heartbeat se spustí před fn a zastaví se, jakmile se dokončí nebo vyvolá výjimku. Vrací hodnotu, kterou vrátí fn.

• Dlouhodobě běžící agenti : jak fibery spolupracují s plánovanými úlohami, plány a asynchronními operacemi
• Plánování úloh, keepAlive podrobnosti a systém alarmů
• Podřízení agenti : trvalé provádění a plánování uvnitř podřízených agentů
• Workflows : trvalé vícekrokové provádění mimo agenta
• Chat agenti, chatRecovery a onChatRecovery