Merge pull request #349 from code-yeongyu/feat/server-fallback-receipt-abort

feat(ai,coding-agent): abort server-side fallback receipts and reroute through client chains

Author: code-yeongyu

packages/agent/src/agent.ts

@@ -119,6 +119,7 @@ export interface AgentOptions {
 	timeoutMs?: number;
 	maxRetryDelayMs?: number;
 	toolExecution?: ToolExecutionMode;
+	abortServerSideFallback?: boolean;
 }
 
 class PendingMessageQueue {
@@ -219,6 +220,8 @@ export class Agent {
 	public maxRetryDelayMs?: number;
 	/** Tool execution strategy for assistant messages that contain multiple tool calls. */
 	public toolExecution: ToolExecutionMode;
+	/** Forwarded to the stream function; providers without server-side fallback ignore it. */
+	public abortServerSideFallback?: boolean;
 
 	constructor(options: AgentOptions) {
 		// Older compiled consumers may omit options or streamFn even though the current API requires them.
@@ -242,6 +245,7 @@ export class Agent {
 		this.timeoutMs = runtimeOptions.timeoutMs;
 		this.maxRetryDelayMs = runtimeOptions.maxRetryDelayMs;
 		this.toolExecution = runtimeOptions.toolExecution ?? "parallel";
+		this.abortServerSideFallback = runtimeOptions.abortServerSideFallback;
 	}
 
 	/**
@@ -495,6 +499,7 @@ export class Agent {
 			thinkingBudgets: this.thinkingBudgets,
 			timeoutMs: this.timeoutMs,
 			maxRetryDelayMs: this.maxRetryDelayMs,
+			abortServerSideFallback: this.abortServerSideFallback,
 			toolExecution: this.toolExecution,
 			beforeToolCall: this.beforeToolCall,
 			afterToolCall: this.afterToolCall,

packages/ai/src/api/anthropic-messages.ts

@@ -31,13 +31,20 @@ import type {
 	ToolResultMessage,
 } from "../types.ts";
 import { isVideoMimeType } from "../types.ts";
+import { combineAbortSignals } from "../utils/abort-signals.ts";
 import { splitDeferredTools } from "../utils/deferred-tools.ts";
 import { AssistantMessageEventStream } from "../utils/event-stream.ts";
 import { headersToRecord, providerHeadersToRecord } from "../utils/headers.ts";
 import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse.ts";
 import { getProviderEnvValue } from "../utils/provider-env.ts";
 import { retryProviderRequest } from "../utils/provider-retry.ts";
 import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts";
+import {
+	applyServerFallbackAbort,
+	parseServerFallbackReceipt,
+	parseStickyFallbackReceipt,
+	type ServerFallbackReceipt,
+} from "../utils/server-fallback-receipt.ts";
 import { isForcedToolChoiceUnsupportedError, omitToolChoiceParam } from "../utils/tool-choice-fallback.ts";
 
 import { resolveCloudflareBaseUrl } from "./cloudflare.ts";
@@ -945,6 +952,10 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 			timestamp: Date.now(),
 		};
 
+		const serverFallbackAbort = new AbortController();
+		let serverFallbackReceipt: ServerFallbackReceipt | undefined;
+		const combinedAbort = combineAbortSignals([options?.signal, serverFallbackAbort.signal]);
+		const requestSignal = combinedAbort.signal;
 		try {
 			let client: Anthropic;
 			let isOAuth: boolean;
@@ -1004,7 +1015,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 				const payloadRequestMetadata = extractPayloadRequestMetadata(params);
 				params = payloadRequestMetadata.params;
 				const requestOptions = {
-					...(options?.signal ? { signal: options.signal } : {}),
+					...(requestSignal ? { signal: requestSignal } : {}),
 					...(options?.timeoutMs !== undefined ? { timeout: options.timeoutMs } : {}),
 					maxRetries: 0,
 					...(payloadRequestMetadata.headers ? { headers: payloadRequestMetadata.headers } : {}),
@@ -1023,7 +1034,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 					throw error;
 				}
 			};
-			const { response } = await retryProviderRequest(
+			const { params: sentParams, response } = await retryProviderRequest(
 				async () => {
 					try {
 						return await createRequest();
@@ -1039,7 +1050,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 				{
 					maxRetries: options?.maxRetries,
 					maxRetryDelayMs: options?.maxRetryDelayMs,
-					signal: options?.signal,
+					signal: requestSignal,
 				},
 			);
 			await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
@@ -1052,7 +1063,7 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 				| (ProviderNativeContent & { partialJson?: string; index?: number });
 			const blocks = output.content as Block[];
 
-			for await (const event of iterateAnthropicEvents(response, options?.signal)) {
+			for await (const event of iterateAnthropicEvents(response, requestSignal)) {
 				if (event.type === "message_start") {
 					output.responseId = event.message.id;
 					// Capture initial token usage from message_start event
@@ -1066,7 +1077,25 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 					output.usage.totalTokens =
 						output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
 					calculateCost(model, output.usage);
+					const stickyReceipt =
+						options?.abortServerSideFallback === true
+							? parseStickyFallbackReceipt(event.message.usage, sentParams.model, event.message.model)
+							: undefined;
+					if (stickyReceipt !== undefined) {
+						serverFallbackReceipt = stickyReceipt;
+						serverFallbackAbort.abort();
+						break;
+					}
 				} else if (event.type === "content_block_start") {
+					const receipt =
+						options?.abortServerSideFallback === true
+							? parseServerFallbackReceipt(event.content_block)
+							: undefined;
+					if (receipt !== undefined) {
+						serverFallbackReceipt = receipt;
+						serverFallbackAbort.abort();
+						break;
+					}
 					if (event.content_block.type === "text") {
 						const block: Block = {
 							type: "text",
@@ -1252,13 +1281,19 @@ export const stream: StreamFunction<"anthropic-messages", AnthropicOptions> = (
 				throw new Error("Request was aborted");
 			}
 
+			if (serverFallbackReceipt !== undefined) {
+				applyServerFallbackAbort(output, serverFallbackReceipt);
+			}
+
 			if (output.stopReason === "aborted" || output.stopReason === "error") {
 				throw new Error(output.errorMessage || "An unknown error occurred");
 			}
 
+			combinedAbort.cleanup();
 			stream.push({ type: "done", reason: output.stopReason, message: output });
 			stream.end();
 		} catch (error) {
+			combinedAbort.cleanup();
 			for (const block of output.content) {
 				delete (block as { index?: number }).index;
 				// An aborted stream never reaches content_block_stop; keep whatever

packages/ai/src/api/simple-options.ts

@@ -145,6 +145,7 @@ export function buildBaseOptions(
 		temperature: options?.temperature,
 		maxTokens: clampMaxTokensToContext(model, context, options?.maxTokens ?? model.maxTokens),
 		signal: options?.signal,
+		abortServerSideFallback: options?.abortServerSideFallback,
 		apiKey: apiKey || options?.apiKey,
 		transport: options?.transport,
 		cacheRetention: options?.cacheRetention ?? model.cacheRetention,

packages/ai/src/changes.md

@@ -622,3 +622,20 @@
 ### Expected merge conflict zones
 
 - LOW: `src/types.ts` `ThinkingContent` interface.
+
+## Client abort on Anthropic server-side fallback receipts (2026-07-25)
+
+### What changed
+
+- `utils/server-fallback-receipt.ts`: new module parsing Anthropic's `fallback` content block and the `fallback_message` entry in `usage.iterations`, plus the refusal-shaped rewrite applied to an aborted turn.
+- `types.ts`: `StreamOptions.abortServerSideFallback` (opt-in), inherited by `SimpleStreamOptions` and `AnthropicOptions`; `api/simple-options.ts` forwards it through `buildBaseOptions`.
+- `api/anthropic-messages.ts`: a provider-local `AbortController`, merged with the caller signal through `combineAbortSignals`, is passed to the request and the SSE iterator. A receipt block or a `fallback_message` usage entry aborts it and finalizes the turn as `{stopReason:"error", stopDetails:{type:"refusal"}}` with empty content plus `server_fallback_aborted` and `billing_incomplete_after_client_abort` diagnostics. A caller abort is checked first and always wins.
+
+### Why the extension system couldn't handle this
+
+Detection has to happen inside the Anthropic SSE loop while the stream is still open; nothing outside the provider can stop reading a response mid-flight.
+
+### Expected merge conflict zones
+
+- MEDIUM: `api/anthropic-messages.ts` streaming event loop and request-option construction.
+- LOW: `types.ts` `StreamOptions`, `api/simple-options.ts` `buildBaseOptions` field list, `index.ts` export list.

packages/ai/src/index.ts

@@ -60,6 +60,7 @@ export * from "./utils/json-parse.ts";
 export { extractOpenAiCodexAccountId } from "./utils/openai-codex-auth.ts";
 export * from "./utils/overflow.ts";
 export * from "./utils/retry.ts";
+export * from "./utils/server-fallback-receipt.ts";
 export * from "./utils/stop-details.ts";
 export { contentText } from "./utils/text.ts";
 export * from "./utils/tool-pair-repair.ts";

packages/ai/src/types.ts

@@ -126,6 +126,15 @@ export interface StreamOptions {
 	temperature?: number;
 	maxTokens?: number;
 	signal?: AbortSignal;
+	/**
+	 * Abort the request when the provider reports that a safety classifier
+	 * declined the requested model and a substitute model served the turn
+	 * instead (Anthropic `server-side-fallback-*` betas). The substitute's output
+	 * is billed but was never requested, so aborting keeps model selection with
+	 * the caller. Providers without server-side fallback ignore this.
+	 * Default: undefined (honor the substituted response).
+	 */
+	abortServerSideFallback?: boolean;
 	apiKey?: string;
 	/**
 	 * Preferred transport for providers that support multiple transports.

packages/ai/src/utils/server-fallback-receipt.ts

@@ -0,0 +1,82 @@
+import type { AssistantMessage } from "../types.ts";
+import { appendAssistantMessageDiagnostic } from "./diagnostics.ts";
+
+export const SERVER_FALLBACK_ABORTED_DIAGNOSTIC = "server_fallback_aborted";
+export const BILLING_INCOMPLETE_DIAGNOSTIC = "billing_incomplete_after_client_abort";
+
+export interface ServerFallbackReceipt {
+	readonly from: string;
+	readonly to: string;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+	return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function readModel(value: unknown): string | undefined {
+	if (!isRecord(value)) return undefined;
+	return typeof value.model === "string" && value.model.length > 0 ? value.model : undefined;
+}
+
+/**
+ * Anthropic marks a classifier handoff with `{type:"fallback", from:{model}, to:{model}}`
+ * (`server-side-fallback-*` betas). Returns undefined for anything else, so an
+ * unrelated or malformed provider-native block falls through to normal handling.
+ */
+export function parseServerFallbackReceipt(block: unknown): ServerFallbackReceipt | undefined {
+	if (!isRecord(block) || block.type !== "fallback") return undefined;
+	const from = readModel(block.from);
+	const to = readModel(block.to);
+	return from !== undefined && to !== undefined ? { from, to } : undefined;
+}
+
+/**
+ * Sticky routing serves later turns of a fallen-back conversation from the
+ * substitute model with no `fallback` block at all; a `fallback_message` entry
+ * in `usage.iterations` is the documented signal. A served-model string
+ * comparison is deliberately NOT used: gateways and Bedrock-style endpoints
+ * rewrite model ids, so a mismatch is not evidence of a fallback.
+ */
+export function parseStickyFallbackReceipt(
+	usage: unknown,
+	requestedModel: string,
+	servedModel?: string,
+): ServerFallbackReceipt | undefined {
+	if (!isRecord(usage) || !Array.isArray(usage.iterations)) return undefined;
+	for (let index = usage.iterations.length - 1; index >= 0; index--) {
+		const entry: unknown = usage.iterations[index];
+		if (!isRecord(entry) || entry.type !== "fallback_message") continue;
+		const entryModel = typeof entry.model === "string" && entry.model.length > 0 ? entry.model : undefined;
+		const to = entryModel ?? servedModel;
+		return to !== undefined ? { from: requestedModel, to } : undefined;
+	}
+	return undefined;
+}
+
+export function serverFallbackRefusalExplanation(receipt: ServerFallbackReceipt): string {
+	return `Server-side fallback (${receipt.from} -> ${receipt.to}) aborted by client policy`;
+}
+
+/**
+ * Rewrites an aborted turn into the same shape a pre-output classifier refusal
+ * has, so `isClassifierRefusal()` routes it through the caller's fallback chain.
+ * Content is dropped: the substitute model's partial output was never requested,
+ * and replaying a half-turn containing a fallback marker breaks Anthropic replay.
+ */
+export function applyServerFallbackAbort(message: AssistantMessage, receipt: ServerFallbackReceipt): void {
+	const explanation = serverFallbackRefusalExplanation(receipt);
+	message.content = [];
+	message.stopReason = "error";
+	message.stopDetails = { type: "refusal", explanation };
+	message.errorMessage = explanation;
+	appendAssistantMessageDiagnostic(message, {
+		type: SERVER_FALLBACK_ABORTED_DIAGNOSTIC,
+		timestamp: Date.now(),
+		details: { from: receipt.from, to: receipt.to },
+	});
+	appendAssistantMessageDiagnostic(message, {
+		type: BILLING_INCOMPLETE_DIAGNOSTIC,
+		timestamp: Date.now(),
+		details: { reason: "per-attempt usage does not arrive after a client abort" },
+	});
+}