diff --git a/src/auth/codex/usage-limit-error.ts b/src/auth/codex/usage-limit-error.ts index 27e8adfec..1456827b9 100644 --- a/src/auth/codex/usage-limit-error.ts +++ b/src/auth/codex/usage-limit-error.ts @@ -13,6 +13,12 @@ * other providers as Codex. */ +import { + coerceBody, + readErrorNode, + resetSecondsFromNode, +} from "../../provider/error-body.js"; + export interface CodexUsageLimitError { readonly code: string; readonly message: string; @@ -26,48 +32,6 @@ const USAGE_LIMIT_CODES = new Set([ "usage_limit_exceeded", ]); -function asRecord(value: unknown): Record | undefined { - if (typeof value !== "object" || value === null || Array.isArray(value)) - return undefined; - return value as Record; -} - -function tryParseJSON(text: string): unknown { - try { - return JSON.parse(text) as unknown; - } catch { - return undefined; - } -} - -function coerceBody(raw: unknown): unknown { - if (typeof raw === "string") { - const trimmed = raw.trim(); - if (trimmed.length === 0) return undefined; - return tryParseJSON(trimmed) ?? raw; - } - return raw; -} - -function readErrorNode(body: unknown): Record | undefined { - const root = asRecord(body); - if (root === undefined) return undefined; - - const detail = asRecord(root["detail"]); - if (detail !== undefined) { - const nested = asRecord(detail["error"]); - if (nested !== undefined) return nested; - // Some gateways put the fields directly under detail. - if (typeof detail["code"] === "string") return detail; - } - - const top = asRecord(root["error"]); - if (top !== undefined) return top; - - if (typeof root["code"] === "string") return root; - return undefined; -} - /** * Nested Codex `detail.error.message` when the harness left it on `raw` while * classifying the HTTP statusText. Unlike `parseCodexUsageLimitError`, this @@ -106,16 +70,7 @@ export function parseCodexUsageLimitError( ? node["planType"] : undefined; - const resetsRaw = - node["resets_in_seconds"] ?? - node["resetsInSeconds"] ?? - node["reset_after_seconds"]; - const resetsInSeconds = - typeof resetsRaw === "number" && - Number.isFinite(resetsRaw) && - resetsRaw >= 0 - ? Math.floor(resetsRaw) - : undefined; + const resetsInSeconds = resetSecondsFromNode(node); return { code, diff --git a/src/inference-gateway-error.test.ts b/src/inference-gateway-error.test.ts index 35be2cfb0..ec4c1fd3e 100644 --- a/src/inference-gateway-error.test.ts +++ b/src/inference-gateway-error.test.ts @@ -308,6 +308,39 @@ describe("normalizeInferenceErrorForRetry", () => { ); }); + test("Codex bare 429 with a body reset window retries on the body, not the header (CL-10664)", () => { + const normalized = normalizeInferenceErrorForRetry({ + ...BARE_429, + providerId: "codex/acme-labs", + raw: { + error: { + code: "rate_limit_exceeded", + message: "Too Many Requests", + resets_in_seconds: 7, + }, + }, + }); + expect(normalized.category).toBe("retryable"); + expect(normalized.message).toBe("Rate limited"); + expect(normalized.retryAfterMs).toBe(7_000); + }); + + test("non-Codex 429s keep the header retry-after; a body reset is not read (CL-10664)", () => { + const normalized = normalizeInferenceErrorForRetry({ + ...BARE_429, + providerId: "xai/alice", + raw: { + error: { + code: "rate_limit_exceeded", + message: "Too Many Requests", + resets_in_seconds: 7, + }, + }, + }); + expect(normalized.category).toBe("retryable"); + expect(normalized.retryAfterMs).toBe(45_000); + }); + test.each([{}, { providerId: "openai" }])( "bare 429 without a known provider keeps intx's quota_exhausted (%j)", (context) => { diff --git a/src/inference-gateway-error.ts b/src/inference-gateway-error.ts index 4bba9d0a7..b84a04d66 100644 --- a/src/inference-gateway-error.ts +++ b/src/inference-gateway-error.ts @@ -19,6 +19,12 @@ import { xaiProfileFromProviderName, } from "./config/xai-providers.js"; import { isXaiGrokLeafProvider } from "./subagent/provider-family.js"; +import { + coerceBody, + readErrorNode, + resetSecondsFromNode, + tryParseJSON, +} from "./provider/error-body.js"; export interface InferenceErrorLike { category: string; @@ -155,14 +161,6 @@ export function isGatewayOverloadInferenceError( return false; } -function tryParseJSON(text: string): unknown { - try { - return JSON.parse(text) as unknown; - } catch { - return undefined; - } -} - /** * Known OpenCode Go errors: provider context (id/flag/request URL) or Go * body markers; bare `provider_rate_limit_exceeded` alone is not enough — @@ -352,25 +350,28 @@ export function isXaiShortRateLimitInferenceError( /** * Shared skeleton behind the twin rate-limit normalizers: known-provider * quota 429s without usage-limit bodies become retryable with scrubbed copy. + * An optional body reader lifts a carrier's reset window onto `retryAfterMs` + * when the wire omits a Retry-After header (body wins, matching the + * usage-limit path). */ function normalizeProviderRateLimitError( error: InferenceErrorWithGoContext, isKnownProvider: (providerId: string | undefined) => boolean, isUsageLimit: (error: InferenceErrorLike) => boolean, + retryFromBody?: (error: InferenceErrorLike) => number | undefined, ): InferenceError { if (error.statusCode !== 429) return error; if (error.category !== "quota_exhausted") return error; if (!isKnownProvider(error.providerId)) return error; if (isUsageLimit(error)) return error; + const retryAfterMs = retryFromBody?.(error) ?? error.retryAfterMs; return { category: "retryable", message: RATE_LIMIT_USER_MESSAGE, statusCode: 429, ...(error.raw !== undefined ? { raw: error.raw } : {}), - ...(error.retryAfterMs !== undefined - ? { retryAfterMs: error.retryAfterMs } - : {}), + ...(retryAfterMs !== undefined ? { retryAfterMs } : {}), }; } @@ -430,6 +431,20 @@ export function isCodexShortRateLimitInferenceError( ); } +/** + * Codex sends its reset window in the 429 body, not a header; when a bare + * rate-limit body carries one, retry on that window instead of the policy + * default. Exact usage-limit codes never reach here (the skeleton vetoes). + */ +function codexRetryFromBody(error: InferenceErrorLike): number | undefined { + const node = readErrorNode(coerceBody(error.raw)); + if (node === undefined) return undefined; + const seconds = resetSecondsFromNode(node); + if (seconds === undefined) return undefined; + if (seconds <= 0) return 0; + return seconds * 1000; +} + /** * Bare 429 in a known-Codex context reclassifies as retryable; intx * defaults bare 429 to quota_exhausted; nested `usage_limit_reached` keeps @@ -442,6 +457,7 @@ export function normalizeCodexRateLimitError( error, isKnownCodexProviderId, hasCodexUsageLimit, + codexRetryFromBody, ); } diff --git a/src/provider/adapter-decorators.ts b/src/provider/adapter-decorators.ts new file mode 100644 index 000000000..7b7c2f222 --- /dev/null +++ b/src/provider/adapter-decorators.ts @@ -0,0 +1,90 @@ +/** + * Shared adapter decorator mechanics. Only the JSON-body / extra-header / + * registry plumbing is shared; the per-adapter semantics stay in each + * adapter's rewrite or header factory (codex parallel_tool_calls, DeepSeek + * reasoning strip and V4 quirks, Bifrost virtual-key sentinel, anthropic + * session header). Every decorator is a null overhead: when its hook bails, + * the base request's body and headers pass through byte-identical. + */ + +import type { + AdapterRegistry, + BuiltRequest, + ProviderAdapter, +} from "@intx/inference"; +import type { + ConversationTurn, + InferenceOptions, + LastCycleSource, +} from "@intx/types/runtime"; + +export type DecoratorContext = { + readonly messages: readonly ConversationTurn[]; + readonly model: string; + readonly options: InferenceOptions; +}; + +/** Parse the built body, apply the rewrite, and re-stringify only when the + * rewrite reports a change. An optional gate keeps adapters that only + * sometimes rewrite on their zero-parse fast path. */ +export function withJsonBodyRewrite( + adapter: ProviderAdapter, + rewrite: ( + body: Record, + context: DecoratorContext, + ) => boolean, + gate?: (request: BuiltRequest, context: DecoratorContext) => boolean, +): ProviderAdapter { + return { + ...adapter, + buildRequest: (messages, model, options) => { + const built = adapter.buildRequest(messages, model, options); + const context: DecoratorContext = { messages, model, options }; + if (gate !== undefined && !gate(built, context)) return built; + let body: Record; + try { + body = JSON.parse(built.body) as Record; + } catch { + return built; + } + if (!rewrite(body, context)) return built; + return { ...built, body: JSON.stringify(body) }; + }, + }; +} + +/** Merge the factory's headers over the built request's own; the spread + * never removes or rewrites a header the base adapter emitted. */ +export function withExtraHeaders( + adapter: ProviderAdapter, + headers: (context: DecoratorContext) => Record, +): ProviderAdapter { + return { + ...adapter, + buildRequest: (messages, model, options) => { + const built = adapter.buildRequest(messages, model, options); + const context: DecoratorContext = { messages, model, options }; + return { + ...built, + headers: { ...built.headers, ...headers(context) }, + }; + }, + }; +} + +/** Wrap an adapter registry so every resolved adapter passes through the + * decorator; membership is delegated untouched. */ +export function withDecoratedAdapters( + adapters: AdapterRegistry, + decorate: ( + adapter: ProviderAdapter, + source: LastCycleSource, + ) => ProviderAdapter, +): AdapterRegistry { + return { + has: (provider) => adapters.has(provider), + resolve(source, quirks) { + return decorate(adapters.resolve(source, quirks), source); + }, + }; +} diff --git a/src/provider/anthropic-cache-breakpoint.ts b/src/provider/anthropic-cache-breakpoint.ts index 34417b84d..f5f90254b 100644 --- a/src/provider/anthropic-cache-breakpoint.ts +++ b/src/provider/anthropic-cache-breakpoint.ts @@ -1,19 +1,13 @@ import type { AdapterRegistry, - BuiltRequest, ExtendedInferenceOptions, ProviderAdapter, } from "@intx/inference"; import { - OPENCODE_GO_MESSAGES_PROVIDER, - ZEN_MESSAGES_PROVIDER, -} from "./anthropic-session-adapter.js"; - -const ANTHROPIC_MESSAGES_PROVIDERS: ReadonlySet = new Set([ - "anthropic", - ZEN_MESSAGES_PROVIDER, - OPENCODE_GO_MESSAGES_PROVIDER, -]); + withDecoratedAdapters, + withJsonBodyRewrite, +} from "./adapter-decorators.js"; +import { ANTHROPIC_MESSAGES_PROTOCOLS } from "./provider-protocols.js"; type WireBlock = Record; @@ -57,46 +51,35 @@ function placeBreakpoint(messages: WireMessage[], before: number): boolean { } function moveBreakpointToPersistedTail( - built: BuiltRequest, + body: Record, options: ExtendedInferenceOptions, -): BuiltRequest { +): boolean { const suffix = ephemeralSuffixLength(options); - if (suffix === 0) return built; - let body: Record; - try { - body = JSON.parse(built.body) as Record; - } catch { - return built; - } - if (!Array.isArray(body.messages)) return built; + if (suffix === 0) return false; + if (!Array.isArray(body.messages)) return false; const messages = body.messages as WireMessage[]; const split = messages.length - suffix; - if (split <= 0) return built; + if (split <= 0) return false; stripBreakpoints(messages, split); - if (!placeBreakpoint(messages, split)) return built; - return { ...built, body: JSON.stringify(body) }; + return placeBreakpoint(messages, split); } +// Only Anthropic-protocol sources are decorated, so non-Anthropic requests +// skip both the gate and the body parse (byte-identical to the base adapter). function withMovedBreakpoint(adapter: ProviderAdapter): ProviderAdapter { - return { - ...adapter, - buildRequest: (turns, model, options) => - moveBreakpointToPersistedTail( - adapter.buildRequest(turns, model, options), - options, - ), - }; + return withJsonBodyRewrite( + adapter, + (body, { options }) => moveBreakpointToPersistedTail(body, options), + (_request, { options }) => ephemeralSuffixLength(options) > 0, + ); } export function withAnthropicCacheBreakpoint( adapters: AdapterRegistry, ): AdapterRegistry { - return { - has: (provider) => adapters.has(provider), - resolve(source, quirks) { - const adapter = adapters.resolve(source, quirks); - if (!ANTHROPIC_MESSAGES_PROVIDERS.has(source.provider)) return adapter; - return withMovedBreakpoint(adapter); - }, - }; + return withDecoratedAdapters(adapters, (adapter, source) => + ANTHROPIC_MESSAGES_PROTOCOLS.has(source.provider) + ? withMovedBreakpoint(adapter) + : adapter, + ); } diff --git a/src/provider/anthropic-session-adapter.ts b/src/provider/anthropic-session-adapter.ts index 04451684b..f7bd5d65e 100644 --- a/src/provider/anthropic-session-adapter.ts +++ b/src/provider/anthropic-session-adapter.ts @@ -1,9 +1,10 @@ -import { type BuiltRequest, type ProviderAdapter } from "@intx/inference"; +import { type ProviderAdapter } from "@intx/inference"; import { createAnthropicAdapter } from "@intx/inference/providers"; import { OPENCODE_SESSION_ID_OPTION, optionString, } from "./opencode-session.js"; +import { withExtraHeaders } from "./adapter-decorators.js"; export const ZEN_MESSAGES_PROVIDER = "zen-messages"; export const OPENCODE_GO_MESSAGES_PROVIDER = "opencode-go-messages"; @@ -14,22 +15,14 @@ export function createSessionHeaderAnthropicAdapter( source: AdapterSource, quirks?: unknown, ): ProviderAdapter { - const base = createAnthropicAdapter(source, quirks); - const buildRequest: ProviderAdapter["buildRequest"] = ( - messages, - model, - options, - ) => { - const built = base.buildRequest(messages, model, options); - const sessionId = optionString(options, OPENCODE_SESSION_ID_OPTION); - if (sessionId === undefined) return built; - const headers: BuiltRequest["headers"] = { - ...built.headers, - "x-opencode-session": sessionId, - }; - return { ...built, headers }; - }; - return { ...base, buildRequest }; + return withExtraHeaders( + createAnthropicAdapter(source, quirks), + ({ options }) => { + // Operator-visible turn continuity on the shared messages gateways. + const sessionId = optionString(options, OPENCODE_SESSION_ID_OPTION); + return sessionId === undefined ? {} : { "x-opencode-session": sessionId }; + }, + ); } // Both providers share the session-header wrapper above; keep both export diff --git a/src/provider/bifrost-adapter.ts b/src/provider/bifrost-adapter.ts index bb1b4ec79..f6b743f3b 100644 --- a/src/provider/bifrost-adapter.ts +++ b/src/provider/bifrost-adapter.ts @@ -1,9 +1,6 @@ -import { - CREDENTIAL_SENTINEL, - type BuiltRequest, - type ProviderAdapter, -} from "@intx/inference"; +import { CREDENTIAL_SENTINEL, type ProviderAdapter } from "@intx/inference"; import { createOpenAICompatibleAdapter } from "./openai-compatible-adapter.js"; +import { withExtraHeaders } from "./adapter-decorators.js"; // Small adapter for Bifrost (https://docs.getbifrost.ai). Bifrost is // OpenAI-compatible for chat but scopes requests to a virtual key's @@ -22,28 +19,10 @@ type AdapterSource = Parameters[0]; // streaming request without it with HTTP 426 — plus the providerOptions merge // and delta patching every other OpenAI-shaped source gets. export function createBifrostAdapter(source: AdapterSource): ProviderAdapter { - const base = createOpenAICompatibleAdapter(source); - - const buildRequest: ProviderAdapter["buildRequest"] = ( - messages, - model, - options, - ) => { - const req = base.buildRequest(messages, model, options); - return { - ...req, - headers: { - ...req.headers, - "x-bf-vk": CREDENTIAL_SENTINEL, - // The base adapter has already emitted authorization as the BEARER - // sentinel; spread keeps it, and gateways that only look at x-bf-vk - // are tolerated. - }, - } as BuiltRequest; - }; - - return { - ...base, - buildRequest, - }; + return withExtraHeaders(createOpenAICompatibleAdapter(source), () => ({ + "x-bf-vk": CREDENTIAL_SENTINEL, + // The base adapter has already emitted authorization as the BEARER + // sentinel; the spread keeps it, and gateways that only look at x-bf-vk + // are tolerated. + })); } diff --git a/src/provider/bounded-model-catalog.ts b/src/provider/bounded-model-catalog.ts index 9d61d435f..a3218e43b 100644 --- a/src/provider/bounded-model-catalog.ts +++ b/src/provider/bounded-model-catalog.ts @@ -1,14 +1,10 @@ -import { type } from "arktype"; - import { endpointErrorMessage, + parseModelsData, type ModelsEndpointDiscoveryState, - ModelsEndpointResponse, requestModelsEndpoint, } from "./models-endpoint.js"; -const CatalogModelsResponse = ModelsEndpointResponse; - export type CatalogDiscoveryState = ModelsEndpointDiscoveryState; function declaredCatalogBytes(response: Response): number | undefined { @@ -144,21 +140,10 @@ export function createBoundedModelCatalog(args: { if (!body.ok) { return { status: "malformed", message: body.message }; } - const parsed = CatalogModelsResponse(body.value); - if (parsed instanceof type.errors) { - return { status: "malformed", message: parsed.summary }; - } - if (parsed.data.length > maxModels) { - return { status: "malformed", message: oversizeMessage("models") }; - } - const models = [ - ...new Set( - parsed.data.map(({ id }) => id.trim()).filter((id) => id.length > 0), - ), - ]; - return models.length > 0 - ? { status: "models", models } - : { status: "empty" }; + return parseModelsData(body.value, { + maxModels, + oversizeMessage: oversizeMessage("models"), + }); } function selectableModelIds(): readonly string[] { diff --git a/src/provider/cache-ttl.ts b/src/provider/cache-ttl.ts index 584409046..3f0b36393 100644 --- a/src/provider/cache-ttl.ts +++ b/src/provider/cache-ttl.ts @@ -21,37 +21,24 @@ // prompt-caching the fold is not attempted. import { isOllamaProviderId } from "./ollama.js"; +import { + ANTHROPIC_MESSAGES_PROTOCOLS, + canonicalProviderSegment, +} from "./provider-protocols.js"; const MINUTE_MS = 60_000; /** Published Anthropic ephemeral TTL. The only idle-recompress window. */ const ANTHROPIC_TTL_MS = 5 * MINUTE_MS; -const ANTHROPIC_PROTOCOL = new Set([ - "anthropic", - "zen-messages", - "opencode-go-messages", -]); - type CacheTtlIdentity = { sourceId?: string; provider?: string; model?: string; }; -// Canonical provider segment of a `provider/model` string: the account or -// adapter name before the first "/" (custom names like `xai/alice` carry the -// provider there), else the head before ":". -function canonicalSegment(model: string): string { - const lower = model.toLowerCase(); - const slash = lower.indexOf("/"); - const head = slash >= 0 ? lower.slice(0, slash) : lower; - const colon = head.indexOf(":"); - return colon >= 0 ? head.slice(0, colon) : head; -} - function ttlForSegment(segment: string): number | undefined { - if (!ANTHROPIC_PROTOCOL.has(segment)) return undefined; + if (!ANTHROPIC_MESSAGES_PROTOCOLS.has(segment)) return undefined; return ANTHROPIC_TTL_MS; } @@ -71,7 +58,7 @@ export function cacheTtlMsFor( ): number | undefined { if (identity === undefined) return undefined; if (typeof identity === "string") - return ttlForSegment(canonicalSegment(identity)); + return ttlForSegment(canonicalProviderSegment(identity)); if (identity.sourceId !== undefined && isOllamaProviderId(identity.sourceId)) return undefined; @@ -79,8 +66,8 @@ export function cacheTtlMsFor( return undefined; if (identity.provider !== undefined && identity.provider.length > 0) - return ttlForSegment(canonicalSegment(identity.provider)); - return ttlForSegment(canonicalSegment(identity.model ?? "")); + return ttlForSegment(canonicalProviderSegment(identity.provider)); + return ttlForSegment(canonicalProviderSegment(identity.model ?? "")); } /** diff --git a/src/provider/codex-responses.test.ts b/src/provider/codex-responses.test.ts index e60b4581a..279387f5f 100644 --- a/src/provider/codex-responses.test.ts +++ b/src/provider/codex-responses.test.ts @@ -8,8 +8,8 @@ import { CODEX_RESPONSES_PROVIDER, CODEX_SESSION_ID_OPTION, createCodexResponsesAdapter, - withParallelToolCalls, } from "./codex-responses.js"; +import { withJsonBodyRewrite } from "./adapter-decorators.js"; const source: LastCycleSource = { sourceId: "codex/test", @@ -84,11 +84,10 @@ describe("createCodexResponsesAdapter", () => { body: "not-json", }), } as unknown as ProviderAdapter; - const request = withParallelToolCalls(stub).buildRequest( - [userTurn("hello")], - "gpt-5.1-codex", - {}, - ); + const request = withJsonBodyRewrite(stub, (body) => { + body["parallel_tool_calls"] = true; + return true; + }).buildRequest([userTurn("hello")], "gpt-5.1-codex", {}); expect(request.body).toBe("not-json"); }); diff --git a/src/provider/codex-responses.ts b/src/provider/codex-responses.ts index c913b7127..fc7b7ed2a 100644 --- a/src/provider/codex-responses.ts +++ b/src/provider/codex-responses.ts @@ -1,3 +1,4 @@ +import { type } from "arktype"; import { createCodexResponsesAdapter as createPackageCodexResponsesAdapter, CODEX_ACCOUNT_ID_OPTION, @@ -5,9 +6,9 @@ import { CODEX_SESSION_ID_OPTION, withCodexContentTypeRepair, } from "@corbits/codex-provider"; -import { type } from "arktype"; import type { AdapterFactory, ProviderAdapter } from "@intx/inference"; import type { InferenceOptions } from "@intx/types/runtime"; +import { withJsonBodyRewrite } from "./adapter-decorators.js"; import { ENVIRONMENT_TAG_NAME, PRODUCT_NAME } from "../branding.js"; export const CODEX_RESPONSES_PROVIDER = "codex-responses"; @@ -52,28 +53,6 @@ function withHostReasoningEffort(adapter: ProviderAdapter): ProviderAdapter { }; } -// The Codex backend accepts parallel_tool_calls:true, but the packaged -// adapter pins false; the host rewrites the field on the wire because the -// reactor already fans out a multi-call batch concurrently. -export function withParallelToolCalls( - adapter: ProviderAdapter, -): ProviderAdapter { - return { - ...adapter, - buildRequest: (messages, model, options) => { - const request = adapter.buildRequest(messages, model, options); - let body: Record; - try { - body = JSON.parse(request.body) as Record; - } catch { - return request; - } - body["parallel_tool_calls"] = true; - return { ...request, body: JSON.stringify(body) }; - }, - }; -} - const FunctionCallItemDone = type({ type: "'response.output_item.done'", item: { type: "'function_call'", call_id: "string", arguments: "string" }, @@ -133,11 +112,18 @@ function withArgumentsFromDone(adapter: ProviderAdapter): ProviderAdapter { // CodexQuirks are required by the package factory; Responses option keys are // not part of that bag, so host reasoning_effort is aliased at buildRequest. +// The parallel_tool_calls rewrite exists because the packaged adapter pins +// false while the reactor already fans out a multi-call batch concurrently. export const createCodexResponsesAdapter: AdapterFactory = (source) => withHostReasoningEffort( - withParallelToolCalls( + withJsonBodyRewrite( withArgumentsFromDone( createPackageCodexResponsesAdapter(source, CODEX_QUIRKS), ), + // Codex accepts parallel_tool_calls:true; pin the field on the wire. + (body) => { + body["parallel_tool_calls"] = true; + return true; + }, ), ); diff --git a/src/provider/deepseek-v4-adapter.ts b/src/provider/deepseek-v4-adapter.ts index 36a8b25c6..cc740798b 100644 --- a/src/provider/deepseek-v4-adapter.ts +++ b/src/provider/deepseek-v4-adapter.ts @@ -1,5 +1,6 @@ import { type ProviderAdapter } from "@intx/inference"; import { createOpenAICompatibleAdapter } from "./openai-compatible-adapter.js"; +import { withJsonBodyRewrite } from "./adapter-decorators.js"; import { DEEPSEEK_V4_MODEL_CARD, isDeepSeekModel, @@ -22,42 +23,36 @@ export function createDeepSeekV4Adapter( ): ProviderAdapter { const base = createOpenAICompatibleAdapter(source); - const buildRequest: ProviderAdapter["buildRequest"] = ( - messages, - model, - options, - ) => { - const req = base.buildRequest(messages, model, options); - if (!isDeepSeekModel(model) || !isDeepSeekV4Model(model)) return req; - const body = JSON.parse(req.body) as Record; - - // V4 effort ladder (none/xhigh/max): none drops reasoning_effort and - // turns thinking off; xhigh/max go out raw with thinking on. Off-ladder - // efforts throw inside mapV4Effort (rejected, never passed to the wire). - if ("reasoning_effort" in body) { - const effort = body["reasoning_effort"] as Parameters< - typeof mapV4Effort - >[0]; - const wire = mapV4Effort(effort); - if (wire === null) { - delete body["reasoning_effort"]; - mergeThinking(body, false); - } else { - body["reasoning_effort"] = wire; - mergeThinking(body, true); + return withJsonBodyRewrite( + base, + (body) => { + // V4 effort ladder (none/xhigh/max): none drops reasoning_effort and + // turns thinking off; xhigh/max go out raw with thinking on. Off-ladder + // efforts throw inside mapV4Effort (rejected, never passed to the wire). + if ("reasoning_effort" in body) { + const effort = body["reasoning_effort"] as Parameters< + typeof mapV4Effort + >[0]; + const wire = mapV4Effort(effort); + if (wire === null) { + delete body["reasoning_effort"]; + mergeThinking(body, false); + } else { + body["reasoning_effort"] = wire; + mergeThinking(body, true); + } } - } - - if (body["top_p"] === undefined) - body["top_p"] = DEEPSEEK_V4_MODEL_CARD.topP; - if (body["temperature"] === undefined) - body["temperature"] = DEEPSEEK_V4_MODEL_CARD.temperature; - body["stream_options"] ??= V4_STREAM_OPTIONS.stream_options; - - return { ...req, body: JSON.stringify(body) }; - }; - return { ...base, buildRequest }; + if (body["top_p"] === undefined) + body["top_p"] = DEEPSEEK_V4_MODEL_CARD.topP; + if (body["temperature"] === undefined) + body["temperature"] = DEEPSEEK_V4_MODEL_CARD.temperature; + body["stream_options"] ??= V4_STREAM_OPTIONS.stream_options; + return true; + }, + // Zero-parse fast path: pre-V4 DeepSeek keeps the generic body. + (_request, { model }) => isDeepSeekModel(model) && isDeepSeekV4Model(model), + ); } // chat_template_kwargs clone with an absent-only thinking toggle (never mutate diff --git a/src/provider/error-body.ts b/src/provider/error-body.ts new file mode 100644 index 000000000..3e9dec0e9 --- /dev/null +++ b/src/provider/error-body.ts @@ -0,0 +1,75 @@ +/** + * Shared JSON error-body reader behind the structured provider error + * classifiers (Codex usage limits, gateway rate-limit normalization): + * coerce a raw string-or-object body, locate the error node across the + * shapes providers and gateways emit, and read body-carried reset windows. + * The matchers that consume a node stay per-provider (Codex exact codes, + * xAI substring markers); only the body-walking mechanics are shared here. + */ + +function asRecord(value: unknown): Record | undefined { + if (typeof value !== "object" || value === null || Array.isArray(value)) + return undefined; + return value as Record; +} + +export function tryParseJSON(text: string): unknown { + try { + return JSON.parse(text) as unknown; + } catch { + return undefined; + } +} + +/** String bodies parse; non-empty non-JSON strings pass through untouched so + * message-only classifiers still see them. */ +export function coerceBody(raw: unknown): unknown { + if (typeof raw === "string") { + const trimmed = raw.trim(); + if (trimmed.length === 0) return undefined; + return tryParseJSON(trimmed) ?? raw; + } + return raw; +} + +/** + * Locate the error node in a coerced body: nested `detail.error` (Codex + * live shape), fields directly under `detail`, top-level `error`, or a + * flat `code` on the root. Undefined for non-JSON bodies without one. + */ +export function readErrorNode( + body: unknown, +): Record | undefined { + const root = asRecord(body); + if (root === undefined) return undefined; + + const detail = asRecord(root["detail"]); + if (detail !== undefined) { + const nested = asRecord(detail["error"]); + if (nested !== undefined) return nested; + // Some gateways put the fields directly under detail. + if (typeof detail["code"] === "string") return detail; + } + + const top = asRecord(root["error"]); + if (top !== undefined) return top; + + if (typeof root["code"] === "string") return root; + return undefined; +} + +/** Body-carried reset window (`resets_in_seconds` et al.); undefined unless a + * finite, non-negative number. */ +export function resetSecondsFromNode( + node: Record, +): number | undefined { + const resetsRaw = + node["resets_in_seconds"] ?? + node["resetsInSeconds"] ?? + node["reset_after_seconds"]; + return typeof resetsRaw === "number" && + Number.isFinite(resetsRaw) && + resetsRaw >= 0 + ? Math.floor(resetsRaw) + : undefined; +} diff --git a/src/provider/models-endpoint.ts b/src/provider/models-endpoint.ts index ce9d67410..5b27953ca 100644 --- a/src/provider/models-endpoint.ts +++ b/src/provider/models-endpoint.ts @@ -27,6 +27,31 @@ export function endpointErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } +/** + * Validate a `/models` response body and project it to its discovery state. + * Dedup happens by trimmed id; an optional cap rebrands an oversized catalog + * as malformed instead of admitting it. Callers keep their own HTTP and + * transport failure lines; this only owns the body. + */ +export function parseModelsData( + raw: unknown, + cap?: { readonly maxModels: number; readonly oversizeMessage: string }, +): ModelsEndpointDiscoveryState { + const parsed = ModelsEndpointResponse(raw); + if (parsed instanceof type.errors) { + return { status: "malformed", message: parsed.summary }; + } + if (cap !== undefined && parsed.data.length > cap.maxModels) { + return { status: "malformed", message: cap.oversizeMessage }; + } + const models = [ + ...new Set( + parsed.data.map(({ id }) => id.trim()).filter((id) => id.length > 0), + ), + ]; + return models.length > 0 ? { status: "models", models } : { status: "empty" }; +} + // Single GET against an OpenAI-compatible /models endpoint, so URL // normalization and the request timeout stay consistent across callers. export async function requestModelsEndpoint(args: { diff --git a/src/provider/ollama.ts b/src/provider/ollama.ts index 058d2c9fc..3d2e3ef89 100644 --- a/src/provider/ollama.ts +++ b/src/provider/ollama.ts @@ -1,9 +1,7 @@ -import { type } from "arktype"; - import { endpointErrorMessage, + parseModelsData, type ModelsEndpointDiscoveryState, - ModelsEndpointResponse, requestModelsEndpoint, } from "./models-endpoint.js"; @@ -41,8 +39,6 @@ export function ollamaOpenAIBaseURL(rootURL: string): string { return `${normalizeOllamaRootURL(rootURL)}/v1`; } -const OllamaModelsResponse = ModelsEndpointResponse; - export type OllamaDiscoveryState = ModelsEndpointDiscoveryState; /** Discover installed Ollama models without leaking transport or parsing failures. */ @@ -92,16 +88,7 @@ export async function discoverOllamaModels(args: { error instanceof Error ? error.message : "Ollama returned invalid JSON", }; } - const parsed = OllamaModelsResponse(raw); - if (parsed instanceof type.errors) { - return { status: "malformed", message: parsed.summary }; - } - const models = [ - ...new Set( - parsed.data.map(({ id }) => id.trim()).filter((id) => id.length > 0), - ), - ]; - return models.length > 0 ? { status: "models", models } : { status: "empty" }; + return parseModelsData(raw); } /** Operator-facing line for a discovery failure. Network stays canned; HTTP diff --git a/src/provider/openai-compatible-adapter.ts b/src/provider/openai-compatible-adapter.ts index 27ee4b803..99e3f618d 100644 --- a/src/provider/openai-compatible-adapter.ts +++ b/src/provider/openai-compatible-adapter.ts @@ -2,6 +2,7 @@ import { type BuiltRequest, type ProviderAdapter } from "@intx/inference"; import { createOpenAIAdapter } from "@intx/inference/providers"; import { normalizeNullDeltaFields } from "./null-delta-fields.js"; import { isDeepSeekModel, isDeepSeekV4Model } from "./deepseek-v4-effort.js"; +import { withJsonBodyRewrite } from "./adapter-decorators.js"; // The stock OpenAI adapter builds the request body from a fixed set of fields // and ignores `options.providerOptions`, so provider-native knobs carried @@ -18,10 +19,6 @@ export function createOpenAICompatibleAdapter( quirks?: unknown, ): ProviderAdapter { const base = createOpenAIAdapter(source, quirks); - // Set by buildRequest for the model the current request targets; only - // DeepSeek/NIM streams need the null-delta-field patch below, so other - // providers' frames skip the reparse and hit base.parseResponse once. - let needsDeepSeekPatch = false; const ensureAccept = (req: BuiltRequest): BuiltRequest => { const has = req.headers.Accept || req.headers.accept; @@ -32,37 +29,55 @@ export function createOpenAICompatibleAdapter( }; }; - const buildRequest: ProviderAdapter["buildRequest"] = ( - messages, - model, - options, - ) => { - const built = base.buildRequest(messages, model, options); - const providerOptions = options.providerOptions; - const hasProviderOptions = - providerOptions !== undefined && Object.keys(providerOptions).length > 0; - // DeepSeek returns HTTP 400 if `reasoning_content` appears in input messages, - // whereas the base adapter emits it for any model with thinking enabled. V4 - // keeps each turn's reasoning on replay (handled by its own adapter), so only - // pre-V4 DeepSeek strips it here. - const m = model.toLowerCase().trim(); - const isDeepSeek = isDeepSeekModel(m); - const stripReasoning = isDeepSeek && !isDeepSeekV4Model(m); - needsDeepSeekPatch = isDeepSeek; - if (!hasProviderOptions && !stripReasoning) return ensureAccept(built); + // Set by the rewrite gate for the model the current request targets; only + // DeepSeek/NIM streams need the null-delta-field patch below, so other + // providers' frames skip the reparse and hit base.parseResponse once. + let needsDeepSeekPatch = false; - const body = JSON.parse(built.body) as Record; - if (hasProviderOptions) Object.assign(body, providerOptions); + const buildRequest = withJsonBodyRewrite( + { + ...base, + buildRequest: (messages, model, options) => + ensureAccept(base.buildRequest(messages, model, options)), + }, + (body, { model, options }) => { + const providerOptions = options.providerOptions; + const hasProviderOptions = + providerOptions !== undefined && + Object.keys(providerOptions).length > 0; + if (hasProviderOptions) Object.assign(body, providerOptions); - if (stripReasoning && Array.isArray(body["messages"])) { - for (const msg of body["messages"]) { - if (msg !== null && typeof msg === "object") - delete (msg as Record)["reasoning_content"]; + // DeepSeek returns HTTP 400 if `reasoning_content` appears in input + // messages, whereas the base adapter emits it for any model with + // thinking enabled. V4 keeps each turn's reasoning on replay (handled + // by its own adapter), so only pre-V4 DeepSeek strips it here. + const m = model.toLowerCase().trim(); + if ( + isDeepSeekModel(m) && + !isDeepSeekV4Model(m) && + Array.isArray(body["messages"]) + ) { + for (const msg of body["messages"]) { + if (msg !== null && typeof msg === "object") + delete (msg as Record)["reasoning_content"]; + } } - } - const merged: BuiltRequest = { ...built, body: JSON.stringify(body) }; - return ensureAccept(merged); - }; + return true; + }, + // Zero-parse fast path: requests with nothing to merge and nothing to + // strip return the inner request untouched (byte-identical). + (_request, { model, options }) => { + const providerOptions = options.providerOptions; + const hasProviderOptions = + providerOptions !== undefined && + Object.keys(providerOptions).length > 0; + const m = model.toLowerCase().trim(); + const isDeepSeek = isDeepSeekModel(m); + needsDeepSeekPatch = isDeepSeek; + if (hasProviderOptions) return true; + return isDeepSeek && !isDeepSeekV4Model(m); + }, + ); // DeepSeek via NVIDIA NIM sends null for delta fields the upstream schema // requires to be non-null; other providers' frames skip the reparse and @@ -72,5 +87,5 @@ export function createOpenAICompatibleAdapter( return base.parseResponse(normalizeNullDeltaFields(sseData)); }; - return { ...base, buildRequest, parseResponse }; + return { ...buildRequest, parseResponse }; } diff --git a/src/provider/provider-protocols.ts b/src/provider/provider-protocols.ts new file mode 100644 index 000000000..9f55a8de9 --- /dev/null +++ b/src/provider/provider-protocols.ts @@ -0,0 +1,30 @@ +/** + * Canonical provider/protocol identities shared by the cache consumers + * (`cache-ttl.ts`, `anthropic-cache-breakpoint.ts`) so the TTL window and + * the breakpoint pass cannot disagree about which providers speak the + * Anthropic messages protocol. A new session-header wrapper joins both by + * adding one entry. + */ + +import { + OPENCODE_GO_MESSAGES_PROVIDER, + ZEN_MESSAGES_PROVIDER, +} from "./anthropic-session-adapter.js"; + +export const ANTHROPIC_MESSAGES_PROTOCOLS: ReadonlySet = new Set([ + "anthropic", + ZEN_MESSAGES_PROVIDER, + OPENCODE_GO_MESSAGES_PROVIDER, +]); + +// Canonical provider segment of a `provider/model` string: the account or +// adapter name before the first "/" (custom names like `xai/alice` carry the +// provider there), else the head before ":". Lowercased; catalog ids carry +// no consistent case. +export function canonicalProviderSegment(value: string): string { + const lower = value.toLowerCase(); + const slash = lower.indexOf("/"); + const head = slash >= 0 ? lower.slice(0, slash) : lower; + const colon = head.indexOf(":"); + return colon >= 0 ? head.slice(0, colon) : head; +} diff --git a/src/provider/replay-sanitizer.ts b/src/provider/replay-sanitizer.ts index 554c48a6f..4db06e904 100644 --- a/src/provider/replay-sanitizer.ts +++ b/src/provider/replay-sanitizer.ts @@ -1,5 +1,6 @@ import { transformMessages, type AdapterRegistry } from "@intx/inference"; import type { ContentBlock, ConversationTurn } from "@intx/types/runtime"; +import { withDecoratedAdapters } from "./adapter-decorators.js"; // Repairs persisted history at the request-build boundary so a turn produced // by one provider replays safely against another. transformMessages (vendored) @@ -98,19 +99,9 @@ export function sanitizeReplayTurns( export function withReplaySanitizer( adapters: AdapterRegistry, ): AdapterRegistry { - return { - has: (provider) => adapters.has(provider), - resolve(source, quirks) { - const adapter = adapters.resolve(source, quirks); - return { - ...adapter, - buildRequest: (turns, model, options) => - adapter.buildRequest( - sanitizeReplayTurns(turns, model), - model, - options, - ), - }; - }, - }; + return withDecoratedAdapters(adapters, (adapter) => ({ + ...adapter, + buildRequest: (turns, model, options) => + adapter.buildRequest(sanitizeReplayTurns(turns, model), model, options), + })); }