Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions .changeset/unified-fiber-architecture.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
---
"agents": minor
"@cloudflare/ai-chat": minor
"@cloudflare/think": minor
---

Add durable fiber execution to the Agent base class.

`runFiber(name, fn)` registers work in SQLite, holds a `keepAlive` ref, and enables recovery via `onFiberRecovered` after DO eviction. `ctx.stash()` and `this.stash()` checkpoint progress that survives eviction.

`AIChatAgent` gains `durableStreaming` — when enabled, each chat turn is wrapped in a fiber. `onChatRecovery` provides provider-specific recovery (Workers AI continuation, OpenAI response retrieval, Anthropic synthetic message). `continueLastTurn()` appends to the interrupted assistant message seamlessly.

`Think` now extends `Agent` directly (no mixin). Fiber support is inherited from the base class.

**Breaking (experimental APIs only):**

- Removed `withFibers` mixin (`agents/experimental/forever`)
- Removed `withDurableChat` mixin (`@cloudflare/ai-chat/experimental/forever`)
- Removed `./experimental/forever` export from both packages
- Think no longer has a `fibers` flag — recovery is automatic via alarm housekeeping
300 changes: 87 additions & 213 deletions design/think.md

Large diffs are not rendered by default.

5 changes: 3 additions & 2 deletions experimental/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ The code here is for exploration, prototyping, and validating patterns that may
## Contents

- **[gadgets.md](./gadgets.md)** — Exploration of facets, isolation, and structural safety for agents. Covers the Gatekeeper/ApprovalQueue pattern, Worker Loader sandboxing, sub-agent facets, multi-room chat, and other patterns worth pulling into the SDK.
- **[forever.md](./forever.md)** — Design doc for durable long-running execution. Covers keepAlive, fibers (spawnFiber/stashFiber), eviction recovery, and AIChatAgent integration. Implemented as mixins in `agents/experimental/forever` and `@cloudflare/ai-chat/experimental/forever`.
- **[forever.md](./forever.md)** — Design doc for durable long-running execution. Covers `keepAlive`, `runFiber` (checkpointing, eviction recovery), and `AIChatAgent` chat recovery via `onChatRecovery`. Built into the `Agent` and `AIChatAgent` base classes.
- **[forever-fibers/](./forever-fibers/)** — Example of `Agent.runFiber()` for durable background work with real-time progress tracking.
- **[forever-chat/](./forever-chat/)** — Example of `AIChatAgent` with `durableStreaming` for multi-provider chat recovery (Workers AI, OpenAI, Anthropic).
- **[session-memory](./session-memory/)** — Example of the Session API for conversation history with automatic compaction. Demonstrates `agents/experimental/memory/session` with micro-compaction and LLM-based summarization.
- **[workspace-chat](../examples/workspace-chat/)** — _(moved to examples/)_ AIChatAgent with a persistent virtual filesystem.
16 changes: 7 additions & 9 deletions experimental/forever-chat/README.md
Original file line number Diff line number Diff line change
@@ -1,24 +1,22 @@
# Forever Chat — Durable AI Streaming

> **WARNING: EXPERIMENTAL.** This example uses APIs from `agents/experimental/forever` and `@cloudflare/ai-chat/experimental/forever` that are under active development and **will break** between releases. Do not use in production. Pin your package versions and expect to rewrite your code when upgrading.

AI chat using `withDurableChat(AIChatAgent)` — the mixin adds keepAlive during streaming and automatic recovery after DO eviction.
AI chat with `durableStreaming` enabled — wraps each chat turn in `runFiber` for automatic keepAlive during streaming and recovery after DO eviction.

See [forever.md](../forever.md) for the full design doc.

## What it shows

