From e22e75e70edd1394db7d74006bc17c7547eef555 Mon Sep 17 00:00:00 2001 From: can1357 Date: Sat, 2 May 2026 00:38:54 +0200 Subject: [PATCH] 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 | 19 +- packages/ai/src/index.ts | 2 +- packages/ai/src/providers/anthropic.ts | 332 +++++++++++++++++- .../src/providers/openai-codex-responses.ts | 176 ++++++++-- .../openai-codex/request-transformer.ts | 2 +- packages/ai/src/types.ts | 4 + packages/ai/src/utils/json-parse.ts | 108 +++++- packages/ai/src/utils/oauth/anthropic.ts | 118 ++++--- packages/ai/test/anthropic-alignment.test.ts | 88 +++++ .../ai/test/anthropic-stream-envelope.test.ts | 164 +++++++++ packages/ai/test/openai-codex-stream.test.ts | 269 +++++++++++++- 11 files changed, 1193 insertions(+), 89 deletions(-) diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index e030cf437..1262ec7f3 100644 --- a/packages/ai/CHANGELOG.md +++ b/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. \ No newline at end of file diff --git a/packages/ai/src/index.ts b/packages/ai/src/index.ts index 09f9902ee..218b79ade 100644 --- a/packages/ai/src/index.ts +++ b/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"; diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 62c5ad6d2..2fb45b917 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/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; + 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 | undefined, headerName: string): string | undefined { if (!headers) return undefined; const normalizedName = headerName.toLowerCase(); @@ -131,6 +136,16 @@ export function buildAnthropicHeaders(options: AnthropicHeaderOptions): Record !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; dynamicHeaders?: Record; 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 | 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 = 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, + signal?: AbortSignal, +): AsyncGenerator { + 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 { + 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(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; +}; + +function hasAnthropicRawResponseRequest(request: unknown): request is AnthropicRawResponseRequest { + return isRecord(request) && typeof request.asResponse === "function"; +} + +type AnthropicStreamWithResponseRequest = { + withResponse(): Promise<{ + data: AsyncIterable; + 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; 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["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 } : {}), }; }); diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 6097141cd..6f571f845 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/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, "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, "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; + } + | 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; + }, +): 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; }, ): 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>): 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, +): 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 { - 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); diff --git a/packages/ai/src/providers/openai-codex/request-transformer.ts b/packages/ai/src/providers/openai-codex/request-transformer.ts index 268b8fb4a..9afd4a39e 100644 --- a/packages/ai/src/providers/openai-codex/request-transformer.ts +++ b/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] : []; diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index 951aecb6d..95eb5e7ad 100644 --- a/packages/ai/src/types.ts +++ b/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; } /** diff --git a/packages/ai/src/utils/json-parse.ts b/packages/ai/src/utils/json-parse.ts index feeb32ad1..c933c4b07 100644 --- a/packages/ai/src/utils/json-parse.ts +++ b/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(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(partialJson: string | undefined): T { +export function parseStreamingJson>(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(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; + } } } } diff --git a/packages/ai/src/utils/oauth/anthropic.ts b/packages/ai/src/utils/oauth/anthropic.ts index 14ff8e9dd..60793d078 100644 --- a/packages/ai/src/utils/oauth/anthropic.ts +++ b/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): Promise { + 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 { @@ -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 { - 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, diff --git a/packages/ai/test/anthropic-alignment.test.ts b/packages/ai/test/anthropic-alignment.test.ts index b089df8c4..5bbb30610 100644 --- a/packages/ai/test/anthropic-alignment.test.ts +++ b/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, diff --git a/packages/ai/test/anthropic-stream-envelope.test.ts b/packages/ai/test/anthropic-stream-envelope.test.ts index 5af3fcd5c..f80dc09f9 100644 --- a/packages/ai/test/anthropic-stream-envelope.test.ts +++ b/packages/ai/test/anthropic-stream-envelope.test.ts @@ -55,6 +55,32 @@ function createMockRequest(events: MockAnthropicEvent[]): MockAnthropicRequest { }, }; } +function createRawSseRequest(frames: string[]): { asResponse(): Promise } { + 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> }).tools?.[0]; + const disabledTool = (payloads[1] as { tools?: Array> }).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" }); + }); }); diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index c7bc5089b..0df025e65 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/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; @@ -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) : 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> = []; + 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>(); + + constructor(_url: string, _options?: { headers?: Record }) { + 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(); + 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; + 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(); + 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> = []; + 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); + 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());