Chat agents·续1

import type { ChatResponseResult } from "@cloudflare/ai-chat";

export class ChatAgent extends AIChatAgent {
    protected async onChatResponse(result: ChatResponseResult) {
        if (result.status === "completed") {
            console.log("Turn completed:", result.requestId);
        }
        if (result.status === "error") {
            console.error("Turn failed:", result.error);
        }
    }
}

The ChatResponseResult contains:

Note

Responses triggered from inside onChatResponse (for example, via saveMessages) do not fire onChatResponse recursively.

sanitizeMessageForPersistence

Override this method to apply custom transformations to messages before they are persisted to storage. This hook runs after the built-in sanitization (OpenAI metadata stripping, Anthropic provider-executed tool payload truncation, empty reasoning part filtering).

export class ChatAgent extends AIChatAgent {
    sanitizeMessageForPersistence(message) {
        return {
            ...message,
            parts: message.parts.map((part) => {
                if (
                    "output" in part &&
                    typeof part.output === "string" &&
                    part.output.length > 1000
                ) {
                    return { ...part, output: "[redacted]" };
                }
                return part;
            }),
        };
    }
}

export class ChatAgent extends AIChatAgent {
    protected sanitizeMessageForPersistence(message: UIMessage): UIMessage {
        return {
            ...message,
            parts: message.parts.map((part) => {
                if (
                    "output" in part &&
                    typeof part.output === "string" &&
                    part.output.length > 1000
                ) {
                    return { ...part, output: "[redacted]" };
                }
                return part;
            }),
        };
    }
}

Turn lifecycle helpers

These methods help you coordinate programmatic turns and wait for pending interactions.

hasPendingInteraction()

Returns true when an assistant message is waiting on a client tool result or approval.

if (this.hasPendingInteraction()) {
    console.log("Waiting for user to approve or provide tool output");
}

if (this.hasPendingInteraction()) {
    console.log("Waiting for user to approve or provide tool output");
}

waitUntilStable()

Waits until the conversation is fully stable — no active stream, no pending client-tool interactions, and no queued continuation turns. Returns true when stable, or false if the timeout expires before a pending interaction resolves.

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (stable) {
    console.log("All turns complete, safe to proceed");
}

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (stable) {
    console.log("All turns complete, safe to proceed");
}

This is especially useful with saveMessages for server-driven flows:

await this.saveMessages((messages) => [...messages, syntheticUserMessage]);
await this.waitUntilStable({ timeout: 60_000 });
// The assistant has finished responding

await this.saveMessages((messages) => [...messages, syntheticUserMessage]);
await this.waitUntilStable({ timeout: 60_000 });
// The assistant has finished responding

resetTurnState()

Aborts the active turn and invalidates queued continuations. The built-in CF_AGENT_CHAT_CLEAR handler calls this automatically, but you can call it manually if needed.

Lifecycle hooks

Override onConnect and onClose to add custom logic. Stream resumption and message sync are handled for you:

export class ChatAgent extends AIChatAgent {
    async onConnect(connection, ctx) {
        // Your custom logic (e.g., logging, auth checks)
        console.log("Client connected:", connection.id);
        // Stream resumption and message sync are handled automatically
    }

async onClose(connection, code, reason, wasClean) {
        console.log("Client disconnected:", connection.id);
        // Connection cleanup is handled automatically
    }
}

export class ChatAgent extends AIChatAgent {
    async onConnect(connection, ctx) {
        // Your custom logic (e.g., logging, auth checks)
        console.log("Client connected:", connection.id);
        // Stream resumption and message sync are handled automatically
    }

async onClose(connection, code, reason, wasClean) {
        console.log("Client disconnected:", connection.id);
        // Connection cleanup is handled automatically
    }
}

The destroy() method cancels any pending chat requests and cleans up stream state. It is called automatically when the Durable Object is evicted, but you can call it manually if needed.

Request cancellation

When a user clicks "stop" in the chat UI, the client sends a CF_AGENT_CHAT_REQUEST_CANCEL message. The server propagates this to the abortSignal in options:

