feat(ai): added Anthropic raw SSE decoding with malformed payload repair

- Added a runtime export for `OpenAICodexResponsesOptions` and `AnthropicCompat` capability flags.
- Added Anthropic raw SSE decoding, envelope filtering, repaired JSON parsing, and thinking/eager-input support.
- Added `repairJson`/`parseJsonWithRepair` and OAuth `postJson()` with `formatErrorDetails()` for richer stream/auth failures.
- Changed OpenAI Codex streaming to default verbosity to `low`, send `previous_response_id` on continuations, and record websocket debug stats.
- Added raw SSE and websocket tests for unknown trace events, malformed payload repair, continuation IDs, and debug counters.

packages/ai/CHANGELOG.md

@@ -2,6 +2,23 @@
 
 ## [Unreleased]
 
+### Added
+
+- Added `thinkingDisplay` option to Anthropic options to control whether adaptive and explicit reasoning is returned as `summarized` or `omitted`
+- Added Anthropic model compatibility flags `supportsEagerToolInputStreaming` and `supportsLongCacheRetention` for API-capability-specific request behavior
+
+### Changed
+
+- Changed Anthropic cache retention handling so `cacheRetention: "long"` now uses `ttl: "1h"` only for canonical Anthropic endpoints with long-cache support
+- Changed Anthropic tool schema generation to include `eager_input_streaming` only on models that advertise support
+- Changed Anthropic OAuth login flow to include browser fallback guidance and richer error context when token exchange or refresh fails
+
+### Fixed
+
+- Fixed Anthropic stream handling to parse raw SSE envelopes directly, ignore unrelated events, and repair malformed JSON in SSE payloads
+- Fixed Anthropic streaming to emit an explicit error when the SSE stream ends without a `message_stop` event
+- Fixed OpenAI Codex websocket continuations to send true `previous_response_id` deltas for `store: false` transcripts, expose request stats, and default text verbosity to `low` unless explicitly overridden.
+
 ## [14.5.14] - 2026-05-01
 ### Added
 
@@ -2208,4 +2225,4 @@ _Dedicated to Peter's shoulder ([@steipete](https://twitter.com/steipete))_
 
 ## [0.9.4] - 2025-11-26
 
-Initial release with multi-provider LLM support.
+Initial release with multi-provider LLM support.

packages/ai/src/index.ts

@@ -18,7 +18,7 @@ export * from "./providers/google-gemini-headers";
 export type * from "./providers/google-vertex";
 export * from "./providers/kimi";
 export * from "./providers/ollama";
-export type { OpenAICodexResponsesOptions } from "./providers/openai-codex-responses";
+export * from "./providers/openai-codex-responses";
 export * from "./providers/openai-completions";
 export * from "./providers/openai-responses";
 export * from "./providers/synthetic";

packages/ai/src/providers/anthropic.ts

@@ -6,6 +6,7 @@ import type {
 	ContentBlockParam,
 	MessageCreateParamsStreaming,
 	MessageParam,
+	RawMessageStreamEvent,
 } from "@anthropic-ai/sdk/resources/messages";
 import { $env, abortableSleep, isEnoent } from "@oh-my-pi/pi-utils";
 import { hasOpus47ApiRestrictions, mapEffortToAnthropicAdaptiveEffort } from "../model-thinking";
@@ -38,7 +39,7 @@ import { AssistantMessageEventStream } from "../utils/event-stream";
 import { isFoundryEnabled } from "../utils/foundry";
 import { finalizeErrorMessage, type RawHttpRequestDump, rewriteCopilotError } from "../utils/http-inspector";
 import { createWatchdog, getStreamFirstEventTimeoutMs } from "../utils/idle-iterator";
-import { parseStreamingJson } from "../utils/json-parse";
+import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse";
 import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot";
 import { notifyProviderResponse } from "../utils/provider-response";
 import { extractHttpStatusFromError, isCopilotRetryableError, isUnexpectedSocketCloseMessage } from "../utils/retry";