- `withDurableChat` mixin from `@cloudflare/ai-chat/experimental/forever`
- `durableStreaming = true` on `AIChatAgent` — wraps chat turns in fibers
- keepAlive during streaming — DO stays alive for long LLM responses
- `onChatRecovery` — provider-specific recovery after eviction
- `continueLastTurn()` — seamlessly continues the interrupted assistant message inline
- Multi-provider support with a dropdown selector:

| Provider | Model | Recovery strategy |
| ---------- | ----------------- | --------------------------------------------------------------------------------------- |
| Workers AI | kimi-k2.5 | Persist partial + inline continuation via `continueLastTurn()` |
| OpenAI | gpt-5.4 | Retrieve completed response via Responses API (`store: true`) — zero wasted tokens |
| Anthropic | claude-sonnet-4.6 | Persist partial + continue via synthetic user message (reasoning disabled for recovery) |
| Provider | Model | Recovery strategy |
| ---------- | ----------------- | ------------------------------------------------------------------------------------------------- |
| Workers AI | kimi-k2.5 | Persist partial + continue via `continueLastTurn()` (text + reasoning merge into existing blocks) |
| OpenAI | gpt-5.4 | Retrieve completed response via Responses API (`store: true`) — zero wasted tokens |
| Anthropic | claude-sonnet-4.6 | Persist partial + continue via synthetic user message (reasoning disabled for recovery) |

## Run it

