使用 fiber 的持久执行

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

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

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

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

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

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

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

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

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.
}

const result = await this.startFiber("reply-to-webhook", reply, {
	idempotencyKey: `webhook:${webhookId}`,
	waitForCompletion: true,
});

if (result.status === "error") {
	console.error(result.error);
}

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

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

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

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

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