export class ChatAgent extends AIChatAgent {
    async onChatMessage(_onFinish, options) {
        const result = streamText({
            model: ai("@cf/zai-org/glm-4.7-flash"),
            messages: await convertToModelMessages(this.messages),
            abortSignal: options?.abortSignal, // Pass through for cancellation
        });

return result.toUIMessageStreamResponse();
    }
}

export class ChatAgent extends AIChatAgent {
    async onChatMessage(_onFinish, options) {
        const result = streamText({
            model: ai("@cf/zai-org/glm-4.7-flash"),
            messages: await convertToModelMessages(this.messages),
            abortSignal: options?.abortSignal, // Pass through for cancellation
        });

return result.toUIMessageStreamResponse();
    }
}

Caution

If you do not pass abortSignal to streamText, the LLM call will continue running in the background even after the user cancels. Always forward it when possible.

Subclasses can also cancel turns from inside the Durable Object:

protected abortRequest(requestId: string, reason?: unknown): void
protected abortAllRequests(): void

Use abortRequest() when you know the request ID. Use abortAllRequests() for single-purpose helpers that should cancel whatever turn is currently running. Prefer SaveMessagesOptions.signal for programmatic turns when you can pass a signal at the call site.

Stream recovery

Automatic stream resumption (the resume option on useAgentChat) is client reconnect recovery — it resumes an active stream when a client disconnects and reconnects. It does not cover Durable Object eviction. If the Worker process or Durable Object is evicted while the model call is in flight, the stream itself is gone. Durable chat recovery handles that case.

A mid-stream Durable Object eviction permanently severs the LLM connection. Durable recovery wraps every AIChatAgent and Think chat turn in a runFiber(). The fiber provides automatic keepAlive during streaming and a recovery hook on restart.

The fiber row survives in SQLite after an eviction. On the next activation, the framework detects the interrupted fiber. It reconstructs the partial response from buffered stream chunks and calls onChatRecovery.

Durable recovery is always on. Use chatRecovery only to tune recovery budgets and terminal behavior:

export class ChatAgent extends AIChatAgent {
    chatRecovery = {
        maxAttempts: 10,
        stableTimeoutMs: 10_000,
        terminalMessage: "The assistant was interrupted and could not recover.",
        // Primary stuck-turn bound. Resets on every progress-bearing attempt, so a
        // turn that keeps producing content survives unbounded interruption.
        noProgressTimeoutMs: 5 * 60 * 1000,
        // Runaway-loop guard. Defaults to 10,000. Set a higher value for a long
        // agentic turn, or Infinity to remove the cap.
        maxRecoveryWork: 2000,
        // How many of one turn's attempts may die before recovery gives up, for
        // example to a memory-limit reset on every run. Defaults to 5, counting
        // the original turn.
        maxInterruptedAttempts: 5,
        // Caller policy consulted from the second recovery attempt onward. Return
        // false to stop recovery. This is where you enforce a token/cost budget.
        // Note: this is called as `config.shouldKeepRecovering(ctx)`, so it is not
        // bound to the agent instance — track spend in your own store keyed by the
        // incident.
        async shouldKeepRecovering(ctx) {
            return (await getSpendForTurn(ctx.recoveryRootRequestId)) < MAX_SPEND;
        },
        async onExhausted(ctx) {
            console.warn("Chat recovery exhausted", ctx.incidentId, ctx.reason);
        },
    };
}