@@ -57,6 +58,7 @@ export type AnthropicHeaderOptions = {
 	extraBetas?: string[];
 	stream?: boolean;
 	modelHeaders?: Record<string, string>;
+	isCloudflareAiGateway?: boolean;
 };
 
 export function normalizeAnthropicBaseUrl(baseUrl?: string): string | undefined {
@@ -88,6 +90,9 @@ const claudeCodeBetaDefaults = [
 	"context-management-2025-06-27",
 	"prompt-caching-scope-2026-01-05",
 ];
+const fineGrainedToolStreamingBeta = "fine-grained-tool-streaming-2025-05-14";
+const interleavedThinkingBeta = "interleaved-thinking-2025-05-14";
+
 function getHeaderCaseInsensitive(headers: Record<string, string> | undefined, headerName: string): string | undefined {
 	if (!headers) return undefined;
 	const normalizedName = headerName.toLowerCase();
@@ -131,6 +136,16 @@ export function buildAnthropicHeaders(options: AnthropicHeaderOptions): Record<s
 		Object.entries(options.modelHeaders ?? {}).filter(([key]) => !enforcedHeaderKeys.has(key.toLowerCase())),
 	);
 
+	if (options.isCloudflareAiGateway) {
+		return {
+			...modelHeaders,
+			Accept: acceptHeader,
+			...sharedHeaders,
+			"Anthropic-Beta": betaHeader,
+			"cf-aig-authorization": `Bearer ${options.apiKey}`,
+		};
+	}
+
 	if (oauthToken) {
 		const incomingUserAgent = getHeaderCaseInsensitive(options.modelHeaders, "User-Agent");
 		const userAgent = isClaudeCodeClientUserAgent(incomingUserAgent)
@@ -235,6 +250,7 @@ function dropAnthropicStrictTools(params: MessageCreateParamsStreaming): void {
 }
 
 function getCacheControl(
+	model: Model<"anthropic-messages">,
 	baseUrl: string,
 	cacheRetention?: CacheRetention,
 ): { retention: CacheRetention; cacheControl?: AnthropicCacheControl } {
@@ -242,7 +258,10 @@ function getCacheControl(
 	if (retention === "none") {
 		return { retention };
 	}
-	const ttl = retention === "long" && baseUrl.includes("api.anthropic.com") ? "1h" : undefined;
+	const ttl =
+		retention === "long" && isAnthropicApiBaseUrl(baseUrl) && getAnthropicCompat(model).supportsLongCacheRetention
+			? "1h"
+			: undefined;
 	return {
 		retention,
 		cacheControl: { type: "ephemeral", ...(ttl && { ttl }) },
@@ -312,6 +331,7 @@ const enforcedHeaderKeys = new Set(
 		"X-App",
 		"Authorization",
 		"X-Api-Key",
+		"cf-aig-authorization",
 	].map(key => key.toLowerCase()),
 );
 
@@ -424,6 +444,7 @@ function convertContentBlocks(content: (TextContent | ImageContent)[]):
 }
 
 export type AnthropicEffort = "low" | "medium" | "high" | "xhigh" | "max";
+export type AnthropicThinkingDisplay = "summarized" | "omitted";
 
 export interface AnthropicOptions extends StreamOptions {
 	/**
@@ -452,6 +473,12 @@ export interface AnthropicOptions extends StreamOptions {
 	 * Converted to adaptive effort when effort is not explicitly provided.
 	 */
 	reasoning?: SimpleStreamOptions["reasoning"];
+	/**
+	 * Controls how Anthropic returns thinking content when the selected thinking
+	 * transport supports a display option. Defaults to "summarized" where the
+	 * API accepts it.
+	 */
+	thinkingDisplay?: AnthropicThinkingDisplay;
 	interleavedThinking?: boolean;
 	toolChoice?: "auto" | "any" | "none" | { type: "tool"; name: string };
 	betas?: string[] | string;
@@ -474,12 +501,13 @@ export type AnthropicClientOptionsArgs = {
 	headers?: Record<string, string>;
 	dynamicHeaders?: Record<string, string>;
 	isOAuth?: boolean;
+	hasTools?: boolean;
 };
 
 export type AnthropicClientOptionsResult = {
 	isOAuthToken: boolean;
 	apiKey: string | null;
-	authToken?: string;
+	authToken?: string | null;
 	baseURL?: string;
 	maxRetries: number;
 	dangerouslyAllowBrowser: boolean;
@@ -624,6 +652,248 @@ function mergeHeaders(...headerSources: (Record<string, string> | undefined)[]):
 // We surface the resulting provider error ourselves, so keep the SDK quiet.
 const ANTHROPIC_SDK_LOG_LEVEL = "off" as const;
 
+interface ServerSentEvent {
+	event: string | null;
+	data: string;
+	raw: string[];
+}
+
+interface SseDecoderState {
+	event: string | null;
+	data: string[];
+	raw: string[];
+}
+
+const ANTHROPIC_MESSAGE_EVENTS: ReadonlySet<string> = new Set([
+	"message_start",
+	"message_delta",
+	"message_stop",
+	"content_block_start",
+	"content_block_delta",
+	"content_block_stop",
+]);
+
+function flushSseEvent(state: SseDecoderState): ServerSentEvent | null {
+	if (!state.event && state.data.length === 0) {
+		return null;
+	}
+
+	const event: ServerSentEvent = {
+		event: state.event,
+		data: state.data.join("\n"),
+		raw: [...state.raw],
+	};
+	state.event = null;
+	state.data = [];
+	state.raw = [];
+	return event;
+}
+
+function decodeSseLine(line: string, state: SseDecoderState): ServerSentEvent | null {
+	if (line === "") {
+		return flushSseEvent(state);
+	}
+
+	state.raw.push(line);
+	if (line.startsWith(":")) {
+		return null;
+	}
+
+	const delimiterIndex = line.indexOf(":");
+	const fieldName = delimiterIndex === -1 ? line : line.slice(0, delimiterIndex);
+	let value = delimiterIndex === -1 ? "" : line.slice(delimiterIndex + 1);
+	if (value.startsWith(" ")) {
+		value = value.slice(1);
+	}
+
+	if (fieldName === "event") {
+		state.event = value;
+	} else if (fieldName === "data") {
+		state.data.push(value);
+	}
+
+	return null;
+}
+
+function nextLineBreakIndex(text: string): number {
+	const carriageReturnIndex = text.indexOf("\r");
+	const newlineIndex = text.indexOf("\n");
+	if (carriageReturnIndex === -1) {
+		return newlineIndex;
+	}
+	if (newlineIndex === -1) {
+		return carriageReturnIndex;
+	}
+	return Math.min(carriageReturnIndex, newlineIndex);
+}
+
+function consumeLine(text: string): { line: string; rest: string } | null {
+	const lineBreakIndex = nextLineBreakIndex(text);
+	if (lineBreakIndex === -1) {
+		return null;
+	}
+
+	let nextIndex = lineBreakIndex + 1;
+	if (text[lineBreakIndex] === "\r" && text[nextIndex] === "\n") {
+		nextIndex += 1;
+	}
+
+	return {
+		line: text.slice(0, lineBreakIndex),
+		rest: text.slice(nextIndex),
+	};
+}
+
+async function* iterateSseMessages(
+	body: ReadableStream<Uint8Array>,
+	signal?: AbortSignal,
+): AsyncGenerator<ServerSentEvent> {
+	const reader = body.getReader();
+	const decoder = new TextDecoder();
+	const state: SseDecoderState = { event: null, data: [], raw: [] };
+	let buffer = "";
+
+	try {
+		while (true) {
+			if (signal?.aborted) {
+				throw new Error("Request was aborted");
+			}
+
+			const { value, done } = await reader.read();
+			if (done) {
+				break;
+			}
+
+			buffer += decoder.decode(value, { stream: true });
+			let consumed = consumeLine(buffer);
+			while (consumed) {
+				buffer = consumed.rest;
+				const event = decodeSseLine(consumed.line, state);
+				if (event) {
+					yield event;
+				}
+				consumed = consumeLine(buffer);
+			}
+		}
+
+		buffer += decoder.decode();
+		let consumed = consumeLine(buffer);
+		while (consumed) {
+			buffer = consumed.rest;
+			const event = decodeSseLine(consumed.line, state);
+			if (event) {
+				yield event;
+			}
+			consumed = consumeLine(buffer);
+		}
+
+		if (buffer.length > 0) {
+			const event = decodeSseLine(buffer, state);
+			if (event) {
+				yield event;
+			}
+		}
+
+		const trailingEvent = flushSseEvent(state);
+		if (trailingEvent) {
+			yield trailingEvent;
+		}
+	} finally {
+		reader.releaseLock();
+	}
+}
+
+async function* iterateAnthropicEvents(
+	response: Response,
+	signal?: AbortSignal,
+): AsyncGenerator<RawMessageStreamEvent> {
+	if (!response.body) {
+		throw new Error("Attempted to iterate over an Anthropic response with no body");
+	}
+
+	let sawMessageStart = false;
+	let sawMessageEnd = false;
+
+	for await (const sse of iterateSseMessages(response.body, signal)) {
+		if (sse.event === "error") {
+			throw new Error(sse.data);
+		}
+
+		if (!ANTHROPIC_MESSAGE_EVENTS.has(sse.event ?? "")) {
+			continue;
+		}
+
+		try {
+			const event = parseJsonWithRepair<RawMessageStreamEvent>(sse.data);
+			if (event.type === "message_start") {
+				sawMessageStart = true;
+			} else if (event.type === "message_stop") {
+				sawMessageEnd = true;
+			}
+			yield event;
+		} catch (error) {
+			const message = error instanceof Error ? error.message : String(error);
+			throw new Error(
+				`Could not parse Anthropic SSE event ${sse.event}: ${message}; data=${sse.data}; raw=${sse.raw.join("\\n")}`,
+			);
+		}
+	}
+
+	if (sawMessageStart && !sawMessageEnd) {
+		throw createAnthropicStreamEnvelopeError("stream ended before message_stop");
+	}
+}
+
+type AnthropicRawResponseRequest = {
+	asResponse(): Promise<Response>;
+};
+
+function hasAnthropicRawResponseRequest(request: unknown): request is AnthropicRawResponseRequest {
+	return isRecord(request) && typeof request.asResponse === "function";
+}
+
+type AnthropicStreamWithResponseRequest = {
+	withResponse(): Promise<{
+		data: AsyncIterable<RawMessageStreamEvent>;
+		response: Response;
+		request_id: string | null;
+	}>;
+};
+
+function hasAnthropicStreamWithResponseRequest(request: unknown): request is AnthropicStreamWithResponseRequest {
+	return isRecord(request) && typeof request.withResponse === "function";
+}
+
+async function getAnthropicStreamResponse(
+	request: unknown,
+	signal?: AbortSignal,
+): Promise<{ events: AsyncIterable<RawMessageStreamEvent>; response: Response; requestId: string | null }> {
+	if (hasAnthropicRawResponseRequest(request)) {
+		const response = await request.asResponse();
+		return {
+			events: iterateAnthropicEvents(response, signal),
+			response,
+			requestId: response.headers.get("request-id"),
+		};
+	}
+	if (hasAnthropicStreamWithResponseRequest(request)) {
+		const { data, response, request_id } = await request.withResponse();
+		return { events: data, response, requestId: request_id };
+	}
+	throw new Error("Anthropic SDK request did not expose a stream response");
+}
+
+function getAnthropicCompat(
+	model: Model<"anthropic-messages">,
+): Required<NonNullable<Model<"anthropic-messages">["compat"]>> {
+	return {
+		disableStrictTools: model.compat?.disableStrictTools ?? false,
+		disableAdaptiveThinking: model.compat?.disableAdaptiveThinking ?? false,
+		supportsEagerToolInputStreaming: model.compat?.supportsEagerToolInputStreaming ?? true,
+		supportsLongCacheRetention: model.compat?.supportsLongCacheRetention ?? true,
+	};
+}
+
 const PROVIDER_MAX_RETRIES = 3;
 const PROVIDER_BASE_DELAY_MS = 2000;
 
@@ -789,6 +1059,7 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (
 					headers: options?.headers,
 					dynamicHeaders: copilotDynamicHeaders?.headers,
 					isOAuth: options?.isOAuth,
+					hasTools: !!context.tools?.length,
 				});
 				client = created.client;
 				isOAuthToken = created.isOAuthToken;
@@ -844,8 +1115,12 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (
 				let streamedReplayUnsafeContent = false;
 
 				try {
-					const { data: anthropicStream, response, request_id } = await anthropicRequest.withResponse();
-					await notifyProviderResponse(options, response, model, request_id);
+					const {
+						events: anthropicStream,
+						response,
+						requestId,
+					} = await getAnthropicStreamResponse(anthropicRequest, requestSignal);
+					await notifyProviderResponse(options, response, model, requestId);
 					const firstEventWatchdog = createWatchdog(firstEventTimeoutMs, () =>
 						activeAbortTracker.abortLocally(firstEventTimeoutAbortError),
 					);
@@ -1199,9 +1474,12 @@ export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): A
 		interleavedThinking = true,
 		headers,
 		dynamicHeaders,
+		hasTools = false,
 		isOAuth,
 	} = args;
+	const compat = getAnthropicCompat(model);
 	const needsInterleavedBeta = interleavedThinking && !supportsAdaptiveThinkingDisplay(model.id);
+	const needsFineGrainedToolStreamingBeta = hasTools && !compat.supportsEagerToolInputStreaming;
 	const oauthToken = isOAuth ?? isAnthropicOAuthToken(apiKey);
 	const baseUrl = resolveAnthropicBaseUrl(model, apiKey);
 	const foundryCustomHeaders = resolveAnthropicCustomHeaders(model);
@@ -1209,6 +1487,9 @@ export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): A
 	if (model.provider === "github-copilot") {
 		const copilotApiKey = parseGitHubCopilotApiKey(apiKey).accessToken;
 		const betaFeatures = [...extraBetas];
+		if (needsFineGrainedToolStreamingBeta) {
+			betaFeatures.push(fineGrainedToolStreamingBeta);
+		}
 		const defaultHeaders = mergeHeaders(
 			{
 				Accept: stream ? "text/event-stream" : "application/json",
@@ -1235,8 +1516,11 @@ export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): A
 	}
 
 	const betaFeatures = [...extraBetas];
+	if (needsFineGrainedToolStreamingBeta) {
+		betaFeatures.push(fineGrainedToolStreamingBeta);
+	}
 	if (needsInterleavedBeta) {
-		betaFeatures.push("interleaved-thinking-2025-05-14");
+		betaFeatures.push(interleavedThinkingBeta);
 	}
 
 	const defaultHeaders = buildAnthropicHeaders({
@@ -1246,8 +1530,22 @@ export function buildAnthropicClientOptions(args: AnthropicClientOptionsArgs): A
 		extraBetas: betaFeatures,
 		stream,
 		modelHeaders: mergeHeaders(model.headers, foundryCustomHeaders, headers, dynamicHeaders),
+		isCloudflareAiGateway: model.provider === "cloudflare-ai-gateway",
 	});
 
+	if (model.provider === "cloudflare-ai-gateway") {
+		return {
+			isOAuthToken: false,
+			apiKey: null,
+			authToken: null,
+			baseURL: baseUrl,
+			maxRetries: 5,
+			dangerouslyAllowBrowser: true,
+			defaultHeaders,
+			logLevel: ANTHROPIC_SDK_LOG_LEVEL,
+		};
+	}
+
 	return {
 		isOAuthToken: oauthToken,
 		apiKey: oauthToken ? null : apiKey,
@@ -1532,7 +1830,7 @@ function buildParams(
 	options?: AnthropicOptions,
 	disableStrictTools = false,
 ): MessageCreateParamsStreaming {
-	const { cacheControl } = getCacheControl(baseUrl, options?.cacheRetention);
+	const { cacheControl } = getCacheControl(model, baseUrl, options?.cacheRetention);
 	const params: AnthropicSamplingParams = {
 		model: model.id,
 		messages: convertAnthropicMessages(context.messages, model, isOAuthToken),
@@ -1558,6 +1856,7 @@ function buildParams(
 			context.tools,
 			isOAuthToken,
 			disableStrictTools || model.provider === "github-copilot",
+			getAnthropicCompat(model).supportsEagerToolInputStreaming,
 		);
 	}
 
@@ -1567,14 +1866,14 @@ function buildParams(
 		const effort =
 			options.effort ?? (requestedEffort ? mapEffortToAnthropicAdaptiveEffort(model, requestedEffort) : undefined);
 
-		const disableAdaptiveThinking = model.compat?.disableAdaptiveThinking ?? false;
-		if (mode === "anthropic-adaptive" && !disableAdaptiveThinking) {
+		const compat = getAnthropicCompat(model);
+		if (mode === "anthropic-adaptive" && !compat.disableAdaptiveThinking) {
 			// Starting with Claude Opus 4.7, adaptive thinking content is omitted from the
 			// response by default. Opt into summarized reasoning so thinking deltas keep
 			// streaming with human-readable content for callers that rely on it.
-			const adaptive: { type: "adaptive"; display?: "summarized" | "omitted" } = { type: "adaptive" };
+			const adaptive: { type: "adaptive"; display?: AnthropicThinkingDisplay } = { type: "adaptive" };
 			if (supportsAdaptiveThinkingDisplay(model.id)) {
-				adaptive.display = "summarized";
+				adaptive.display = options.thinkingDisplay ?? "summarized";
 			}
 			params.thinking = adaptive as typeof params.thinking;
 			if (effort) {
@@ -1586,7 +1885,8 @@ function buildParams(
 			params.thinking = {
 				type: "enabled",
 				budget_tokens: options.thinkingBudgetTokens || 1024,
-			};
+				display: options.thinkingDisplay ?? "summarized",
+			} as typeof params.thinking;
 			if (mode === "anthropic-budget-effort" && effort) {
 				params.output_config = { effort } as typeof params.output_config;
 			}
@@ -2108,7 +2408,12 @@ function buildAnthropicToolSchemaPlans(tools: Tool[], disableStrictTools = false
 	return plans;
 }
 
-function convertTools(tools: Tool[], isOAuthToken: boolean, disableStrictTools = false): Anthropic.Messages.Tool[] {
+function convertTools(
+	tools: Tool[],
+	isOAuthToken: boolean,
+	disableStrictTools = false,
+	supportsEagerToolInputStreaming = true,
+): Anthropic.Messages.Tool[] {
 	if (!tools) return [];
 	const schemaPlans = buildAnthropicToolSchemaPlans(tools, disableStrictTools);
 
@@ -2118,6 +2423,7 @@ function convertTools(tools: Tool[], isOAuthToken: boolean, disableStrictTools =
 			name: isOAuthToken ? applyClaudeToolPrefix(tool.name) : tool.name,
 			description: tool.description || "",
 			input_schema: plan.inputSchema,
+			...(supportsEagerToolInputStreaming ? { eager_input_streaming: true } : {}),
 			...(plan.strict ? { strict: true } : {}),
 		};
 	});

packages/ai/src/providers/openai-codex-responses.ts

@@ -101,10 +101,19 @@ type CodexTransport = "sse" | "websocket";
 type CodexEventItem = ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | ResponseCustomToolCall;
 type CodexOutputBlock = ThinkingContent | TextContent | (ToolCall & { partialJson: string });
 
+export interface OpenAICodexWebSocketDebugStats {
+	fullContextRequests: number;
+	deltaRequests: number;
+	lastInputItems: number;
+	lastDeltaInputItems?: number;
+	lastPreviousResponseId?: string;
+}
+
 type CodexWebSocketSessionState = {
 	disableWebsocket: boolean;
 	lastRequest?: RequestBody;
 	lastResponseId?: string;
+	lastResponseItems?: InputItem[];
 	canAppend: boolean;
 	turnState?: string;
 	modelsEtag?: string;
@@ -114,6 +123,7 @@ type CodexWebSocketSessionState = {
 	fallbackCount: number;
 	lastFallbackAt?: number;
 	prewarmed: boolean;
+	stats: OpenAICodexWebSocketDebugStats;
 };
 
 interface CodexProviderSessionState extends ProviderSessionState {
@@ -370,6 +380,53 @@ function createEmptyUsage(): AssistantMessage["usage"] {
 	};
 }
 
+function getCodexUserAgent(): string {
+	return `pi/${packageJson.version} (${os.platform()} ${os.release()}; ${os.arch()})`;
+}
+
+function getCodexServiceTierCostMultiplier(
+	model: Pick<Model<"openai-codex-responses">, "id">,
+	serviceTier: ServiceTier | "default" | undefined,
+): number {
+	switch (serviceTier) {
+		case "flex":
+			return 0.5;
+		case "priority":
+			return model.id === "gpt-5.5" ? 2.5 : 2;
+		default:
+			return 1;
+	}
+}
+
+function resolveCodexCostServiceTier(res: unknown, req?: unknown): ServiceTier | "default" | undefined {
+	switch (res) {
+		case "flex":
+			return "flex";
+		case "priority":
+			return "priority";
+		default:
+			if (req === "flex" || req === "priority") {
+				return req;
+			}
+			return "default";
+	}
+}
+
+function applyCodexServiceTierPricing(
+	model: Pick<Model<"openai-codex-responses">, "id">,
+	usage: AssistantMessage["usage"],
+	resTier: unknown,
+	reqTier: unknown,
+): void {
+	const multiplier = getCodexServiceTierCostMultiplier(model, resolveCodexCostServiceTier(resTier, reqTier));
+	if (multiplier === 1) return;
+	usage.cost.input *= multiplier;
+	usage.cost.output *= multiplier;
+	usage.cost.cacheRead *= multiplier;
+	usage.cost.cacheWrite *= multiplier;
+	usage.cost.total = usage.cost.input + usage.cost.output + usage.cost.cacheRead + usage.cost.cacheWrite;
+}
+
 function createAssistantOutput(model: Model<"openai-codex-responses">): AssistantMessage {
 	return {
 		role: "assistant",
@@ -727,6 +784,7 @@ async function processCodexResponseStream(
 					rawEvent,
 					firstTokenTime,
 				});
+				if (runtime.sawTerminalEvent) break;
 			}
 			return { firstTokenTime };
 		} catch (error) {
@@ -1122,6 +1180,7 @@ function handleResponseCompleted(
 					output_tokens_details?: { reasoning_tokens?: number };
 				};
 				status?: string;
+				service_tier?: ServiceTier | "default";
 			};
 		}
 	).response;
@@ -1148,11 +1207,13 @@ function handleResponseCompleted(
 		state.lastRequest = structuredCloneJSON(runtime.requestBodyForState);
 		if (typeof response?.id === "string" && response.id.length > 0) {
 			state.lastResponseId = response.id;
+			state.lastResponseItems = stripInputItemIds(structuredCloneJSON(runtime.nativeOutputItems));
 		}
 		state.canAppend = rawEvent.type === "response.done";
 	}
 
 	calculateCost(model, output.usage);
+	applyCodexServiceTierPricing(model, output.usage, response?.service_tier, runtime.requestBodyForState.service_tier);
 	output.stopReason = mapOpenAIResponsesStopReason(response?.status as OpenAI.Responses.ResponseStatus | undefined);
 	if (output.content.some(block => block.type === "toolCall") && output.stopReason === "stop") {
 		output.stopReason = "toolUse";
@@ -1525,6 +1586,11 @@ function getCodexWebSocketSessionState(
 		canAppend: false,
 		fallbackCount: 0,
 		prewarmed: false,
+		stats: {
+			fullContextRequests: 0,
+			deltaRequests: 0,
+			lastInputItems: 0,
+		},
 	};
 	providerSessionState.webSocketSessions.set(sessionKey, created);
 	return created;
@@ -1534,6 +1600,7 @@ function resetCodexWebSocketAppendState(state: CodexWebSocketSessionState): void
 	state.canAppend = false;
 	state.lastRequest = undefined;
 	state.lastResponseId = undefined;
+	state.lastResponseItems = undefined;
 }
 
 function resetCodexSessionMetadata(state: CodexWebSocketSessionState): void {
@@ -1575,6 +1642,37 @@ export interface OpenAICodexTransportDetails {
 	lastFallbackAt?: number;
 }
 
+function getCodexWebSocketStateForPublicSession(
+	model: Model<"openai-codex-responses">,
+	options:
+		| {
+				sessionId?: string;
+				baseUrl?: string;
+				providerSessionState?: Map<string, ProviderSessionState>;
+		  }
+		| undefined,
+): CodexWebSocketSessionState | undefined {
+	const baseUrl = options?.baseUrl || model.baseUrl || CODEX_BASE_URL;
+	const providerSessionState = getCodexProviderSessionState(options?.providerSessionState);
+	const publicSessionKey = getCodexPublicSessionKey(options?.sessionId, model, baseUrl);
+	const privateSessionKey = publicSessionKey
+		? providerSessionState?.webSocketPublicToPrivate.get(publicSessionKey)
+		: undefined;
+	return privateSessionKey ? providerSessionState?.webSocketSessions.get(privateSessionKey) : undefined;
+}
+
+export function getOpenAICodexWebSocketDebugStats(
+	model: Model<"openai-codex-responses">,
+	options?: {
+		sessionId?: string;
+		baseUrl?: string;
+		providerSessionState?: Map<string, ProviderSessionState>;
+	},
+): OpenAICodexWebSocketDebugStats | undefined {
+	const stats = getCodexWebSocketStateForPublicSession(model, options)?.stats;
+	return stats ? { ...stats } : undefined;
+}
+
 export function getOpenAICodexTransportDetails(
 	model: Model<"openai-codex-responses">,
 	options?: {
@@ -1584,17 +1682,11 @@ export function getOpenAICodexTransportDetails(
 		providerSessionState?: Map<string, ProviderSessionState>;
 	},
 ): OpenAICodexTransportDetails {
-	const baseUrl = options?.baseUrl || model.baseUrl || CODEX_BASE_URL;
 	const websocketPreferred =
 		options?.preferWebsockets === false
 			? false
 			: isCodexWebSocketEnvEnabled() || options?.preferWebsockets === true || model.preferWebsockets === true;
-	const providerSessionState = getCodexProviderSessionState(options?.providerSessionState);
-	const publicSessionKey = getCodexPublicSessionKey(options?.sessionId, model, baseUrl);
-	const privateSessionKey = publicSessionKey
-		? providerSessionState?.webSocketPublicToPrivate.get(publicSessionKey)
-		: undefined;
-	const state = privateSessionKey ? providerSessionState?.webSocketSessions.get(privateSessionKey) : undefined;
+	const state = getCodexWebSocketStateForPublicSession(model, options);
 
 	return {
 		websocketPreferred,
@@ -1609,41 +1701,70 @@ export function getOpenAICodexTransportDetails(
 	};
 }
 
-function buildAppendInput(previous: RequestBody | undefined, current: RequestBody): InputItem[] | null {
+function buildAppendInput(
+	previous: RequestBody | undefined,
+	previousResponseItems: InputItem[] | undefined,
+	current: RequestBody,
+): InputItem[] | null {
 	if (!previous) return null;
 	if (!Array.isArray(previous.input) || !Array.isArray(current.input)) return null;
-	if (current.input.length <= previous.input.length) return null;
 	const previousWithoutInput = { ...previous, input: undefined };
 	const currentWithoutInput = { ...current, input: undefined };
 	if (JSON.stringify(previousWithoutInput) !== JSON.stringify(currentWithoutInput)) {
 		return null;
 	}
-	for (let index = 0; index < previous.input.length; index += 1) {
-		if (JSON.stringify(previous.input[index]) !== JSON.stringify(current.input[index])) {
+	const baseline = [...previous.input, ...(previousResponseItems ?? [])];
+	if (current.input.length <= baseline.length) return null;
+	for (let index = 0; index < baseline.length; index += 1) {
+		if (JSON.stringify(baseline[index]) !== JSON.stringify(current.input[index])) {
 			return null;
 		}
 	}
-	return current.input.slice(previous.input.length) as InputItem[];
+	return current.input.slice(baseline.length) as InputItem[];
+}
+
+function stripInputItemIds(items: Array<Record<string, unknown>>): InputItem[] {
+	return items.map(item => {
+		if (item.id == null) return item as InputItem;
+		const { id: _id, ...rest } = item;
+		return rest as InputItem;
+	});
+}
+
+function recordCodexWebSocketRequestStats(
+	state: CodexWebSocketSessionState | undefined,
+	request: Record<string, unknown>,
+): void {
+	if (!state) return;
+	const input = request.input;
+	state.stats.lastInputItems = Array.isArray(input) ? input.length : 0;
+	if (typeof request.previous_response_id === "string" && request.previous_response_id.length > 0) {
+		state.stats.deltaRequests += 1;
+		state.stats.lastDeltaInputItems = state.stats.lastInputItems;
+		state.stats.lastPreviousResponseId = request.previous_response_id;
+		return;
+	}
+	state.stats.fullContextRequests += 1;
+	state.stats.lastDeltaInputItems = undefined;
+	state.stats.lastPreviousResponseId = undefined;
 }
 
 function buildCodexWebSocketRequest(
 	requestBody: RequestBody,
 	state: CodexWebSocketSessionState | undefined,
 ): Record<string, unknown> {
-	const appendInput = state?.canAppend ? buildAppendInput(state.lastRequest, requestBody) : null;
-	if (appendInput && appendInput.length > 0) {
-		if (state?.lastResponseId) {
-			return {
-				type: "response.create",
-				...requestBody,
-				previous_response_id: state.lastResponseId,
-				input: appendInput,
-			};
-		}
-		return {
-			type: "response.append",
+	const appendInput = state?.canAppend
+		? buildAppendInput(state.lastRequest, state.lastResponseItems, requestBody)
+		: null;
+	if (appendInput && appendInput.length > 0 && state?.lastResponseId) {
+		const request = {
+			type: "response.create",
+			...requestBody,
+			previous_response_id: state.lastResponseId,
 			input: appendInput,
 		};
+		recordCodexWebSocketRequestStats(state, request);
+		return request;
 	}
 	if (state?.canAppend) {
 		logCodexDebug("codex websocket append reset", {
@@ -1653,10 +1774,12 @@ function buildCodexWebSocketRequest(
 		resetCodexWebSocketAppendState(state);
 		resetCodexSessionMetadata(state);
 	}
-	return {
+	const request = {
 		type: "response.create",
 		...requestBody,
 	};
+	recordCodexWebSocketRequestStats(state, request);
+	return request;
 }
 
 function toWebSocketUrl(url: string): string {
@@ -2027,10 +2150,11 @@ function createCodexHeaders(
 			: OPENAI_HEADER_VALUES.BETA_RESPONSES;
 	headers.set(OPENAI_HEADERS.BETA, betaHeader);
 	headers.set(OPENAI_HEADERS.ORIGINATOR, OPENAI_HEADER_VALUES.ORIGINATOR_CODEX);
-	headers.set("User-Agent", `pi/${packageJson.version} (${os.platform()} ${os.release()}; ${os.arch()})`);
+	headers.set("User-Agent", getCodexUserAgent());
 	if (promptCacheKey) {
 		headers.set(OPENAI_HEADERS.CONVERSATION_ID, promptCacheKey);
 		headers.set(OPENAI_HEADERS.SESSION_ID, promptCacheKey);
+		headers.set("x-client-request-id", promptCacheKey);
 	} else {
 		headers.delete(OPENAI_HEADERS.CONVERSATION_ID);
 		headers.delete(OPENAI_HEADERS.SESSION_ID);

packages/ai/src/providers/openai-codex/request-transformer.ts

@@ -144,7 +144,7 @@ export async function transformRequestBody(
 
 	body.text = {
 		...body.text,
-		verbosity: options.textVerbosity || "medium",
+		verbosity: options.textVerbosity || "low",
 	};
 
 	const include = Array.isArray(options.include) ? [...options.include] : [];

packages/ai/src/types.ts

@@ -598,6 +598,10 @@ export interface AnthropicCompat {
 	 * tags: 'disabled', 'enabled'`.
 	 */
 	disableAdaptiveThinking?: boolean;
+	/** Whether tools may include Anthropic's per-tool eager_input_streaming flag. Default: true. */
+	supportsEagerToolInputStreaming?: boolean;
+	/** Whether long prompt-cache retention (`ttl: "1h"`) is supported. Default: true for canonical Anthropic API. */
+	supportsLongCacheRetention?: boolean;
 }
 
 /**

packages/ai/src/utils/json-parse.ts

@@ -1,5 +1,98 @@
 import { parse as partialParse } from "partial-json";
 
+const VALID_JSON_ESCAPES = new Set(['"', "\\", "/", "b", "f", "n", "r", "t", "u"]);
+
+function isControlCharacter(char: string): boolean {
+	const codePoint = char.codePointAt(0);
+	return codePoint !== undefined && codePoint >= 0x00 && codePoint <= 0x1f;
+}
+
+function escapeControlCharacter(char: string): string {
+	switch (char) {
+		case "\b":
+			return "\\b";
+		case "\f":
+			return "\\f";
+		case "\n":
+			return "\\n";
+		case "\r":
+			return "\\r";
+		case "\t":
+			return "\\t";
+		default:
+			return `\\u${char.codePointAt(0)?.toString(16).padStart(4, "0") ?? "0000"}`;
+	}
+}
+
+/**
+ * Repairs malformed JSON string literals by escaping raw control characters
+ * inside strings and preserving invalid escapes as literal backslashes.
+ */
+export function repairJson(json: string): string {
+	let repaired = "";
+	let inString = false;
+
+	for (let index = 0; index < json.length; index++) {
+		const char = json[index];
+
+		if (!inString) {
+			repaired += char;
+			if (char === '"') {
+				inString = true;
+			}
+			continue;
+		}
+
+		if (char === '"') {
+			repaired += char;
+			inString = false;
+			continue;
+		}
+
+		if (char === "\\") {
+			const nextChar = json[index + 1];
+			if (nextChar === undefined) {
+				repaired += "\\\\";
+				continue;
+			}
+
+			if (nextChar === "u") {
+				const unicodeDigits = json.slice(index + 2, index + 6);
+				if (/^[0-9a-fA-F]{4}$/.test(unicodeDigits)) {
+					repaired += `\\u${unicodeDigits}`;
+					index += 5;
+					continue;
+				}
+			}
+
+			if (VALID_JSON_ESCAPES.has(nextChar)) {
+				repaired += `\\${nextChar}`;
+				index += 1;
+				continue;
+			}
+
+			repaired += "\\\\";
+			continue;
+		}
+
+		repaired += isControlCharacter(char) ? escapeControlCharacter(char) : char;
+	}
+
+	return repaired;
+}
+
+export function parseJsonWithRepair<T>(json: string): T {
+	try {
+		return JSON.parse(json) as T;
+	} catch (error) {
+		const repairedJson = repairJson(json);
+		if (repairedJson !== json) {
+			return JSON.parse(repairedJson) as T;
+		}
+		throw error;
+	}
+}
+
 /**
  * Attempts to parse potentially incomplete JSON during streaming.
  * Always returns a valid object, even if the JSON is incomplete.
@@ -7,22 +100,27 @@ import { parse as partialParse } from "partial-json";
  * @param partialJson The partial JSON string from streaming
  * @returns Parsed object or empty object if parsing fails
  */
-export function parseStreamingJson<T = any>(partialJson: string | undefined): T {
+export function parseStreamingJson<T = Record<string, unknown>>(partialJson: string | undefined): T {
 	if (!partialJson || partialJson.trim() === "") {
 		return {} as T;
 	}
 
-	// Try standard parsing first (fastest for complete JSON)
+	// Try standard/repaired parsing first (fastest for complete JSON)
 	try {
-		return JSON.parse(partialJson) as T;
+		return parseJsonWithRepair<T>(partialJson);
 	} catch {
 		// Try partial-json for incomplete JSON
 		try {
 			const result = partialParse(partialJson);
 			return (result ?? {}) as T;
 		} catch {
-			// If all parsing fails, return empty object
-			return {} as T;
+			try {
+				const result = partialParse(repairJson(partialJson));
+				return (result ?? {}) as T;
+			} catch {
+				// If all parsing fails, return empty object
+				return {} as T;
+			}
 		}
 	}
 }

packages/ai/src/utils/oauth/anthropic.ts

@@ -13,6 +13,62 @@ const CALLBACK_PORT = 54545;
 const CALLBACK_PATH = "/callback";
 const SCOPES = "org:create_api_key user:profile user:inference";
 
+function formatErrorDetails(error: unknown): string {
+	if (error instanceof Error) {
+		const details: string[] = [`${error.name}: ${error.message}`];
+		const errorWithCode = error as Error & { code?: string; errno?: number | string; cause?: unknown };
+		if (errorWithCode.code) details.push(`code=${errorWithCode.code}`);
+		if (typeof errorWithCode.errno !== "undefined") details.push(`errno=${String(errorWithCode.errno)}`);
+		if (typeof error.cause !== "undefined") {
+			details.push(`cause=${formatErrorDetails(error.cause)}`);
+		}
+		if (error.stack) {
+			details.push(`stack=${error.stack}`);
+		}
+		return details.join("; ");
+	}
+	return String(error);
+}
+
+async function postJson(url: string, body: Record<string, string | number>): Promise<string> {
+	const response = await fetch(url, {
+		method: "POST",
+		headers: {
+			"Content-Type": "application/json",
+			Accept: "application/json",
+		},
+		body: JSON.stringify(body),
+		signal: AbortSignal.timeout(30_000),
+	});
+
+	const responseBody = await response.text();
+	if (!response.ok) {
+		throw new Error(`HTTP request failed. status=${response.status}; url=${url}; body=${responseBody}`);
+	}
+	return responseBody;
+}
+
+function parseOAuthTokenResponse(
+	responseBody: string,
+	operation: string,
+): {
+	access_token: string;
+	refresh_token: string;
+	expires_in: number;
+} {
+	try {
+		return JSON.parse(responseBody) as {
+			access_token: string;
+			refresh_token: string;
+			expires_in: number;
+		};
+	} catch (error) {
+		throw new Error(
+			`Anthropic ${operation} returned invalid JSON. url=${TOKEN_URL}; body=${responseBody}; details=${formatErrorDetails(error)}`,
+		);
+	}
+}
+
 export class AnthropicOAuthFlow extends OAuthCallbackFlow {
 	#verifier: string = "";
 	#challenge: string = "";
@@ -36,9 +92,13 @@ export class AnthropicOAuthFlow extends OAuthCallbackFlow {
 			code_challenge_method: "S256",
 			state,
 		});
-
 		const url = `${AUTHORIZE_URL}?${authParams.toString()}`;
-		return { url };
+
+		return {
+			url,
+			instructions:
+				"Complete login in your browser. If the browser cannot reach this machine, paste the final redirect URL or authorization code when prompted.",
+		};
 	}
 
 	async exchangeToken(code: string, state: string, redirectUri: string): Promise<OAuthCredentials> {
@@ -53,37 +113,23 @@ export class AnthropicOAuthFlow extends OAuthCallbackFlow {
 			}
 		}
 
-		const tokenResponse = await fetch(TOKEN_URL, {
-			method: "POST",
-			headers: {
-				"Content-Type": "application/json",
-				Accept: "application/json",
-			},
-			body: JSON.stringify({
+		let responseBody: string;
+		try {
+			responseBody = await postJson(TOKEN_URL, {
 				grant_type: "authorization_code",
 				client_id: CLIENT_ID,
 				code: exchangeCode,
 				state: exchangeState,
 				redirect_uri: redirectUri,
 				code_verifier: this.#verifier,
-			}),
-		});
-
-		if (!tokenResponse.ok) {
-			let error: string;
-			try {
-				error = await tokenResponse.text();
-			} catch {
-				error = `HTTP ${tokenResponse.status}`;
-			}
-			throw new Error(`Token exchange failed: ${error}`);
+			});
+		} catch (error) {
+			throw new Error(
+				`Token exchange request failed. url=${TOKEN_URL}; redirect_uri=${redirectUri}; response_type=authorization_code; details=${formatErrorDetails(error)}`,
+			);
 		}
 
-		const tokenData = (await tokenResponse.json()) as {
-			access_token: string;
-			refresh_token: string;
-			expires_in: number;
-		};
+		const tokenData = parseOAuthTokenResponse(responseBody, "token exchange");
 
 		return {
 			refresh: tokenData.refresh_token,
@@ -105,26 +151,18 @@ export async function loginAnthropic(ctrl: OAuthController): Promise<OAuthCreden
  * Refresh Anthropic OAuth token
  */
 export async function refreshAnthropicToken(refreshToken: string): Promise<OAuthCredentials> {
-	const response = await fetch(TOKEN_URL, {
-		method: "POST",
-		headers: { "Content-Type": "application/json", Accept: "application/json" },
-		body: JSON.stringify({
+	let responseBody: string;
+	try {
+		responseBody = await postJson(TOKEN_URL, {
 			grant_type: "refresh_token",
 			client_id: CLIENT_ID,
 			refresh_token: refreshToken,
-		}),
-	});
-
-	if (!response.ok) {
-		const error = await response.text();
-		throw new Error(`Anthropic token refresh failed: ${error}`);
+		});
+	} catch (error) {
+		throw new Error(`Anthropic token refresh request failed. url=${TOKEN_URL}; details=${formatErrorDetails(error)}`);
 	}
 
-	const data = (await response.json()) as {
-		access_token: string;
-		refresh_token: string;
-		expires_in: number;
-	};
+	const data = parseOAuthTokenResponse(responseBody, "token refresh");
 
 	return {
 		refresh: data.refresh_token || refreshToken,

packages/ai/test/anthropic-alignment.test.ts

@@ -37,6 +37,14 @@ const ANTHROPIC_MODEL: Model<"anthropic-messages"> = {
 	maxTokens: 8_192,
 };
 
+const CLOUDFLARE_ANTHROPIC_MODEL: Model<"anthropic-messages"> = {
+	...ANTHROPIC_MODEL,
+	id: "anthropic/claude-sonnet-4-5",
+	name: "Claude Sonnet 4.5 via Cloudflare",
+	provider: "cloudflare-ai-gateway",
+	baseUrl: "https://gateway.ai.cloudflare.com/v1/account/gateway/anthropic",
+};
+
 function createAbortedSignal(): AbortSignal {
 	const controller = new AbortController();
 	controller.abort();
@@ -549,6 +557,86 @@ describe("Anthropic request fingerprint alignment", () => {
 		expect(beta).not.toContain("fine-grained-tool-streaming-2025-05-14");
 	});
 
+	it("adds legacy fine-grained tool-streaming beta only for tool requests on incompatible models", () => {
+		const incompatibleModel: Model<"anthropic-messages"> = {
+			...ANTHROPIC_MODEL,
+			compat: { supportsEagerToolInputStreaming: false },
+		};
+
+		const withoutTools = buildAnthropicClientOptions({
+			model: incompatibleModel,
+			apiKey: "sk-ant-api-test",
+			extraBetas: [],
+			stream: true,
+			interleavedThinking: false,
+			hasTools: false,
+		});
+		const withCompatibleTools = buildAnthropicClientOptions({
+			model: ANTHROPIC_MODEL,
+			apiKey: "sk-ant-api-test",
+			extraBetas: [],
+			stream: true,
+			interleavedThinking: false,
+			hasTools: true,
+		});
+		const withIncompatibleTools = buildAnthropicClientOptions({
+			model: incompatibleModel,
+			apiKey: "sk-ant-api-test",
+			extraBetas: [],
+			stream: true,
+			interleavedThinking: false,
+			hasTools: true,
+		});
+
+		expect(withoutTools.defaultHeaders["Anthropic-Beta"]).not.toContain("fine-grained-tool-streaming-2025-05-14");
+		expect(withCompatibleTools.defaultHeaders["Anthropic-Beta"]).not.toContain(
+			"fine-grained-tool-streaming-2025-05-14",
+		);
+		expect(withIncompatibleTools.defaultHeaders["Anthropic-Beta"]).toContain(
+			"fine-grained-tool-streaming-2025-05-14",
+		);
+	});
+
+	it("uses Cloudflare AI Gateway authorization without Anthropic credential headers", () => {
+		const options = buildAnthropicClientOptions({
+			model: CLOUDFLARE_ANTHROPIC_MODEL,
+			apiKey: "cf-gateway-token",
+			extraBetas: [],
+			stream: true,
+			interleavedThinking: false,
+			dynamicHeaders: {},
+		});
+
+		expect(options.baseURL).toBe("https://gateway.ai.cloudflare.com/v1/account/gateway/anthropic");
+		expect(options.apiKey).toBeNull();
+		expect(options.authToken).toBeNull();
+		expect(options.defaultHeaders["cf-aig-authorization"]).toBe("Bearer cf-gateway-token");
+		expect(options.defaultHeaders.Authorization).toBeUndefined();
+		expect(options.defaultHeaders["X-Api-Key"]).toBeUndefined();
+	});
+
+	it("keeps Cloudflare gateway auth authoritative over caller-supplied auth headers", () => {
+		const options = buildAnthropicClientOptions({
+			model: {
+				...CLOUDFLARE_ANTHROPIC_MODEL,
+				headers: {
+					Authorization: "Bearer anthropic-oauth",
+					"X-Api-Key": "sk-ant-api-leak",
+					"cf-aig-authorization": "Bearer stale-token",
+				},
+			},
+			apiKey: "cf-gateway-token",
+			extraBetas: [],
+			stream: true,
+			interleavedThinking: false,
+			dynamicHeaders: {},
+		});
+
+		expect(options.defaultHeaders["cf-aig-authorization"]).toBe("Bearer cf-gateway-token");
+		expect(options.defaultHeaders.Authorization).toBeUndefined();
+		expect(options.defaultHeaders["X-Api-Key"]).toBeUndefined();
+	});
+
 	it("applies Claude Code TLS profile for direct Anthropic transport", () => {
 		const options = buildAnthropicClientOptions({
 			model: ANTHROPIC_MODEL,

packages/ai/test/anthropic-stream-envelope.test.ts

@@ -55,6 +55,32 @@ function createMockRequest(events: MockAnthropicEvent[]): MockAnthropicRequest {
 		},
 	};
 }
+function createRawSseRequest(frames: string[]): { asResponse(): Promise<Response> } {
+	const body = new TextEncoder().encode(frames.join(""));
+	return {
+		async asResponse() {
+			return new Response(body, {
+				status: 200,
+				headers: {
+					"content-type": "text/event-stream",
+					"request-id": "req_raw_mock",
+				},
+			});
+		},
+	};
+}
+
+function sseFrame(event: string, data: unknown): string {
+	return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
+}
+
+function sseRawFrame(event: string, data: string): string {
+	return `event: ${event}\ndata: ${data}\n\n`;
+}
+
+function createTextSuccessSseFrames(text: string, preamble: string[] = []): string[] {
+	return [...preamble, ...createTextSuccessEvents(text).map(event => sseFrame(String(event.type), event))];
+}
 
 function createRejectedMockRequest(error: Error): MockAnthropicRequest {
 	return {
@@ -386,4 +412,142 @@ describe("anthropic stream envelope handling", () => {
 		}
 		expect("partialJson" in toolCall).toBe(false);
 	});
+	it("parses raw SSE directly so unknown events do not fail Anthropic streams", async () => {
+		vi.spyOn(Messages.prototype, "create").mockImplementation(
+			() =>
+				createRawSseRequest(
+					createTextSuccessSseFrames("hello", [
+						sseFrame("anthropic_internal_trace", { type: "anthropic_internal_trace", trace_id: "trace_123" }),
+					]),
+				) as never,
+		);
+
+		const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" });
+		const events: AssistantMessageEvent[] = [];
+		for await (const event of stream) {
+			events.push(event);
+		}
+		const result = await stream.result();
+
+		expect(countEvents(events, "error")).toBe(0);
+		expect(countEvents(events, "done")).toBe(1);
+		expect(result.stopReason).toBe("stop");
+		expect(result.content).toEqual([{ type: "text", text: "hello" }]);
+	});
+
+	it("surfaces an error when a raw SSE stream closes before message_stop", async () => {
+		const incompleteFrames = createTextSuccessSseFrames("partial").filter(
+			frame => !frame.includes("event: message_stop"),
+		);
+		vi.spyOn(Messages.prototype, "create").mockImplementation(() => createRawSseRequest(incompleteFrames) as never);
+
+		const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" });
+		const events: AssistantMessageEvent[] = [];
+		for await (const event of stream) {
+			events.push(event);
+		}
+		const result = await stream.result();
+
+		expect(countEvents(events, "error")).toBe(1);
+		expect(countEvents(events, "done")).toBe(0);
+		expect(result.stopReason).toBe("error");
+		expect(result.errorMessage).toContain("stream ended before message_stop");
+		expect(result.content).toEqual([{ type: "text", text: "partial" }]);
+	});
+
+	it("repairs malformed JSON in raw SSE event data before parsing", async () => {
+		const malformedTextDelta =
+			'{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"line\\qbreak"}}';
+		const successEvents = createTextSuccessEvents("unused");
+		const frames = [
+			sseFrame("message_start", successEvents[0]),
+			sseFrame("content_block_start", successEvents[1]),
+			sseRawFrame("content_block_delta", malformedTextDelta),
+			sseFrame("content_block_stop", { type: "content_block_stop", index: 0 }),
+			sseFrame("message_delta", successEvents[5]),
+			sseFrame("message_stop", { type: "message_stop" }),
+		];
+		vi.spyOn(Messages.prototype, "create").mockImplementation(() => createRawSseRequest(frames) as never);
+
+		const stream = streamAnthropic(model, context, { apiKey: "sk-ant-test" });
+		for await (const _event of stream) {
+			// drain stream
+		}
+		const result = await stream.result();
+
+		expect(result.stopReason).toBe("stop");
+		expect(result.content).toEqual([{ type: "text", text: "line\\qbreak" }]);
+	});
+
+	it("emits per-tool eager_input_streaming only when Anthropic compat allows it", async () => {
+		const toolContext: Context = {
+			...context,
+			tools: [
+				{
+					name: "lookup_weather",
+					description: "Lookup weather",
+					parameters: Type.Object({ city: Type.String() }),
+				},
+			],
+		};
+		const payloads: unknown[] = [];
+		vi.spyOn(Messages.prototype, "create").mockImplementation((params: unknown) => {
+			payloads.push(params);
+			return createMockRequest(createTextSuccessEvents("ok")) as never;
+		});
+
+		const eagerStream = streamAnthropic(model, toolContext, { apiKey: "sk-ant-test" });
+		for await (const _event of eagerStream) {
+			// drain stream
+		}
+		await eagerStream.result();
+
+		const disabledStream = streamAnthropic(
+			{ ...model, compat: { supportsEagerToolInputStreaming: false } },
+			toolContext,
+			{ apiKey: "sk-ant-test" },
+		);
+		for await (const _event of disabledStream) {
+			// drain stream
+		}
+		await disabledStream.result();
+
+		const eagerTool = (payloads[0] as { tools?: Array<Record<string, unknown>> }).tools?.[0];
+		const disabledTool = (payloads[1] as { tools?: Array<Record<string, unknown>> }).tools?.[0];
+		expect(eagerTool?.eager_input_streaming).toBe(true);
+		expect(disabledTool).not.toHaveProperty("eager_input_streaming");
+	});
+
+	it("emits 1h cache TTL only for canonical Anthropic API with compatible long-cache support", async () => {
+		const payloads: unknown[] = [];
+		vi.spyOn(Messages.prototype, "create").mockImplementation((params: unknown) => {
+			payloads.push(params);
+			return createMockRequest(createTextSuccessEvents("ok")) as never;
+		});
+
+		for (const testModel of [
+			model,
+			{ ...model, compat: { supportsLongCacheRetention: false } },
+			{ ...model, baseUrl: "https://proxy.example.com/anthropic" },
+		]) {
+			const stream = streamAnthropic(testModel, context, {
+				apiKey: "sk-ant-test",
+				cacheRetention: "long",
+			});
+			for await (const _event of stream) {
+				// drain stream
+			}
+			await stream.result();
+		}
+
+		const cacheControls = payloads.map(payload => {
+			const messages = (payload as { messages: Array<{ content: unknown }> }).messages;
+			const content = messages.at(-1)?.content;
+			if (!Array.isArray(content)) return undefined;
+			return (content.at(-1) as { cache_control?: { ttl?: string; type: string } } | undefined)?.cache_control;
+		});
+		expect(cacheControls[0]).toEqual({ type: "ephemeral", ttl: "1h" });
+		expect(cacheControls[1]).toEqual({ type: "ephemeral" });
+		expect(cacheControls[2]).toEqual({ type: "ephemeral" });
+	});
 });

packages/ai/test/openai-codex-stream.test.ts

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, vi } from "bun:test";
 import { enrichModelThinking } from "@oh-my-pi/pi-ai/model-thinking";
 import {
 	getOpenAICodexTransportDetails,
+	getOpenAICodexWebSocketDebugStats,
 	prewarmOpenAICodexResponses,
 	streamOpenAICodexResponses,
 } from "@oh-my-pi/pi-ai/providers/openai-codex-responses";
@@ -165,7 +166,7 @@ describe("openai-codex streaming", () => {
 			`data: ${JSON.stringify({ type: "response.content_part.added", part: { type: "output_text", text: "" } })}`,
 			`data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello" })}`,
 			`data: ${JSON.stringify({ type: "response.output_item.done", item: { type: "message", id: "msg_1", role: "assistant", status: "completed", content: [{ type: "output_text", text: "Hello" }] } })}`,
-			`data: ${JSON.stringify({ type: "response.completed", response: { status: "completed", usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } } } })}`,
+			`data: ${JSON.stringify({ type: "response.completed", response: { status: "completed", service_tier: "default", usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } } } })}`,
 		].join("\n\n")}\n\n`;
 		const fetchMock = vi.fn(async (_input: string | URL, init?: RequestInit) => {
 			capturedBody = JSON.parse(String(init?.body)) as Record<string, unknown>;
@@ -184,7 +185,7 @@ describe("openai-codex streaming", () => {
 			baseUrl: "https://chatgpt.com/backend-api",
 			reasoning: true,
 			input: ["text"],
-			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+			cost: { input: 1, output: 2, cacheRead: 0.5, cacheWrite: 0 },
 			contextWindow: 400000,
 			maxTokens: 128000,
 		};
@@ -200,6 +201,9 @@ describe("openai-codex streaming", () => {
 		}).result();
 		expect(result.stopReason).toBe("stop");
 		expect(capturedBody?.service_tier).toBe("priority");
+		expect(result.usage.cost.input).toBeCloseTo(0.00001);
+		expect(result.usage.cost.output).toBeCloseTo(0.000012);
+		expect(result.usage.cost.total).toBeCloseTo(0.000022);
 	});
 
 	it("fails truncated SSE streams that never emit a terminal response event", async () => {
@@ -266,6 +270,48 @@ describe("openai-codex streaming", () => {
 		expect(result.errorMessage).toContain("terminal completion event");
 	});
 
+	it("stops reading SSE responses after a terminal response event", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-stream-");
+		setAgentDir(tempDir.path());
+
+		const payload = Buffer.from(
+			JSON.stringify({ "https://api.openai.com/auth": { chatgpt_account_id: "acc_test" } }),
+			"utf8",
+		).toBase64();
+		const token = `aaa.${payload}.bbb`;
+		const sse = `${[
+			`data: ${JSON.stringify({ type: "response.output_item.added", item: { type: "message", id: "msg_1", role: "assistant", status: "in_progress", content: [] } })}`,
+			`data: ${JSON.stringify({ type: "response.content_part.added", part: { type: "output_text", text: "" } })}`,
+			`data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello" })}`,
+			`data: ${JSON.stringify({ type: "response.output_item.done", item: { type: "message", id: "msg_1", role: "assistant", status: "completed", content: [{ type: "output_text", text: "Hello" }] } })}`,
+			`data: ${JSON.stringify({ type: "response.done", response: { status: "completed", usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } } } })}`,
+			`data: ${JSON.stringify({ type: "response.failed", code: "server_error", message: "late failure after terminal event" })}`,
+		].join("\n\n")}\n\n`;
+
+		global.fetch = vi.fn(async () => new Response(sse, { status: 200, headers: { "content-type": "text/event-stream" } })) as unknown as typeof fetch;
+
+		const model: Model<"openai-codex-responses"> = {
+			id: "gpt-5.1-codex",
+			name: "GPT-5.1 Codex",
+			api: "openai-codex-responses",
+			provider: "openai-codex",
+			baseUrl: "https://chatgpt.com/backend-api",
+			reasoning: true,
+			input: ["text"],
+			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+			contextWindow: 400000,
+			maxTokens: 128000,
+		};
+		const context: Context = {
+			systemPrompt: "You are a helpful assistant.",
+			messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }],
+		};
+
+		const result = await streamOpenAICodexResponses(model, context, { apiKey: token }).result();
+		expect(result.stopReason).toBe("stop");
+		expect(result.content.find(block => block.type === "text")?.text).toBe("Hello");
+	});
+
 	it("surfaces 429 errors after retry budget checks without body reuse failures", async () => {
 		const tempDir = TempDir.createSync("@pi-codex-stream-");
 		setAgentDir(tempDir.path());
@@ -450,6 +496,7 @@ describe("openai-codex streaming", () => {
 				// Verify sessionId is set in headers
 				expect(headers?.get("conversation_id")).toBe(sessionId);
 				expect(headers?.get("session_id")).toBe(sessionId);
+				expect(headers?.get("x-client-request-id")).toBe(sessionId);
 
 				// Verify sessionId is set in request body as prompt_cache_key
 				const body = typeof init?.body === "string" ? (JSON.parse(init.body) as Record<string, unknown>) : null;
@@ -1174,6 +1221,224 @@ describe("openai-codex streaming", () => {
 		expect(sentRequests[0]?.service_tier).toBe("priority");
 	});
 
+	it("sends websocket continuation deltas after prior assistant response items and records stats", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-stream-");
+		setAgentDir(tempDir.path());
+		const payload = Buffer.from(
+			JSON.stringify({ "https://api.openai.com/auth": { chatgpt_account_id: "acc_test" } }),
+			"utf8",
+		).toBase64();
+		const token = `aaa.${payload}.bbb`;
+		const sentRequests: Array<Record<string, unknown>> = [];
+		const fetchMock = vi.fn(async () => {
+			throw new Error("SSE fallback should not be called");
+		});
+		global.fetch = fetchMock as unknown as typeof fetch;
+
+		type WsListener = (event: Event) => void;
+		class DeltaWebSocket {
+			static readonly CONNECTING = 0;
+			static readonly OPEN = 1;
+			static readonly CLOSING = 2;
+			static readonly CLOSED = 3;
+			readyState = DeltaWebSocket.CONNECTING;
+			#listeners = new Map<string, Set<WsListener>>();
+
+			constructor(_url: string, _options?: { headers?: Record<string, string> }) {
+				setTimeout(() => {
+					this.readyState = DeltaWebSocket.OPEN;
+					this.#emit("open", new Event("open"));
+				}, 0);
+			}
+
+			addEventListener(type: string, listener: unknown): void {
+				if (typeof listener !== "function") return;
+				const listeners = this.#listeners.get(type) ?? new Set<WsListener>();
+				listeners.add(listener as WsListener);
+				this.#listeners.set(type, listeners);
+			}
+
+			removeEventListener(type: string, listener: unknown): void {
+				if (typeof listener !== "function") return;
+				const listeners = this.#listeners.get(type);
+				listeners?.delete(listener as WsListener);
+			}
+
+			send(data: string): void {
+				const request = JSON.parse(data) as Record<string, unknown>;
+				sentRequests.push(request);
+				const responseIndex = sentRequests.length;
+				const responseId = `resp_${responseIndex}`;
+				const messageId = `msg_${responseIndex}`;
+				const text = responseIndex === 1 ? "First answer" : "Second answer";
+				this.#emit("message", {
+					data: JSON.stringify({ type: "response.created", response: { id: responseId } }),
+				} as unknown as Event);
+				this.#emit("message", {
+					data: JSON.stringify({
+						type: "response.output_item.added",
+						item: { type: "message", id: messageId, role: "assistant", status: "in_progress", content: [] },
+					}),
+				} as unknown as Event);
+				this.#emit("message", {
+					data: JSON.stringify({ type: "response.content_part.added", part: { type: "output_text", text: "" } }),
+				} as unknown as Event);
+				this.#emit("message", {
+					data: JSON.stringify({ type: "response.output_text.delta", delta: text }),
+				} as unknown as Event);
+				this.#emit("message", {
+					data: JSON.stringify({
+						type: "response.output_item.done",
+						item: {
+							type: "message",
+							id: messageId,
+							role: "assistant",
+							status: "completed",
+							content: [{ type: "output_text", text }],
+						},
+					}),
+				} as unknown as Event);
+				this.#emit("message", {
+					data: JSON.stringify({
+						type: "response.done",
+						response: {
+							id: responseId,
+							status: "completed",
+							usage: {
+								input_tokens: 5,
+								output_tokens: 3,
+								total_tokens: 8,
+								input_tokens_details: { cached_tokens: 0 },
+							},
+						},
+					}),
+				} as unknown as Event);
+			}
+
+			close(): void {
+				this.readyState = DeltaWebSocket.CLOSED;
+			}
+
+			#emit(type: string, event: Event): void {
+				const listeners = this.#listeners.get(type);
+				if (!listeners) return;
+				for (const listener of listeners) {
+					listener(event);
+				}
+			}
+		}
+
+		global.WebSocket = DeltaWebSocket as unknown as typeof WebSocket;
+		const model: Model<"openai-codex-responses"> = {
+			id: "gpt-5.3-codex-spark",
+			name: "GPT-5.3 Codex Spark",
+			api: "openai-codex-responses",
+			provider: "openai-codex",
+			baseUrl: "https://chatgpt.com/backend-api",
+			reasoning: true,
+			preferWebsockets: true,
+			input: ["text"],
+			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+			contextWindow: 128000,
+			maxTokens: 128000,
+		};
+		const providerSessionState = new Map<string, ProviderSessionState>();
+		const firstContext: Context = {
+			systemPrompt: "You are a helpful assistant.",
+			messages: [{ role: "user", content: "First question", timestamp: Date.now() }],
+		};
+		const firstResponse = await streamOpenAICodexResponses(model, firstContext, {
+			apiKey: token,
+			sessionId: "ws-delta-session",
+			providerSessionState,
+		}).result();
+		const secondContext: Context = {
+			systemPrompt: "You are a helpful assistant.",
+			messages: [
+				...firstContext.messages,
+				firstResponse,
+				{ role: "user", content: "Second question", timestamp: Date.now() },
+			],
+		};
+		await streamOpenAICodexResponses(model, secondContext, {
+			apiKey: token,
+			sessionId: "ws-delta-session",
+			providerSessionState,
+		}).result();
+
+		expect(fetchMock).not.toHaveBeenCalled();
+		expect(sentRequests).toHaveLength(2);
+		expect(sentRequests[0]?.previous_response_id).toBeUndefined();
+		expect(sentRequests[0]?.prompt_cache_key).toBe("ws-delta-session");
+		expect(sentRequests[1]?.type).toBe("response.create");
+		expect(sentRequests[1]?.previous_response_id).toBe("resp_1");
+		expect(sentRequests[1]?.prompt_cache_key).toBe("ws-delta-session");
+		const deltaInput = sentRequests[1]?.input;
+		expect(Array.isArray(deltaInput)).toBe(true);
+		const deltaItems = deltaInput as Array<{ role?: string }>;
+		expect(deltaItems).toHaveLength(1);
+		expect(deltaItems[0]?.role).toBe("user");
+		expect(JSON.stringify(deltaItems)).toContain("Second question");
+		expect(JSON.stringify(deltaItems)).not.toContain("First answer");
+
+		const stats = getOpenAICodexWebSocketDebugStats(model, {
+			sessionId: "ws-delta-session",
+			providerSessionState,
+		});
+		expect(stats).toEqual({
+			fullContextRequests: 1,
+			deltaRequests: 1,
+			lastInputItems: 1,
+			lastDeltaInputItems: 1,
+			lastPreviousResponseId: "resp_1",
+		});
+	});
+
+	it("uses low Codex text verbosity by default while preserving explicit overrides", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-verbosity-");
+		setAgentDir(tempDir.path());
+		const payload = Buffer.from(
+			JSON.stringify({ "https://api.openai.com/auth": { chatgpt_account_id: "acc_test" } }),
+			"utf8",
+		).toBase64();
+		const token = `aaa.${payload}.bbb`;
+		const capturedBodies: Array<Record<string, unknown>> = [];
+		const sse = `${[
+			`data: ${JSON.stringify({ type: "response.output_item.added", item: { type: "message", id: "msg_verbosity", role: "assistant", status: "in_progress", content: [] } })}`,
+			`data: ${JSON.stringify({ type: "response.content_part.added", part: { type: "output_text", text: "" } })}`,
+			`data: ${JSON.stringify({ type: "response.output_text.delta", delta: "Hello" })}`,
+			`data: ${JSON.stringify({ type: "response.output_item.done", item: { type: "message", id: "msg_verbosity", role: "assistant", status: "completed", content: [{ type: "output_text", text: "Hello" }] } })}`,
+			`data: ${JSON.stringify({ type: "response.completed", response: { status: "completed", usage: { input_tokens: 5, output_tokens: 3, total_tokens: 8, input_tokens_details: { cached_tokens: 0 } } } })}`,
+		].join("\n\n")}\n\n`;
+		const fetchMock = vi.fn(async (_input: string | URL, init?: RequestInit) => {
+			capturedBodies.push(JSON.parse(String(init?.body)) as Record<string, unknown>);
+			return new Response(sse, { status: 200, headers: { "content-type": "text/event-stream" } });
+		});
+		global.fetch = fetchMock as unknown as typeof fetch;
+		const model: Model<"openai-codex-responses"> = {
+			id: "gpt-5.1-codex",
+			name: "GPT-5.1 Codex",
+			api: "openai-codex-responses",
+			provider: "openai-codex",
+			baseUrl: "https://chatgpt.com/backend-api",
+			reasoning: true,
+			input: ["text"],
+			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+			contextWindow: 400000,
+			maxTokens: 128000,
+		};
+		const context: Context = {
+			systemPrompt: "You are a helpful assistant.",
+			messages: [{ role: "user", content: "Say hello", timestamp: Date.now() }],
+		};
+
+		await streamOpenAICodexResponses(model, context, { apiKey: token }).result();
+		await streamOpenAICodexResponses(model, context, { apiKey: token, textVerbosity: "high" }).result();
+
+		expect((capturedBodies[0]?.text as { verbosity?: string } | undefined)?.verbosity).toBe("low");
+		expect((capturedBodies[1]?.text as { verbosity?: string } | undefined)?.verbosity).toBe("high");
+	});
+
 	it("uses websocket v2 beta header when v2 mode is enabled", async () => {
 		const tempDir = TempDir.createSync("@pi-codex-stream-");
 		setAgentDir(tempDir.path());