Expand Down
68 changes: 37 additions & 31 deletions experimental/forever-chat/src/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,25 @@
* Forever Chat — Durable AI streaming with multi-provider recovery.
*
* Demonstrates three recovery strategies after DO eviction:
* - Workers AI: persist partial + inline continuation via continueLastTurn()
* - Workers AI: persist partial + continue via continueLastTurn()
* (text + reasoning merge into existing blocks)
* - OpenAI: retrieve completed response via Responses API (store: true)
* - Anthropic: persist partial + continue via synthetic user message
* (no prefill support, reasoning disabled for recovery)
*
* Uses the withDurableChat mixin for automatic keepAlive during
* Uses durableStreaming for automatic keepAlive during
* streaming and onChatRecovery for provider-specific recovery.
*/
import { createWorkersAI } from "workers-ai-provider";
import { createOpenAI } from "@ai-sdk/openai";
import { createAnthropic } from "@ai-sdk/anthropic";
import { routeAgentRequest } from "agents";
import { AIChatAgent } from "@cloudflare/ai-chat";
import type { OnChatMessageOptions } from "@cloudflare/ai-chat";
import {
withDurableChat,
AIChatAgent,
type ChatRecoveryContext,
type ChatRecoveryOptions
} from "@cloudflare/ai-chat/experimental/forever";
type ChatRecoveryOptions,
type OnChatMessageOptions
} from "@cloudflare/ai-chat";
import {
streamText,
convertToModelMessages,
Expand Down Expand Up @@ -100,11 +100,13 @@ const SYSTEM_PROMPT =
"You can check the weather and perform calculations. " +
"For calculations with large numbers (over 1000), you need user approval first.";

// ── Agent ─────────────────────────────────────────────────────────────
const RECOVERY_SUFFIX =
" Do not think or reason — continue the text output directly.";

const DurableChatAgent = withDurableChat(AIChatAgent);
// ── Agent ─────────────────────────────────────────────────────────────

export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {
export class ForeverChatAgent extends AIChatAgent<Env, AgentState> {
override durableStreaming = true;
maxPersistedMessages = 200;

// ── Recovery ────────────────────────────────────────────────────
Expand All @@ -124,11 +126,24 @@ export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {
return { continue: false };
}

// Workers AI: use continueLastTurn() (prefill works). Set the
// recovering flag so onChatMessage can hint the model to skip
// reasoning. Even if the model still reasons, the framework merges
// reasoning into the existing block automatically.
if (provider === "workersai" || !provider) {
await this.schedule(0, "_continueWorkersAI", undefined, {
idempotent: true
});
return { continue: false };
}

if (provider !== "openai") return {};

// OpenAI Responses API: the generation continues server-side even
// after our connection drops. Retrieve the completed response by ID.
const responseId = this._getStoredResponseId();
// after our connection drops. Retrieve the completed response by ID
// (stashed via this.stash() during streaming).
const responseId = (ctx.recoveryData as { responseId?: string } | null)
?.responseId;
if (!responseId) return {};

try {
Expand Down Expand Up @@ -173,6 +188,14 @@ export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {
}
}

async _continueWorkersAI() {
const ready = await this.waitUntilStable({ timeout: 10_000 });
if (!ready) return;

this._lastBody = { ...this._lastBody, recovering: true };
await this.continueLastTurn();
}

async _continueWithUserMessage() {
const ready = await this.waitUntilStable({ timeout: 10_000 });
if (!ready) return;
Expand Down Expand Up @@ -206,7 +229,7 @@ export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {

const result = streamText({
model: this._getModel(provider),
system: SYSTEM_PROMPT,
system: recovering ? SYSTEM_PROMPT + RECOVERY_SUFFIX : SYSTEM_PROMPT,
messages: pruneMessages({
messages: await convertToModelMessages(this.messages),
toolCalls: "before-last-2-messages",
Expand Down Expand Up @@ -265,7 +288,7 @@ export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {
| { type?: string; response?: { id?: string } }
| undefined;
if (raw?.type === "response.created" && raw.response?.id) {
this._storeResponseId(raw.response.id);
this.stash({ responseId: raw.response.id });
}
}
};
Expand All @@ -281,23 +304,6 @@ export class ForeverChatAgent extends DurableChatAgent<Env, AgentState> {
}
return {};
}

// ── OpenAI response ID persistence ─────────────────────────────

private _storeResponseId(id: string) {
this.sql`
INSERT OR REPLACE INTO cf_ai_chat_request_context (key, value)
VALUES ('openaiResponseId', ${id})
`;
}

private _getStoredResponseId(): string | undefined {
const rows = this.sql<{ value: string }>`
SELECT value FROM cf_ai_chat_request_context
WHERE key = 'openaiResponseId'
`;
return rows?.[0]?.value;
}
}

export default {
Expand Down
11 changes: 3 additions & 8 deletions experimental/forever-fibers/README.md
Original file line number Diff line number Diff line change
@@ -1,19 +1,14 @@
# Forever Fibers — Durable Execution Demo

> **WARNING: EXPERIMENTAL.** This example uses APIs from `agents/experimental/forever` that are under active development and **will break** between releases. Do not use in production. Pin your package versions and expect to rewrite your code when upgrading.

Demonstrates durable long-running execution with fibers — fire-and-forget work that survives Durable Object eviction via SQLite checkpointing and alarm-based recovery.
Demonstrates durable long-running execution with `Agent.runFiber()` — work that survives Durable Object eviction via SQLite checkpointing and alarm-based recovery.

See [forever.md](../forever.md) for the full design doc.

## What it shows

- `withFibers` mixin from `agents/experimental/forever`
- `spawnFiber()` — start a multi-step research task that runs in the background
- `stashFiber()` — checkpoint progress after each step (persisted in SQLite)
- `runFiber()` — start a multi-step research task that runs in the background
- `ctx.stash()` — checkpoint progress after each step (persisted in SQLite)
- `onFiberRecovered()` — automatically resume from the last checkpoint after eviction
- `cancelFiber()` — stop a running fiber
- Simulated eviction — demonstrates the recovery flow via the UI button
- Real eviction testing — kill the wrangler process externally and restart; alarms persist to disk, so recovery happens automatically (same as production)

## Run it
Expand Down
90 changes: 9 additions & 81 deletions experimental/forever-fibers/src/app.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,8 @@ function ModeToggle() {
export default function App() {
const [topic, setTopic] = useState("quantum computing");
const [steps, setSteps] = useState<StepInfo[]>([]);
const [fiberId, setFiberId] = useState<string | null>(null);
const [status, setStatus] = useState<
"idle" | "running" | "complete" | "cancelled" | "recovered"
"idle" | "running" | "complete" | "recovered"
>("idle");
const [results, setResults] = useState<ResearchStep[]>([]);
const [connectionStatus, setConnectionStatus] =
Expand All @@ -94,7 +93,6 @@ export default function App() {

switch (msg.type) {
case "research:started":
setFiberId(msg.fiberId);
setStatus("running");
setResults([]);
setSteps(
Expand Down Expand Up @@ -137,10 +135,6 @@ export default function App() {
setResults(msg.results);
break;

case "research:cancelled":
setStatus("cancelled");
break;

case "research:failed":
setStatus("idle");
break;
Expand All @@ -158,22 +152,6 @@ export default function App() {
await agent.call("startResearch", [topic.trim()]);
}, [topic, agent]);

const handleCancel = useCallback(async () => {
if (!agent) return;
await agent.call("cancelResearch", []);
}, [agent]);

const handleKillAndRecover = useCallback(async () => {
if (!agent) return;
setStatus("recovered");
await agent.call("simulateKillAndRecover", []);
// Recovery is async — the fiber restarts in the background
// and broadcasts progress updates as it resumes
setTimeout(() => {
if (status === "recovered") setStatus("running");
}, 2000);
}, [agent, status]);

const isRunning = status === "running" || status === "recovered";

return (
Expand Down Expand Up @@ -206,33 +184,20 @@ export default function App() {
if (e.key === "Enter") handleStart();
}}
/>
{isRunning ? (
<Button variant="destructive" onClick={handleCancel}>
Cancel
</Button>
) : (
<Button
variant="primary"
onClick={handleStart}
disabled={!topic.trim()}
>
Start Research
</Button>
)}
<Button
variant="primary"
onClick={handleStart}
disabled={!topic.trim() || isRunning}
>
{isRunning ? "Running..." : "Start Research"}
</Button>
</div>
</div>

{/* Steps */}
{steps.length > 0 && (
<div className="rounded-lg border border-kumo-line p-4">
<div className="flex items-center justify-between">
<Text variant="heading3">Progress</Text>
{fiberId && (
<span className="font-mono text-xs text-kumo-inactive">
{fiberId.slice(0, 8)}...
</span>
)}
</div>
<Text variant="heading3">Progress</Text>

<div className="mt-3 flex flex-col gap-2">
{steps.map((step, i) => (
Expand Down Expand Up @@ -261,23 +226,6 @@ export default function App() {
</div>
)}

{/* Eviction Demo */}
{isRunning && (
<div className="rounded-lg border border-kumo-line p-4">
<Text variant="heading3">Simulate Eviction</Text>
<p className="mt-1 text-sm text-kumo-inactive">
In production, Durable Objects can be evicted by code updates or
inactivity. This simulates that — the fiber state persists in
SQLite, and recovery picks up from the last checkpoint.
</p>
<div className="mt-3">
<Button variant="secondary" onClick={handleKillAndRecover}>
Simulate Kill &amp; Recover
</Button>
</div>
</div>
)}

{/* Recovery banner */}
{status === "recovered" && (
<div className="rounded-lg border border-green-500/30 bg-green-500/5 p-4">
Expand Down Expand Up @@ -307,26 +255,6 @@ export default function App() {
setSteps([]);
setResults([]);
setStatus("idle");
setFiberId(null);
}}
>
Start New Research
</Button>
</div>
</div>
)}

{/* Cancelled */}
{status === "cancelled" && (
<div className="rounded-lg border border-kumo-line p-4">
<Text variant="body">Research cancelled.</Text>
<div className="mt-3">
<Button
variant="secondary"
onClick={() => {
setSteps([]);
setStatus("idle");
setFiberId(null);
}}
>
Start New Research
Expand Down
Loading
Loading