export class ChatAgent extends AIChatAgent {
    override chatRecovery = {
        maxAttempts: 10,
        stableTimeoutMs: 10_000,
        terminalMessage: "The assistant was interrupted and could not recover.",
        // Primary stuck-turn bound. Resets on every progress-bearing attempt, so a
        // turn that keeps producing content survives unbounded interruption.
        noProgressTimeoutMs: 5 * 60 * 1000,
        // Runaway-loop guard. Defaults to 10,000. Set a higher value for a long
        // agentic turn, or Infinity to remove the cap.
        maxRecoveryWork: 2000,
        // How many of one turn's attempts may die before recovery gives up, for
        // example to a memory-limit reset on every run. Defaults to 5, counting
        // the original turn.
        maxInterruptedAttempts: 5,
        // Caller policy consulted from the second recovery attempt onward. Return
        // false to stop recovery. This is where you enforce a token/cost budget.
        // Note: this is called as `config.shouldKeepRecovering(ctx)`, so it is not
        // bound to the agent instance — track spend in your own store keyed by the
        // incident.
        async shouldKeepRecovering(ctx) {
            return (await getSpendForTurn(ctx.recoveryRootRequestId)) < MAX_SPEND;
        },
        async onExhausted(ctx) {
            console.warn("Chat recovery exhausted", ctx.incidentId, ctx.reason);
        },
    };
}

The chatRecovery object accepts the following configuration options:

ChatRecoveryProgressContext (the ctx passed to shouldKeepRecovering) contains the following fields:

A progressing turn survives repeated interruptions as long as it stays within the maxRecoveryWork limit and fewer than maxInterruptedAttempts of its attempts die. Recovery is sealed by one of these ctx.reason values:

• no_progress_timeout — no forward progress within the no-progress window (a stuck turn).
• max_attempts_exceeded — the attempt cap was spent on a tight no-progress alarm loop.
• work_budget_exceeded — the turn kept producing content but exceeded maxRecoveryWork (a runaway loop).
• interrupted_attempts_exceeded — the turn's attempts kept dying before they finished, for example in a memory-limit reset on every run, until maxInterruptedAttempts of them had been interrupted.
• out_of_memory — the alarm memory-limit circuit breaker sealed the recovery after repeated memory-limit resets.
• recovery_aborted — your shouldKeepRecovering hook returned false.
• stable_timeout — recovery attempts kept timing out waiting for stable state until the budget drained (extreme churn).

Tip

The maxRecoveryWork default prevents a progressing turn from running forever. Increase it for long agentic turns. Use shouldKeepRecovering with durable token or cost data for a precise budget.

Turns waiting on a human are not sealed

A turn can pause on a client interaction it cannot resolve on its own: a client-side tool call (a tool with no server execute, whose result the client replays), or an approval-requested part. Such a turn is waiting on the human, not stuck.

While the interaction is pending, the turn is exempt from every recovery budget. The no-progress window, attempt cap, maxRecoveryWork, and shouldKeepRecovering are all suspended. A user who takes minutes to answer a prompt that was interrupted by a deploy never trips a seal. Recovery parks the turn instead of failing it, and the user's eventual approval or tool result resumes it through the normal continuation path.

This exemption is client-only. A server tool whose execute() was killed mid-flight is a genuine orphan, so it is not exempt and recovers through transcript repair instead.

Monitor terminal exhaustion through observability:

import { subscribe } from "agents/observability";

const unsubscribe = subscribe("chat", (event) => {
    if (event.type === "chat:recovery:exhausted") {
        console.error("Chat recovery exhausted", event.payload);
    }
});

import { subscribe } from "agents/observability";

const unsubscribe = subscribe("chat", (event) => {
    if (event.type === "chat:recovery:exhausted") {
        console.error("Chat recovery exhausted", event.payload);
    }
});

onChatRecovery

Override to implement provider-specific recovery. The default behavior persists the partial response and schedules a continuation via continueLastTurn().

export class ChatAgent extends AIChatAgent {
    async onChatRecovery(ctx) {
        console.log(`Recovered ${ctx.partialText.length} chars of partial text`);

// Default: persist partial + schedule continuation
        return {};
    }
}

import type {
    ChatRecoveryContext,
    ChatRecoveryOptions,
} from "@cloudflare/ai-chat";

export class ChatAgent extends AIChatAgent {
    override async onChatRecovery(
        ctx: ChatRecoveryContext,
    ): Promise<ChatRecoveryOptions> {
        console.log(`Recovered ${ctx.partialText.length} chars of partial text`);

// Default: persist partial + schedule continuation
        return {};
    }
}

ChatRecoveryContext: