import { useAgent } from "agents/react"; import { useState } from "react"; import { Button, Input, Surface, Text } from "@cloudflare/kumo"; import { DemoWrapper } from "../../layout"; import { CodeExplanation, ConnectionStatus, LogPanel, type CodeSection } from "../../components"; import { useLogs, useToast, useUserId } from "../../hooks"; import type { DurableExecutionAgent, DurableExecutionState, ResearchJob } from "./durable-execution-agent"; const codeSections: CodeSection[] = [ { title: "Start durable work with runFiber", description: "runFiber records the job in the agent's SQLite storage before it starts. While the function runs, keepAlive prevents ordinary idle eviction and ctx.stash writes checkpoints.", code: `class ReportAgent extends Agent { @callable() async startResearchReport(topic: string) { const id = crypto.randomUUID(); void this.runFiber(\`research-report:\${id}\`, async (ctx) => { for (const step of steps) { ctx.stash({ id, topic, step, updatedAt: Date.now() }); await runStep(step); } }); return id; } }` }, { title: "Recover interrupted fibers", description: "If the Durable Object restarts while a fiber row still exists, the framework calls onFiberRecovered with the last checkpoint. Your app decides whether to resume, mark stale, or ask a user.", code: `async onFiberRecovered(ctx: FiberRecoveryContext) { const snapshot = ctx.snapshot as { id: string; step: string; } | null; await this.notifyClients({ status: "recovered", jobId: snapshot?.id, lastCheckpoint: snapshot?.step, }); }` }, { title: "Use the right durable primitive", description: "Fibers are best for one long-running operation with checkpoints. Queue is best for many small tasks with retries. Workflows are best when the process itself is the durable product object.", code: `// One long-running operation await this.runFiber("index-docs", async (ctx) => { ctx.stash({ phase: "embedding" }); await embedDocuments(); }); // Many independent tasks await this.queue("processItem", { id }, { retry: { maxAttempts: 5 } }); // A product-level durable process await this.runWorkflow("ApprovalWorkflow", { requestId });` } ]; function formatTime(timestamp?: number) { if (!timestamp) return "Not yet"; return new Date(timestamp).toLocaleTimeString(); } function statusClass(status: ResearchJob["status"]) { switch (status) { case "completed": return "bg-kumo-success-tint text-kumo-success"; case "failed": return "bg-kumo-danger-tint text-kumo-danger"; case "recovered": return "bg-kumo-warning-tint text-kumo-warning"; default: return "bg-kumo-info-tint text-kumo-info"; } } function JobCard({ job }: { job: ResearchJob }) { const progress = Math.round((job.currentStep / job.totalSteps) * 100); return (

{job.topic}

Started {formatTime(job.startedAt)} · Completed{" "} {formatTime(job.completedAt)}

{job.status}

Step {job.currentStep} of {job.totalSteps} {progress}%

{job.checkpoint}

{job.notes.length > 0 && (

{note}

)} ); } export function DurableExecutionDemo() { const userId = useUserId(); const { logs, addLog, clearLogs } = useLogs(); const { toast } = useToast(); const [topic, setTopic] = useState("Compare MCP transports for my app"); const agent = useAgent({ agent: "durable-execution-agent", name: `durable-execution-${userId}`, onOpen: () => addLog("info", "connected"), onClose: () => addLog("info", "disconnected"), onError: () => addLog("error", "error", "Connection error"), onMessage: (message: MessageEvent) => { try { const data = JSON.parse(message.data as string) as { type?: string; payload?: unknown; }; if (data.type) { addLog("in", data.type, data.payload); } } catch { // Ignore protocol messages that are not JSON. } } }); const jobs = agent.state?.jobs ?? []; const recoveries = agent.state?.recoveries ?? []; const isRunning = jobs.some((job) => job.status === "running"); const startJob = async () => { if (!topic.trim()) return; addLog("out", "startResearchReport", { topic }); try { const id = await agent.call("startResearchReport", [topic]); addLog("in", "job_id", { id }); toast("Durable research job started", "success"); } catch (e) { addLog("error", "error", e instanceof Error ? e.message : String(e)); } }; const clear = async () => { clearLogs(); try { await agent.call("clearJobs", []); } catch { // Ignore cleanup failures while disconnected. } }; return ( Use durable execution when an agent needs to keep working after the user navigates away, the Worker sleeps, or a long task needs checkpoints. This demo starts a report job with{" "}


            runFiber()

, writes checkpoints with{" "}


            ctx.stash()

, and shows where recovery hooks fit. } statusIndicator={ } >

Fiber Research Job

Start a multi-step job, then refresh the page while it runs. The agent state and event history reconnect through the same Durable Object instance.

) => setTopic(e.target.value) } placeholder="Topic" className="w-full" />

Which Primitive Should I Use?

{[ [ "Fiber", "One long-running operation with checkpoints and custom recovery." ], [ "Queue", "Many small callbacks that need retry, backoff, batching, or dedupe." ], [ "Workflow", "A product-level process that needs progress, approvals, listing, and external visibility." ] ].map(([name, description]) => (

{name}

{description}

))}

Jobs {isRunning ? "Working" : "Idle"}

{jobs.length === 0 ? (

Start a job to see checkpoints and completion state.

) : (

{jobs.map((job) => ( ))}

)} {recoveries.length > 0 && (

Recoveries

{recoveries.map((recovery) => (

{recovery.name}

Recovered {formatTime(recovery.recoveredAt)}

))}

)}

); }