diff --git a/src/gateway/model-pricing-cache.test.ts b/src/gateway/model-pricing-cache.test.ts index 08b2b47d305e..921bd28cfbe2 100644 --- a/src/gateway/model-pricing-cache.test.ts +++ b/src/gateway/model-pricing-cache.test.ts @@ -454,6 +454,45 @@ describe("model-pricing-cache", () => { }); }); + it("cancels remote pricing error response bodies", async () => { + const config = { + agents: { + defaults: { + model: { primary: "custom/gpt-remote" }, + }, + }, + models: { + providers: { + custom: { + baseUrl: "https://models.example/v1", + api: "openai-completions", + models: [{ id: "gpt-remote" }], + }, + }, + }, + } as unknown as OpenClawConfig; + const openRouterResponse = new Response("rate limited", { status: 429 }); + const cancel = vi.spyOn(openRouterResponse.body!, "cancel").mockResolvedValue(undefined); + const fetchImpl = withFetchPreconnect(async (input: RequestInfo | URL) => { + const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; + if (url.includes("openrouter.ai")) { + return openRouterResponse; + } + return new Response(JSON.stringify({}), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + }); + + await refreshGatewayModelPricingCache({ config, fetchImpl }); + + expect(cancel).toHaveBeenCalledOnce(); + const health = getGatewayModelPricingHealth(); + expect(health.state).toBe("degraded"); + expect(health.sources[0]?.source).toBe("openrouter"); + expect(health.sources[0]?.detail).toContain("HTTP 429"); + }); + it("records malformed remote pricing catalog JSON as source failures", async () => { const config = { agents: { @@ -1222,6 +1261,7 @@ describe("model-pricing-cache", () => { }, } as unknown as OpenClawConfig; + const liteLLMCancel = vi.fn(async () => undefined); const fetchImpl = withFetchPreconnect(async (input: RequestInfo | URL) => { const url = typeof input === "string" ? input : input instanceof URL ? input.href : input.url; if (url.includes("openrouter.ai")) { @@ -1244,17 +1284,20 @@ describe("model-pricing-cache", () => { }, ); } - return new Response("{}", { + const liteLLMResponse = new Response("{}", { status: 200, headers: { "Content-Type": "application/json", "Content-Length": "6000000", }, }); + vi.spyOn(liteLLMResponse.body!, "cancel").mockImplementation(liteLLMCancel); + return liteLLMResponse; }); await refreshGatewayModelPricingCache({ config, fetchImpl }); + expect(liteLLMCancel).toHaveBeenCalledOnce(); expect(getCachedGatewayModelPricing({ provider: "kimi", model: "kimi-k2.6" })).toEqual({ input: 0.95, output: 4, diff --git a/src/gateway/model-pricing-cache.ts b/src/gateway/model-pricing-cache.ts index 107a29e62f78..977fd639de0f 100644 --- a/src/gateway/model-pricing-cache.ts +++ b/src/gateway/model-pricing-cache.ts @@ -275,6 +275,12 @@ function toCachedModelPricing( }; } +async function cancelUnreadResponseBody(response: Response | undefined): Promise { + if (response?.bodyUsed !== true) { + await response?.body?.cancel().catch(() => undefined); + } +} + async function readPricingJsonObject( response: Response, source: string, @@ -299,6 +305,28 @@ async function readPricingJsonObject( return payload as Record; } +async function fetchPricingJsonObject(params: { + fetchImpl: typeof fetch; + url: string; + source: string; + failureLabel: string; + signal?: AbortSignal; +}): Promise> { + let response: Response | undefined; + try { + response = await params.fetchImpl(params.url, { + headers: { Accept: "application/json" }, + signal: createPricingFetchSignal(params.signal), + }); + if (!response.ok) { + throw new Error(`${params.failureLabel}: HTTP ${response.status}`); + } + return await readPricingJsonObject(response, params.source); + } finally { + await cancelUnreadResponseBody(response); + } +} + // --------------------------------------------------------------------------- // LiteLLM tiered-pricing parsing // --------------------------------------------------------------------------- @@ -383,14 +411,13 @@ async function fetchLiteLLMPricingCatalog( fetchImpl: typeof fetch, signal?: AbortSignal, ): Promise { - const response = await fetchImpl(LITELLM_PRICING_URL, { - headers: { Accept: "application/json" }, - signal: createPricingFetchSignal(signal), + const payload = await fetchPricingJsonObject({ + fetchImpl, + url: LITELLM_PRICING_URL, + source: "LiteLLM", + failureLabel: "LiteLLM pricing fetch failed", + signal, }); - if (!response.ok) { - throw new Error(`LiteLLM pricing fetch failed: HTTP ${response.status}`); - } - const payload = await readPricingJsonObject(response, "LiteLLM"); const catalog: LiteLLMPricingCatalog = new Map(); for (const [key, value] of Object.entries(payload)) { if (!value || typeof value !== "object") { @@ -1059,14 +1086,13 @@ async function fetchOpenRouterPricingCatalog( fetchImpl: typeof fetch, signal?: AbortSignal, ): Promise> { - const response = await fetchImpl(OPENROUTER_MODELS_URL, { - headers: { Accept: "application/json" }, - signal: createPricingFetchSignal(signal), + const payload = await fetchPricingJsonObject({ + fetchImpl, + url: OPENROUTER_MODELS_URL, + source: "OpenRouter", + failureLabel: "OpenRouter /models failed", + signal, }); - if (!response.ok) { - throw new Error(`OpenRouter /models failed: HTTP ${response.status}`); - } - const payload = await readPricingJsonObject(response, "OpenRouter"); const entries = Array.isArray(payload.data) ? payload.data : []; const catalog = new Map(); for (const entry of entries) {