From 40793e6f8b5d9362e9b94d9e18cd95581c6bfc18 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 13:54:56 +0100 Subject: [PATCH] fix: bound pasted input and provider response bodies (#110627) * refactor(runtime): centralize bounded I/O handling * fix(runtime): finalize bounded I/O contracts * chore(security): drop stale audit import * docs(changelog): credit bounded I/O fixes Co-authored-by: Pick-cat * fix(agents): keep provider HTTP options private * fix(ai): traverse OpenAI schema maps by value * fix(runtime): preserve absolute download deadlines * ci(ui): stabilize compressed CSS budget * fix(ai): traverse legacy schema applicators --------- Co-authored-by: Pick-cat --- CHANGELOG.md | 1 + .../.generated/plugin-sdk-api-baseline.sha256 | 8 +- .../video-generation-provider.test.ts | 124 ++++++++- .../byteplus/video-generation-provider.ts | 70 +++-- extensions/matrix/src/cli.ts | 19 +- .../minimax/music-generation-provider.test.ts | 27 +- .../minimax/music-generation-provider.ts | 92 ++----- .../minimax/provider-http.test-helpers.ts | 44 ++- .../minimax/video-generation-provider.ts | 61 ++++- .../openai/video-generation-provider.ts | 32 ++- .../runway/video-generation-provider.ts | 73 +++-- .../synology-chat/src/client.loopback.test.ts | 48 ++-- .../together/video-generation-provider.ts | 15 +- extensions/vydra/shared.test.ts | 29 +- extensions/vydra/shared.ts | 137 ++++------ .../xai/video-generation-provider.test.ts | 3 +- extensions/xai/video-generation-transport.ts | 17 +- packages/ai/src/internal/openai.ts | 1 + .../providers/openai-tool-schema-compat.ts | 245 +++++++++++++++++ .../src/providers/openai-tool-schema.test.ts | 49 ++++ .../ai/src/providers/openai-tool-schema.ts | 71 +---- .../src/read-byte-stream-with-limit.test.ts | 5 +- .../src/read-byte-stream-with-limit.ts | 9 +- scripts/check-control-ui-performance.mjs | 6 +- src/agents/provider-http-errors.test.ts | 50 ++++ src/agents/provider-http-errors.ts | 96 +++++-- src/cli/config-cli.ts | 22 +- src/cli/exec-approvals-cli.ts | 17 +- src/commands/models/auth.ts | 17 +- src/infra/http-body.response.test.ts | 43 +++ src/infra/http-body.ts | 14 +- src/media-understanding/shared.test.ts | 83 +++++- src/media-understanding/shared.ts | 123 +++++++-- src/music-generation/provider-assets.ts | 21 +- src/plugin-sdk/provider-tools.test.ts | 43 +++ src/plugin-sdk/provider-tools.ts | 255 +----------------- .../test-helpers/provider-http-mocks.ts | 44 ++- test/scripts/control-ui-performance.test.ts | 43 +++ 38 files changed, 1289 insertions(+), 768 deletions(-) create mode 100644 packages/ai/src/providers/openai-tool-schema-compat.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 75763220837..14ce3db2188 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,7 @@ Docs: https://docs.openclaw.ai ### Fixes +- **Bounded input and provider responses:** cap pasted auth/config input and enforce wall-clock deadlines across generated-media downloads, polling JSON, and failed response details so oversized or slow-drip streams cannot exceed resource budgets (thanks @Pick-cat). - **Cloud worker derived workspace caches:** exclude Python caches, dependency trees, and macOS metadata symmetrically from outbound sync and inbound reconciliation so local cache rewrites cannot fence later cloud results or worker reclaim. - **Codex model status diagnostics:** report a configured Codex route as unavailable when its harness plugin is disabled, missing, or quarantined, while preserving the separate credential result and making `models status --check` fail instead of silently treating fallback execution as healthy. Thanks @shakkernerd. - **Gateway control-plane rate limiting:** use per-method buckets with a 30-per-minute budget so interactive admin writes remain responsive while retaining runaway-loop protection. diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 7b67958017e..9648715b86e 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -215,7 +215,7 @@ b32f041136217c510e559e5fcf7ed4361cf51ebfad0fa15799d69fe539780feb module/provide ee5184ab251bcbe49a4cb9c845eee31975733bce1dc2300854cff14e99cf6db7 module/provider-catalog-shared ca5d408937bc0ff92021673320aeba6ae6fce705711b251f15797887b8f34d3a module/provider-entry 52085bfaa917f4fd262508a93af923e5e834901b5f0f017c93b02a0798073344 module/provider-env-vars -404840e551472a1debea4e7ff26c6df99e219bfcfdaf7491553eed0e843f77bf module/provider-http +3b311719eae8dbb061ac2e79f0a58f8540f7a416ec281f3ee0c1cc3cfa6b401f module/provider-http 47564e55d5ae9174fc402d6a1ef7120a799a92bb3625106327b642d4d9eb3bd9 module/provider-model-shared 966830793fccc5e5a7e6ae2f7df84f5f39120ac9a6af8c10be1e070eaf257f32 module/provider-model-types f1452aec01b1bc1d8764b666131b6237bce5b78477ce1137d622c4aa0b691a8e module/provider-oauth-runtime @@ -225,7 +225,7 @@ c543747f32aebde42508e31606db325739f0c54fcc7cf683f68ac941c835f737 module/provide 413af0d4597c1bab45e2f1a260b520445a25b9d11abd46a98fdfb74c60ddd601 module/provider-stream ea8bd63655e983dacfd306fe77d0793097166228a2dabd9b7fc8e33252eb359f module/provider-stream-family cdcc81cbd0aff7e341cd7edeec77b02b55d0d27af03278d13ef07487aa8eeae7 module/provider-stream-shared -4a30f74ae2706ec08d26ed37c86c3fb07ca03bba9ae3bff3a1e4375f8cfa0fbb module/provider-tools +5a907292055c941ec7420163ea50e8bc90ead816b3c5867d94a23024d3b579e8 module/provider-tools 8236935f56423a26aeb7219173c0c8c8c7aacb4a13d0d851f05bdf2d35790cd4 module/provider-transport-runtime 009a594b19aa9f0d0c0c28f54968f807cad4b3ee40ec13c685367116d63aafdc module/provider-usage b6c43bdb895139e6bbdb30f861b7e0e66bd2e8960e4129cc2bdb562c575c5f2f module/provider-web-fetch @@ -284,8 +284,8 @@ cd431f6ba8327b81438b7a63b1963120f200f5abd145fb6aa7c5c561339cb0b1 module/setup-t f2095165f836537aa6c145aecfff32543c9cc7b3935d33d107869b5f975475ee module/simple-completion-runtime 18e384ec43d9eaee52c8e286e127bda2048370e2337964a754d94b236724ca9e module/skill-commands-runtime 7eaea43619ee15e8b0d2d17c81c63b057576b251fecebcb5034d4e06e5940aea module/skills-runtime -91e261cfc2a847bb6adc7147027b2f7a83acd6207b51b225b4452742104312da module/speech -08d11b92b5947cf43636fa58c0b1505d1f656106d22653f6b0fd14020ca839f2 module/speech-core +6639d26ccc022ee5c55e23aa52071e30dbbddb17e7c4ecf62b152358380e087f module/speech +5c228af0457bf14b2833c2d0994c98212f6a7451abec7d5fe5cad0632115bb84 module/speech-core ae469f32799380e6b045abaefefee6eb3f00d714ffbf36b6eeef5025dc529472 module/speech-settings ba7bd1528069e8d1dc6a6fc3626aebb0cd8f8d3f7715dba0916d315ea3787b2a module/sqlite-runtime e1379adab6398cdeda7671a4e6cc268508d21270815b53fef793249698a8d601 module/ssrf-dispatcher diff --git a/extensions/byteplus/video-generation-provider.test.ts b/extensions/byteplus/video-generation-provider.test.ts index bf8ae396c70..cb98e9d96d2 100644 --- a/extensions/byteplus/video-generation-provider.test.ts +++ b/extensions/byteplus/video-generation-provider.test.ts @@ -19,36 +19,87 @@ vi.mock("openclaw/plugin-sdk/provider-auth-runtime", () => ({ vi.mock("openclaw/plugin-sdk/provider-http", async (importActual) => { const actual = await importActual(); - const resolveTimeoutMs = (timeoutMs: unknown): number => - typeof timeoutMs === "function" ? (timeoutMs() as number) : ((timeoutMs as number) ?? 60_000); return { // REAL byte-bounded JSON reader under test — not stubbed. readProviderJsonResponse: actual.readProviderJsonResponse, postJsonRequest: postJsonRequestMock, - fetchProviderOperationResponse: async (params: { + pollProviderOperationJson: async (params: { url: string; - init?: RequestInit; - timeoutMs?: unknown; - fetchFn: typeof fetch; - }) => fetchWithTimeoutMock(params.url, params.init ?? {}, resolveTimeoutMs(params.timeoutMs)), + headers: Headers; + defaultTimeoutMs: number; + maxAttempts: number; + requestFailedMessage: string; + timeoutMessage: string; + isComplete: (payload: unknown) => boolean; + getFailureMessage?: (payload: unknown) => string | undefined; + }) => { + for (let attempt = 0; attempt < params.maxAttempts; attempt += 1) { + const response = await fetchWithTimeoutMock( + params.url, + { method: "GET", headers: params.headers }, + params.defaultTimeoutMs, + ); + const payload = await actual.readProviderJsonResponse( + response, + params.requestFailedMessage, + ); + if (params.isComplete(payload)) { + return payload; + } + const failureMessage = params.getFailureMessage?.(payload); + if (failureMessage) { + throw new Error(failureMessage); + } + } + throw new Error(params.timeoutMessage); + }, fetchProviderDownloadResponse: async (params: { url: string; init?: RequestInit; - timeoutMs?: unknown; + deadline: { deadlineAtMs?: number; timeoutMs?: number }; fetchFn: typeof fetch; - }) => fetchWithTimeoutMock(params.url, params.init ?? {}, resolveTimeoutMs(params.timeoutMs)), + }) => + fetchWithTimeoutMock( + params.url, + params.init ?? {}, + params.deadline.deadlineAtMs === undefined + ? (params.deadline.timeoutMs ?? 60_000) + : Math.max(1, params.deadline.deadlineAtMs - Date.now()), + ), assertOkOrThrowHttpError: async () => {}, createProviderOperationDeadline: ({ label, timeoutMs, }: { label: string; - timeoutMs?: number; - }) => ({ label, timeoutMs }), + timeoutMs?: number | (() => number); + }) => { + const resolvedTimeoutMs = typeof timeoutMs === "function" ? timeoutMs() : timeoutMs; + return { + label, + timeoutMs: resolvedTimeoutMs, + deadlineAtMs: + typeof resolvedTimeoutMs === "number" ? Date.now() + resolvedTimeoutMs : undefined, + }; + }, createProviderOperationTimeoutResolver: - ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) => - () => + ({ + deadline, defaultTimeoutMs, + }: { + deadline: { deadlineAtMs?: number; label: string; timeoutMs?: number }; + defaultTimeoutMs: number; + }) => + () => { + if (typeof deadline.deadlineAtMs !== "number") { + return defaultTimeoutMs; + } + const remainingMs = deadline.deadlineAtMs - Date.now(); + if (remainingMs <= 0) { + throw new Error(`${deadline.label} timed out after ${deadline.timeoutMs}ms`); + } + return Math.min(defaultTimeoutMs, remainingMs); + }, resolveProviderOperationTimeoutMs: ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) => defaultTimeoutMs, resolveProviderHttpRequestConfig: (params: { @@ -62,7 +113,6 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importActual) => { headers: new Headers(params.defaultHeaders), dispatcherPolicy: undefined, }), - waitProviderOperationPollInterval: async () => {}, }; }); @@ -76,6 +126,7 @@ afterEach(() => { postJsonRequestMock.mockReset(); fetchWithTimeoutMock.mockReset(); resolveApiKeyForProviderMock.mockClear(); + vi.useRealTimers(); }); function mockSuccessfulBytePlusTask(params?: { model?: string }) { @@ -246,6 +297,51 @@ describe("byteplus video generation provider", () => { ).rejects.toThrow("BytePlus generated video download exceeds 1 bytes"); }); + it("shares one wall-clock deadline across download headers and body", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + postJsonRequestMock.mockResolvedValue({ + response: streamedJsonResponse({ id: "task_slow_download" }), + release: vi.fn(async () => {}), + }); + fetchWithTimeoutMock + .mockResolvedValueOnce( + streamedJsonResponse({ + id: "task_slow_download", + status: "succeeded", + content: { video_url: "https://example.com/slow.mp4" }, + }), + ) + .mockImplementationOnce(async () => { + vi.setSystemTime(1_090); + return new Response( + new ReadableStream({ + async pull(controller) { + controller.enqueue(new Uint8Array([1])); + await new Promise((resolve) => { + setTimeout(resolve, 20); + }); + }, + }), + { headers: { "content-type": "video/mp4" } }, + ); + }); + + const result = buildBytePlusVideoGenerationProvider().generateVideo({ + provider: "byteplus", + model: "seedance-1-0-pro-250528", + prompt: "slow download", + timeoutMs: 100, + cfg: {}, + }); + const assertion = expect(result).rejects.toThrow( + "BytePlus generated video download timed out after 100ms", + ); + + await vi.advanceTimersByTimeAsync(11); + await assertion; + }); + it("keeps the unified model for image requests and lowercases resolution", async () => { mockSuccessfulBytePlusTask({ model: "seedance-1-0-pro-250528" }); diff --git a/extensions/byteplus/video-generation-provider.ts b/extensions/byteplus/video-generation-provider.ts index 0f417f0f568..0d5bbd48220 100644 --- a/extensions/byteplus/video-generation-provider.ts +++ b/extensions/byteplus/video-generation-provider.ts @@ -10,12 +10,11 @@ import { createProviderOperationDeadline, createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, - fetchProviderOperationResponse, + pollProviderOperationJson, postJsonRequest, readProviderJsonResponse, resolveProviderOperationTimeoutMs, resolveProviderHttpRequestConfig, - waitProviderOperationPollInterval, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; @@ -163,40 +162,24 @@ async function pollBytePlusTask(params: { timeoutMs: params.timeoutMs, label: `BytePlus video generation task ${params.taskId}`, }); - for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { - const response = await fetchProviderOperationResponse({ - stage: "poll", - url: `${params.baseUrl}/contents/generations/tasks/${params.taskId}`, - init: { - method: "GET", - headers: params.headers, - }, - timeoutMs: createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: DEFAULT_TIMEOUT_MS, - }), - fetchFn: params.fetchFn, - provider: "byteplus", - requestFailedMessage: "BytePlus video status request failed", - }); - const payload = await readBytePlusJsonResponse( - response, - "BytePlus video status request failed", - ); - switch (readBytePlusTaskStatus(payload)) { - case "succeeded": - return payload; - case "failed": - case "cancelled": - throw new Error( - readBytePlusErrorMessage(payload.error) || "BytePlus video generation failed", - ); - default: - await waitProviderOperationPollInterval({ deadline, pollIntervalMs: POLL_INTERVAL_MS }); - break; - } - } - throw new Error(`BytePlus video generation task ${params.taskId} did not finish in time`); + return await pollProviderOperationJson({ + url: `${params.baseUrl}/contents/generations/tasks/${params.taskId}`, + headers: params.headers, + deadline, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, + fetchFn: params.fetchFn, + maxAttempts: MAX_POLL_ATTEMPTS, + pollIntervalMs: POLL_INTERVAL_MS, + requestFailedMessage: "BytePlus video status request failed", + timeoutMessage: `BytePlus video generation task ${params.taskId} did not finish in time`, + isComplete: (payload) => readBytePlusTaskStatus(payload) === "succeeded", + getFailureMessage: (payload) => { + const status = readBytePlusTaskStatus(payload); + return status === "failed" || status === "cancelled" + ? readBytePlusErrorMessage(payload.error) || "BytePlus video generation failed" + : undefined; + }, + }); } async function downloadBytePlusVideo(params: { @@ -205,16 +188,29 @@ async function downloadBytePlusVideo(params: { fetchFn: typeof fetch; maxBytes: number; }): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "BytePlus generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const response = await fetchProviderDownloadResponse({ url: params.url, init: { method: "GET" }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + deadline, fetchFn: params.fetchFn, provider: "byteplus", requestFailedMessage: "BytePlus generated video download failed", }); const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `BytePlus generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`BytePlus generated video download exceeds ${maxBytes} bytes`), }); diff --git a/extensions/matrix/src/cli.ts b/extensions/matrix/src/cli.ts index 6a6f5a09e77..56357364703 100644 --- a/extensions/matrix/src/cli.ts +++ b/extensions/matrix/src/cli.ts @@ -3,6 +3,7 @@ import type { Command } from "commander"; import { normalizeAccountId } from "openclaw/plugin-sdk/account-id"; import { createLazyRuntimeModule } from "openclaw/plugin-sdk/lazy-runtime"; import { parseStrictInteger, timestampMsToIsoString } from "openclaw/plugin-sdk/number-runtime"; +import { readByteStreamWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import type { ChannelSetupInput } from "openclaw/plugin-sdk/setup"; import { resolveMatrixAccount, resolveMatrixAccountConfig } from "./matrix/accounts.js"; import { listMatrixOwnDevices, pruneMatrixStaleGatewayDevices } from "./matrix/actions/devices.js"; @@ -68,19 +69,11 @@ function markCliFailure(): void { } async function readMatrixCliRecoveryKeyFromStdin(): Promise { - const chunks: Buffer[] = []; - let totalBytes = 0; - for await (const chunk of process.stdin) { - const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk)); - totalBytes += buffer.byteLength; - if (totalBytes > MATRIX_CLI_RECOVERY_KEY_STDIN_MAX_BYTES) { - throw new Error( - `Matrix recovery key stdin exceeds ${MATRIX_CLI_RECOVERY_KEY_STDIN_MAX_BYTES} bytes.`, - ); - } - chunks.push(buffer); - } - const recoveryKey = Buffer.concat(chunks, totalBytes).toString("utf8").trim(); + const bytes = await readByteStreamWithLimit(process.stdin, { + maxBytes: MATRIX_CLI_RECOVERY_KEY_STDIN_MAX_BYTES, + onOverflow: ({ maxBytes }) => new Error(`Matrix recovery key stdin exceeds ${maxBytes} bytes.`), + }); + const recoveryKey = bytes.toString("utf8").trim(); if (!recoveryKey) { throw new Error("Matrix recovery key was requested from stdin, but stdin was empty."); } diff --git a/extensions/minimax/music-generation-provider.test.ts b/extensions/minimax/music-generation-provider.test.ts index 910fc1e01f1..b2790d8a8c1 100644 --- a/extensions/minimax/music-generation-provider.test.ts +++ b/extensions/minimax/music-generation-provider.test.ts @@ -68,6 +68,19 @@ function expectMinimaxGuardedFetchCall(index: number, url: string) { }; } +function expectDownloadFetchTimeout(url: string, totalTimeoutMs: number): void { + const call = fetchWithTimeoutMock.mock.calls[0]; + if (!call) { + throw new Error("expected generated music download"); + } + const [actualUrl, init, timeoutMs, fetchFn] = call; + expect(actualUrl).toBe(url); + expect(init).toEqual({ method: "GET" }); + expect(timeoutMs).toBeGreaterThan(totalTimeoutMs - 1_000); + expect(timeoutMs).toBeLessThanOrEqual(totalTimeoutMs); + expect(fetchFn).toBe(fetch); +} + function expectAllowPrivateNetworkPolicy(options: Record | undefined): void { expect(options).toEqual({ ssrfPolicy: { allowPrivateNetwork: true }, @@ -269,12 +282,7 @@ describe("minimax music generation provider", () => { lyrics: "our city wakes", }); - expect(fetchWithTimeoutMock).toHaveBeenCalledWith( - "https://example.com/url-audio.mp3", - { method: "GET" }, - 120000, - fetch, - ); + expectDownloadFetchTimeout("https://example.com/url-audio.mp3", 120_000); expect(result.tracks[0]?.buffer.byteLength).toBeGreaterThan(0); expect(result.lyrics).toEqual(["our city wakes"]); expect(result.metadata?.taskId).toBe("task-url"); @@ -326,12 +334,7 @@ describe("minimax music generation provider", () => { }); expect(mockCallArg(postJsonRequestMock).timeoutMs).toBe(600000); - expect(fetchWithTimeoutMock).toHaveBeenCalledWith( - "https://example.com/long-timeout.mp3", - { method: "GET" }, - 600000, - fetch, - ); + expectDownloadFetchTimeout("https://example.com/long-timeout.mp3", 600_000); }); it("applies explicit caller timeouts while reading streaming response bodies", async () => { diff --git a/extensions/minimax/music-generation-provider.ts b/extensions/minimax/music-generation-provider.ts index eee8710e854..70e3b440976 100644 --- a/extensions/minimax/music-generation-provider.ts +++ b/extensions/minimax/music-generation-provider.ts @@ -10,6 +10,7 @@ import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runt import { assertOkOrThrowHttpError, createProviderOperationDeadline, + createProviderOperationTimeoutResolver, executeProviderOperationWithRetry, fetchWithTimeoutGuarded, postJsonRequest, @@ -141,6 +142,14 @@ async function downloadTrackFromUrl(params: { maxBytes: number; policy: MinimaxRequestPolicy; }): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "MiniMax generated music download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const result = await executeProviderOperationWithRetry({ provider: "minimax", stage: "download", @@ -148,7 +157,7 @@ async function downloadTrackFromUrl(params: { const guardedResult = await fetchWithTimeoutGuarded( params.url, { method: "GET" }, - params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + timeoutMs(), params.fetchFn, resolveMinimaxGuardedRequestOptions(params.policy), ); @@ -170,6 +179,11 @@ async function downloadTrackFromUrl(params: { const ext = extensionForMime(mimeType)?.replace(/^\./u, "") || "mp3"; return { buffer: await readResponseWithLimit(result.response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `MiniMax generated music download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`MiniMax generated music download exceeds ${maxBytes} bytes`), }), @@ -181,12 +195,6 @@ async function downloadTrackFromUrl(params: { } } -function createMinimaxMusicTimeoutError(deadline: ProviderOperationDeadline): Error { - const timeoutLabel = - typeof deadline.timeoutMs === "number" ? ` after ${deadline.timeoutMs}ms` : ""; - return new Error(`${deadline.label} timed out${timeoutLabel}`); -} - function resolveBodyReadTimeoutMs(deadline: ProviderOperationDeadline): number { return resolveProviderOperationTimeoutMs({ deadline, @@ -198,6 +206,12 @@ function createGeneratedMusicTooLargeError(maxBytes: number): Error { return new Error(`MiniMax generated music download exceeds ${maxBytes} bytes`); } +function createMinimaxMusicTimeoutError(deadline: ProviderOperationDeadline): Error { + const timeoutLabel = + typeof deadline.timeoutMs === "number" ? ` after ${deadline.timeoutMs}ms` : ""; + return new Error(`${deadline.label} timed out${timeoutLabel}`); +} + function resolveStreamEnvelopeMaxBytes(maxBytes: number): number { return Math.max( STREAM_ENVELOPE_OVERHEAD_BYTES, @@ -210,65 +224,11 @@ async function readResponseBufferWithDeadline( deadline: ProviderOperationDeadline, maxBytes: number, ): Promise { - const body = response.body; - if (!body) { - return Buffer.alloc(0); - } - - const reader = body.getReader(); - const chunks: Uint8Array[] = []; - let totalBytes = 0; - try { - while (true) { - let timeoutId: ReturnType | undefined; - try { - const timeoutMs = resolveBodyReadTimeoutMs(deadline); - const timeoutPromise = new Promise((_, reject) => { - timeoutId = setTimeout(() => reject(createMinimaxMusicTimeoutError(deadline)), timeoutMs); - }); - const result = await Promise.race([reader.read(), timeoutPromise]); - if (result.done) { - break; - } - if (!result.value || result.value.length === 0) { - continue; - } - const nextTotalBytes = totalBytes + result.value.byteLength; - if (nextTotalBytes > maxBytes) { - const error = createGeneratedMusicTooLargeError(maxBytes); - try { - await reader.cancel(error); - } catch { - // Preserve the size-limit failure that caused cancellation. - } - throw error; - } - chunks.push(result.value); - totalBytes = nextTotalBytes; - } catch (error) { - try { - await reader.cancel(error); - } catch { - // Preserve the timeout or stream read failure that caused cancellation. - } - throw error; - } finally { - if (timeoutId) { - clearTimeout(timeoutId); - } - } - } - } finally { - reader.releaseLock(); - } - - const buffer = Buffer.allocUnsafe(totalBytes); - let offset = 0; - for (const chunk of chunks) { - buffer.set(chunk, offset); - offset += chunk.byteLength; - } - return buffer; + return await readResponseWithLimit(response, maxBytes, { + timeoutMs: () => resolveBodyReadTimeoutMs(deadline), + onTimeout: () => createMinimaxMusicTimeoutError(deadline), + onOverflow: ({ maxBytes: limit }) => createGeneratedMusicTooLargeError(limit), + }); } async function readStreamingTrack( diff --git a/extensions/minimax/provider-http.test-helpers.ts b/extensions/minimax/provider-http.test-helpers.ts index 78bb4ba0ae9..48ca377603a 100644 --- a/extensions/minimax/provider-http.test-helpers.ts +++ b/extensions/minimax/provider-http.test-helpers.ts @@ -91,6 +91,15 @@ function resolveMockProviderTimeoutMs( return typeof timeoutMs === "function" ? timeoutMs() : (timeoutMs ?? 60_000); } +function resolveMockProviderDownloadTimeoutMs(params: FetchProviderDownloadResponseParams) { + if (!params.deadline) { + return resolveMockProviderTimeoutMs(params.timeoutMs); + } + return params.deadline.deadlineAtMs === undefined + ? (params.deadline.timeoutMs ?? 60_000) + : Math.max(1, params.deadline.deadlineAtMs - Date.now()); +} + minimaxProviderHttpMocks.fetchProviderOperationResponseMock.mockImplementation( async (params: FetchProviderOperationResponseParams) => { const response = await minimaxProviderHttpMocks.fetchWithTimeoutMock( @@ -114,7 +123,7 @@ minimaxProviderHttpMocks.fetchProviderDownloadResponseMock.mockImplementation( const response = await minimaxProviderHttpMocks.fetchWithTimeoutMock( params.url, params.init ?? {}, - resolveMockProviderTimeoutMs(params.timeoutMs), + resolveMockProviderDownloadTimeoutMs(params), params.fetchFn, ); await minimaxProviderHttpMocks.assertOkOrThrowHttpErrorMock( @@ -156,15 +165,34 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importActual) => { timeoutMs, }: { label: string; - timeoutMs?: number; - }) => ({ - label, - timeoutMs, - }), + timeoutMs?: number | (() => number); + }) => { + const resolvedTimeoutMs = typeof timeoutMs === "function" ? timeoutMs() : timeoutMs; + return { + label, + timeoutMs: resolvedTimeoutMs, + deadlineAtMs: + typeof resolvedTimeoutMs === "number" ? Date.now() + resolvedTimeoutMs : undefined, + }; + }, createProviderOperationTimeoutResolver: - ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) => - () => + ({ + deadline, defaultTimeoutMs, + }: { + deadline: { deadlineAtMs?: number; label: string; timeoutMs?: number }; + defaultTimeoutMs: number; + }) => + () => { + if (typeof deadline.deadlineAtMs !== "number") { + return defaultTimeoutMs; + } + const remainingMs = deadline.deadlineAtMs - Date.now(); + if (remainingMs <= 0) { + throw new Error(`${deadline.label} timed out after ${deadline.timeoutMs}ms`); + } + return Math.min(defaultTimeoutMs, remainingMs); + }, executeProviderOperationWithRetry: minimaxProviderHttpMocks.executeProviderOperationWithRetryMock, fetchProviderDownloadResponse: minimaxProviderHttpMocks.fetchProviderDownloadResponseMock, diff --git a/extensions/minimax/video-generation-provider.ts b/extensions/minimax/video-generation-provider.ts index b84e1841b60..cc0f707e63b 100644 --- a/extensions/minimax/video-generation-provider.ts +++ b/extensions/minimax/video-generation-provider.ts @@ -248,6 +248,10 @@ async function pollMinimaxVideo(params: { timeoutMs: params.timeoutMs, label: `MiniMax video generation task ${params.taskId}`, }); + const resolveTimeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, + }); for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { const url = new URL(`${params.baseUrl}/v1/query/video_generation`); url.searchParams.set("task_id", params.taskId); @@ -258,10 +262,7 @@ async function pollMinimaxVideo(params: { method: "GET", headers: params.headers, }, - timeoutMs: createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: DEFAULT_TIMEOUT_MS, - }), + timeoutMs: resolveTimeoutMs, fetchFn: params.fetchFn, requestFailedMessage: "MiniMax video status request failed", policy: params.policy, @@ -271,6 +272,11 @@ async function pollMinimaxVideo(params: { payload = await readProviderJsonResponse( response, "MiniMax video generation failed", + { + timeoutMs: resolveTimeoutMs, + onTimeout: ({ timeoutMs }) => + new Error(`MiniMax video generation timed out after ${timeoutMs}ms`), + }, ); } finally { await release(); @@ -299,11 +305,19 @@ async function downloadVideoFromUrl(params: { maxBytes: number; policy: MinimaxRequestPolicy; }): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "MiniMax generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const { response, release } = await fetchMinimaxResponse({ stage: "download", url: params.url, init: { method: "GET" }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + timeoutMs, fetchFn: params.fetchFn, requestFailedMessage: "MiniMax generated video download failed", policy: params.policy, @@ -311,6 +325,11 @@ async function downloadVideoFromUrl(params: { try { const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `MiniMax generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`MiniMax generated video download exceeds ${maxBytes} bytes`), }); @@ -335,6 +354,14 @@ async function downloadVideoFromFileId(params: { }): Promise { const url = new URL(`${params.baseUrl}/v1/files/retrieve`); url.searchParams.set("file_id", params.fileId); + const metadataDeadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "MiniMax generated video metadata", + }); + const metadataTimeoutMs = createProviderOperationTimeoutResolver({ + deadline: metadataDeadline, + defaultTimeoutMs: metadataDeadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const { response: metadataResponse, release: releaseMetadata } = await fetchMinimaxResponse({ stage: "download", url: url.toString(), @@ -342,7 +369,7 @@ async function downloadVideoFromFileId(params: { method: "GET", headers: params.headers, }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + timeoutMs: metadataTimeoutMs, fetchFn: params.fetchFn, requestFailedMessage: "MiniMax generated video metadata request failed", policy: params.policy, @@ -352,6 +379,13 @@ async function downloadVideoFromFileId(params: { metadata = await readProviderJsonResponse( metadataResponse, "MiniMax generated video metadata", + { + timeoutMs: metadataTimeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `MiniMax generated video metadata timed out after ${metadataDeadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), + }, ); } finally { await releaseMetadata(); @@ -361,11 +395,19 @@ async function downloadVideoFromFileId(params: { if (!downloadUrl) { throw new Error("MiniMax generated video metadata missing download_url"); } + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "MiniMax generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const { response, release } = await fetchMinimaxResponse({ stage: "download", url: downloadUrl, init: { method: "GET" }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + timeoutMs, fetchFn: params.fetchFn, requestFailedMessage: "MiniMax generated video download failed", policy: params.policy, @@ -373,6 +415,11 @@ async function downloadVideoFromFileId(params: { try { const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `MiniMax generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`MiniMax generated video download exceeds ${maxBytes} bytes`), }); diff --git a/extensions/openai/video-generation-provider.ts b/extensions/openai/video-generation-provider.ts index e4e6fdfa46f..a93d42a57e1 100644 --- a/extensions/openai/video-generation-provider.ts +++ b/extensions/openai/video-generation-provider.ts @@ -183,26 +183,23 @@ async function pollOpenAIVideo( }); } -function resolveOpenAIVideoDownloadTimeoutMs(timeoutMs: ProviderOperationTimeoutMs | undefined) { - const resolved = typeof timeoutMs === "function" ? timeoutMs() : timeoutMs; - return typeof resolved === "number" && Number.isFinite(resolved) && resolved > 0 - ? resolved - : DEFAULT_TIMEOUT_MS; -} - async function fetchOpenAIVideoDownload( params: { url: string; init: RequestInit; - timeoutMs?: ProviderOperationTimeoutMs; + deadline: ReturnType; fetchFn: typeof fetch; } & OpenAIVideoRequestPolicy, ) { + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline: params.deadline, + defaultTimeoutMs: params.deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); if (!params.allowPrivateNetwork && !params.dispatcherPolicy) { const response = await fetchProviderDownloadResponse({ url: params.url, init: params.init, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + deadline: params.deadline, fetchFn: params.fetchFn, provider: "openai", requestFailedMessage: "OpenAI video download failed", @@ -220,7 +217,7 @@ async function fetchOpenAIVideoDownload( const result = await fetchWithTimeoutGuarded( params.url, params.init, - resolveOpenAIVideoDownloadTimeoutMs(params.timeoutMs), + timeoutMs(), params.fetchFn, { ...(params.allowPrivateNetwork ? { ssrfPolicy: { allowPrivateNetwork: true } } : {}), @@ -251,6 +248,14 @@ async function downloadOpenAIVideo( ): Promise { const url = new URL(`${params.baseUrl}/videos/${params.videoId}/content`); url.searchParams.set("variant", "video"); + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "OpenAI generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const { response, release } = await fetchOpenAIVideoDownload({ url: url.toString(), init: { @@ -260,7 +265,7 @@ async function downloadOpenAIVideo( Accept: "application/binary", }), }, - timeoutMs: params.timeoutMs, + deadline, fetchFn: params.fetchFn, allowPrivateNetwork: params.allowPrivateNetwork, dispatcherPolicy: params.dispatcherPolicy, @@ -268,6 +273,11 @@ async function downloadOpenAIVideo( try { const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `OpenAI generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`OpenAI generated video download exceeds ${maxBytes} bytes`), }); diff --git a/extensions/runway/video-generation-provider.ts b/extensions/runway/video-generation-provider.ts index 8b82a9111c5..5d7f1b55d60 100644 --- a/extensions/runway/video-generation-provider.ts +++ b/extensions/runway/video-generation-provider.ts @@ -7,12 +7,11 @@ import { createProviderOperationDeadline, createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, - fetchProviderOperationResponse, + pollProviderOperationJson, postJsonRequest, readProviderJsonResponse, resolveProviderOperationTimeoutMs, resolveProviderHttpRequestConfig, - waitProviderOperationPollInterval, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; @@ -279,42 +278,25 @@ async function pollRunwayTask(params: { timeoutMs: params.timeoutMs, label: `Runway video generation task ${params.taskId}`, }); - for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { - const response = await fetchProviderOperationResponse({ - stage: "poll", - url: `${params.baseUrl}/v1/tasks/${params.taskId}`, - init: { - method: "GET", - headers: params.headers, - }, - timeoutMs: createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: DEFAULT_TIMEOUT_MS, - }), - fetchFn: params.fetchFn, - provider: "runway", - requestFailedMessage: "Runway video status request failed", - }); - const payload = await readRunwayJsonResponse( - response, - "Runway video status request failed", - ); - const status = readRunwayTaskStatus(payload); - switch (status) { - case "SUCCEEDED": - return payload; - case "FAILED": - case "CANCELLED": - throw new Error( - readRunwayFailureMessage(payload.failure) || - `Runway video generation ${normalizeLowercaseStringOrEmpty(status)}`, - ); - default: - await waitProviderOperationPollInterval({ deadline, pollIntervalMs: POLL_INTERVAL_MS }); - break; - } - } - throw new Error(`Runway video generation task ${params.taskId} did not finish in time`); + return await pollProviderOperationJson({ + url: `${params.baseUrl}/v1/tasks/${params.taskId}`, + headers: params.headers, + deadline, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, + fetchFn: params.fetchFn, + maxAttempts: MAX_POLL_ATTEMPTS, + pollIntervalMs: POLL_INTERVAL_MS, + requestFailedMessage: "Runway video status request failed", + timeoutMessage: `Runway video generation task ${params.taskId} did not finish in time`, + isComplete: (payload) => readRunwayTaskStatus(payload) === "SUCCEEDED", + getFailureMessage: (payload) => { + const status = readRunwayTaskStatus(payload); + return status === "FAILED" || status === "CANCELLED" + ? readRunwayFailureMessage(payload.failure) || + `Runway video generation ${normalizeLowercaseStringOrEmpty(status)}` + : undefined; + }, + }); } async function downloadRunwayVideos(params: { @@ -325,16 +307,29 @@ async function downloadRunwayVideos(params: { }): Promise { const videos: GeneratedVideoAsset[] = []; for (const [index, url] of params.urls.entries()) { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "Runway generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const response = await fetchProviderDownloadResponse({ url, init: { method: "GET" }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + deadline, fetchFn: params.fetchFn, provider: "runway", requestFailedMessage: "Runway generated video download failed", }); const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `Runway generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`Runway generated video download exceeds ${maxBytes} bytes`), }); diff --git a/extensions/synology-chat/src/client.loopback.test.ts b/extensions/synology-chat/src/client.loopback.test.ts index 6449cb6cc4f..6004cbd86bb 100644 --- a/extensions/synology-chat/src/client.loopback.test.ts +++ b/extensions/synology-chat/src/client.loopback.test.ts @@ -8,6 +8,18 @@ const USER_LIST_RESPONSE_MAX_BYTES = 1 * 1024 * 1024; describe("Synology Chat user_list loopback", () => { let server: http.Server | undefined; + async function listenLoopback(handler: http.RequestListener): Promise { + server = http.createServer(handler); + server.on("clientError", (_err, socket) => socket.destroy()); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("expected loopback server address"); + } + return address.port; + } + afterEach(async () => { vi.restoreAllMocks(); if (server) { @@ -21,7 +33,7 @@ describe("Synology Chat user_list loopback", () => { it("aborts a streamed overflow and returns the stale cached identity", async () => { let requestCount = 0; - server = http.createServer((_req, res) => { + const port = await listenLoopback((_req, res) => { requestCount += 1; res.on("error", () => {}); res.writeHead(200, { "Content-Type": "application/json" }); @@ -37,16 +49,8 @@ describe("Synology Chat user_list loopback", () => { res.write(Buffer.alloc(USER_LIST_RESPONSE_MAX_BYTES, 0x78)); res.end(Buffer.from("x")); }); - server.on("clientError", (_err, socket) => socket.destroy()); - server.listen(0, "127.0.0.1"); - await once(server, "listening"); - - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("expected loopback server address"); - } const incomingUrl = - `http://127.0.0.1:${address.port}/webapi/entry.cgi?` + + `http://127.0.0.1:${port}/webapi/entry.cgi?` + "api=SYNO.Chat.External&method=chatbot&version=2"; const now = vi.spyOn(Date, "now"); now.mockReturnValue(1_700_000_000_000); @@ -76,7 +80,7 @@ describe("Synology Chat user_list loopback", () => { it("bounds a dripping user_list body with a wall-clock deadline", async () => { let requestCount = 0; - server = http.createServer((_req, res) => { + const port = await listenLoopback((_req, res) => { requestCount += 1; res.on("error", () => {}); res.writeHead(200, { @@ -102,16 +106,8 @@ describe("Synology Chat user_list loopback", () => { res.on("close", () => clearInterval(dripTimer)); res.write("x"); }); - server.on("clientError", (_err, socket) => socket.destroy()); - server.listen(0, "127.0.0.1"); - await once(server, "listening"); - - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("expected loopback server address"); - } const incomingUrl = - `http://127.0.0.1:${address.port}/webapi/entry.cgi?` + + `http://127.0.0.1:${port}/webapi/entry.cgi?` + "api=SYNO.Chat.External&method=chatbot&version=2"; const now = vi.spyOn(Date, "now"); now.mockReturnValue(1_700_000_100_000); @@ -152,7 +148,7 @@ describe("Synology Chat user_list loopback", () => { it("bounds a dripping chatbot response with a wall-clock deadline", async () => { let requestCount = 0; - server = http.createServer((_req, res) => { + const port = await listenLoopback((_req, res) => { requestCount += 1; res.on("error", () => {}); res.writeHead(200, { @@ -167,15 +163,7 @@ describe("Synology Chat user_list loopback", () => { res.on("close", () => clearInterval(dripTimer)); res.write("x"); }); - server.on("clientError", (_err, socket) => socket.destroy()); - server.listen(0, "127.0.0.1"); - await once(server, "listening"); - - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("expected loopback server address"); - } - const incomingUrl = `http://127.0.0.1:${address.port}/webapi/entry.cgi`; + const incomingUrl = `http://127.0.0.1:${port}/webapi/entry.cgi`; const timeoutMs = 250; const nativeSetTimeout = globalThis.setTimeout; const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); diff --git a/extensions/together/video-generation-provider.ts b/extensions/together/video-generation-provider.ts index 97a2334282d..594c569a54d 100644 --- a/extensions/together/video-generation-provider.ts +++ b/extensions/together/video-generation-provider.ts @@ -153,16 +153,29 @@ async function downloadTogetherVideo(params: { fetchFn: typeof fetch; maxBytes: number; }): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + label: "Together generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, + }); const response = await fetchProviderDownloadResponse({ url: params.url, init: { method: "GET" }, - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + deadline, fetchFn: params.fetchFn, provider: "together", requestFailedMessage: "Together generated video download failed", }); const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `Together generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`Together generated video download exceeds ${maxBytes} bytes`), }); diff --git a/extensions/vydra/shared.test.ts b/extensions/vydra/shared.test.ts index 02000c03921..035f51b6e7a 100644 --- a/extensions/vydra/shared.test.ts +++ b/extensions/vydra/shared.test.ts @@ -172,31 +172,12 @@ describe("downloadVydraAsset", () => { it("does not bound a dripping body when only chunk idle timeout is used", async () => { // Negative control: chunkTimeoutMs resets on every drip, so idle alone never fires. - server = http.createServer((_req, res) => { - res.on("error", () => {}); - res.writeHead(200, { - "Content-Type": "image/png", - "Transfer-Encoding": "chunked", - }); - const drip = () => { - if (res.writableEnded || res.destroyed) { - return; - } - res.write(Buffer.from([0x00])); - const timer = setTimeout(drip, 20); - dripTimers.add(timer); - }; - drip(); + const port = await listenDripServer({ + statusCode: 200, + contentType: "image/png", + chunk: Buffer.from([0x00]), }); - server.listen(0, "127.0.0.1"); - await once(server, "listening"); - - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("expected loopback server address"); - } - - const response = await fetch(`http://127.0.0.1:${address.port}/`); + const response = await fetch(`http://127.0.0.1:${port}/`); let settled = false; void readResponseWithLimit(response, 1024 * 1024, { chunkTimeoutMs: 100, diff --git a/extensions/vydra/shared.ts b/extensions/vydra/shared.ts index 33b87fb9bee..0c1474c5e6e 100644 --- a/extensions/vydra/shared.ts +++ b/extensions/vydra/shared.ts @@ -4,20 +4,15 @@ import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { assertOkOrThrowHttpError, - createProviderHttpError, createProviderOperationDeadline, + createProviderOperationTimeoutResolver, fetchWithTimeout, - readProviderJsonResponse, - resolveProviderOperationTimeoutMs, + pollProviderOperationJson, resolveProviderHttpRequestConfig, - waitProviderOperationPollInterval, type ProviderOperationDeadline, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { - readResponseTextPrefix, - readResponseWithLimit, -} from "openclaw/plugin-sdk/response-limit-runtime"; +import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { normalizeOptionalLowercaseString, normalizeOptionalString, @@ -38,8 +33,6 @@ type VydraAuthStore = Parameters[0]["store"]; type VydraMediaKind = "audio" | "image" | "video"; -class VydraDownloadBodyTimeoutError extends Error {} - type VydraJobPayload = { id?: string; jobId?: string; @@ -222,6 +215,12 @@ function resolveVydraHttpTimeoutMs(timeoutMs: ProviderOperationTimeoutMs | undef return resolved; } +function createVydraTimeoutError(deadline: ProviderOperationDeadline): Error { + const timeoutLabel = + typeof deadline.timeoutMs === "number" ? ` after ${deadline.timeoutMs}ms` : ""; + return new Error(`${deadline.label} timed out${timeoutLabel}`); +} + export function resolveVydraGeneratedMediaMaxBytes(params: { cfg: { agents?: { defaults?: { mediaMaxMb?: number } } }; kind: VydraMediaKind; @@ -239,25 +238,6 @@ export function resolveVydraGeneratedMediaMaxBytes(params: { return DEFAULT_GENERATED_VIDEO_MAX_BYTES; } -function resolveVydraDownloadBodyTimeout(params: { - kind: VydraMediaKind; - timeoutMs: number; - deadlineMs: number; -}) { - return { - chunkTimeoutMs: params.timeoutMs, - timeoutMs: Math.max(1, params.deadlineMs - Date.now()), - onIdleTimeout: ({ chunkTimeoutMs }: { chunkTimeoutMs: number }) => - new VydraDownloadBodyTimeoutError( - `Vydra ${params.kind} download stalled after ${chunkTimeoutMs}ms`, - ), - onTimeout: () => - new VydraDownloadBodyTimeoutError( - `Vydra ${params.kind} download timed out after ${params.timeoutMs}ms`, - ), - }; -} - export async function downloadVydraAsset(params: { url: string; kind: VydraMediaKind; @@ -266,47 +246,30 @@ export async function downloadVydraAsset(params: { maxBytes: number; }): Promise<{ buffer: Buffer; mimeType: string; fileName: string }> { const timeoutMs = resolveVydraHttpTimeoutMs(params.timeoutMs); - // fetchWithTimeout clears its abort after headers land. Keep one wall-clock deadline across - // non-2xx error-detail and successful-body reads so a slow drip cannot reset chunk idle forever. - const deadlineMs = Date.now() + timeoutMs; - const response = await fetchWithTimeout(params.url, { method: "GET" }, timeoutMs, params.fetchFn); - if (!response.ok) { - // Shared assertOkOrThrowHttpError reads error detail without a wall-clock bound; a dripping - // non-2xx body would hang before the success-path reader runs. - let errorBody: string | null = null; - try { - errorBody = - ( - await readResponseTextPrefix( - response, - 16 * 1024, - resolveVydraDownloadBodyTimeout({ kind: params.kind, timeoutMs, deadlineMs }), - ) - ).text || null; - } catch (error) { - if (error instanceof VydraDownloadBodyTimeoutError) { - throw error; - } - // The shared formatter historically ignored unreadable error bodies. Preserve the known - // HTTP status and request id even when the provider closes a failed response mid-stream. - } - // Re-run the bounded body through the shared formatter so provider error metadata and - // sensitive-text redaction remain identical to the previous assertOkOrThrowHttpError path. - throw await createProviderHttpError( - new Response(errorBody, { - headers: response.headers, - status: response.status, - statusText: response.statusText, - }), - `Vydra ${params.kind} download failed`, - { statusPrefix: "HTTP " }, - ); - } + const deadline = createProviderOperationDeadline({ + timeoutMs, + label: `Vydra ${params.kind} download`, + }); + const resolveTimeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: timeoutMs, + }); + const response = await fetchWithTimeout( + params.url, + { method: "GET" }, + resolveTimeoutMs(), + params.fetchFn, + ); + await assertOkOrThrowHttpError(response, `Vydra ${params.kind} download failed`, { + bodyTimeoutMs: resolveTimeoutMs, + onBodyTimeout: () => createVydraTimeoutError(deadline), + }); const mimeType = response.headers.get("content-type")?.trim() || (params.kind === "image" ? "image/png" : params.kind === "audio" ? "audio/mpeg" : "video/mp4"); const buffer = await readResponseWithLimit(response, params.maxBytes, { - ...resolveVydraDownloadBodyTimeout({ kind: params.kind, timeoutMs, deadlineMs }), + timeoutMs: resolveTimeoutMs, + onTimeout: () => createVydraTimeoutError(deadline), onOverflow: ({ maxBytes }) => new Error(`Vydra ${params.kind} download exceeds ${maxBytes} bytes`), }); @@ -334,28 +297,26 @@ async function waitForVydraJob(params: { timeoutMs: params.timeoutMs, label: `Vydra job ${params.jobId}`, }); - for (let attempt = 0; attempt < MAX_POLL_ATTEMPTS; attempt += 1) { - const response = await fetchWithTimeout( - `${params.baseUrl}/jobs/${params.jobId}`, - { - method: "GET", - headers: params.headers, - }, - resolveProviderOperationTimeoutMs({ deadline, defaultTimeoutMs: DEFAULT_HTTP_TIMEOUT_MS }), - params.fetchFn, - ); - await assertOkOrThrowHttpError(response, "Vydra job status request failed"); - const payload = await readProviderJsonResponse(response, "Vydra job status"); - const status = resolveVydraResponseStatus(payload); - if (status === "completed" || extractVydraResultUrls(payload, params.kind).length > 0) { - return payload; - } - if (status === "failed" || status === "error" || status === "cancelled") { - throw new Error(resolveVydraErrorMessage(payload) ?? `Vydra job ${params.jobId} failed`); - } - await waitProviderOperationPollInterval({ deadline, pollIntervalMs: POLL_INTERVAL_MS }); - } - throw new Error(`Vydra job ${params.jobId} did not finish in time`); + return await pollProviderOperationJson({ + url: `${params.baseUrl}/jobs/${params.jobId}`, + headers: params.headers, + deadline, + defaultTimeoutMs: DEFAULT_HTTP_TIMEOUT_MS, + fetchFn: params.fetchFn, + maxAttempts: MAX_POLL_ATTEMPTS, + pollIntervalMs: POLL_INTERVAL_MS, + requestFailedMessage: "Vydra job status request failed", + timeoutMessage: `Vydra job ${params.jobId} did not finish in time`, + isComplete: (payload) => + resolveVydraResponseStatus(payload) === "completed" || + extractVydraResultUrls(payload, params.kind).length > 0, + getFailureMessage: (payload) => { + const status = resolveVydraResponseStatus(payload); + return status === "failed" || status === "error" || status === "cancelled" + ? (resolveVydraErrorMessage(payload) ?? `Vydra job ${params.jobId} failed`) + : undefined; + }, + }); } export async function resolveCompletedVydraPayload(params: { diff --git a/extensions/xai/video-generation-provider.test.ts b/extensions/xai/video-generation-provider.test.ts index 1c79d356633..a328bb234d6 100644 --- a/extensions/xai/video-generation-provider.test.ts +++ b/extensions/xai/video-generation-provider.test.ts @@ -359,7 +359,8 @@ describe("xai video generation provider", () => { expect(pollRequest.url).toBe("https://api.x.ai/v1/videos/req_123"); expect(pollRequest.init?.method).toBe("GET"); expect(provider.defaultTimeoutMs).toBe(600_000); - expect(pollRequest.timeoutMs).toBe(600_000); + expect(pollRequest.timeoutMs).toBeGreaterThan(599_000); + expect(pollRequest.timeoutMs).toBeLessThanOrEqual(600_000); expect(result.videos[0]?.mimeType).toBe("video/webm"); expect(result.videos[0]?.fileName).toBe("video-1.webm"); expect(result.metadata?.requestId).toBe("req_123"); diff --git a/extensions/xai/video-generation-transport.ts b/extensions/xai/video-generation-transport.ts index 0ff08f42a46..d5065fbe957 100644 --- a/extensions/xai/video-generation-transport.ts +++ b/extensions/xai/video-generation-transport.ts @@ -1,6 +1,8 @@ import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; import { assertOkOrThrowHttpError, + createProviderOperationDeadline, + createProviderOperationTimeoutResolver, executeProviderOperationWithRetry, fetchWithTimeoutGuarded, type ProviderOperationTimeoutMs, @@ -71,13 +73,21 @@ export async function downloadXaiVideo( maxBytes: number; } & XaiVideoRequestPolicy, ): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs ?? params.defaultTimeoutMs, + label: "xAI generated video download", + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? params.defaultTimeoutMs, + }); const { response, release } = await fetchXaiVideoResponse({ url: params.url, stage: "download", requestFailedMessage: "xAI generated video download failed", auditContext: "xai-video-download", init: { method: "GET" }, - timeoutMs: params.timeoutMs, + timeoutMs, defaultTimeoutMs: params.defaultTimeoutMs, allowPrivateNetwork: params.allowPrivateNetwork, dispatcherPolicy: params.dispatcherPolicy, @@ -86,6 +96,11 @@ export async function downloadXaiVideo( try { const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4"; const buffer = await readResponseWithLimit(response, params.maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `xAI generated video download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes }) => new Error(`xAI generated video download exceeds ${maxBytes} bytes`), }); diff --git a/packages/ai/src/internal/openai.ts b/packages/ai/src/internal/openai.ts index 3461d7df5c1..7b58db5ca9c 100644 --- a/packages/ai/src/internal/openai.ts +++ b/packages/ai/src/internal/openai.ts @@ -10,6 +10,7 @@ export * from "../providers/openai-responses-stream-compat.js"; export * from "../providers/openai-responses-tool-call-tracker.js"; export * from "../providers/openai-stop-reason.js"; export * from "../providers/openai-tool-projection.js"; +export * from "../providers/openai-tool-schema-compat.js"; export * from "../providers/openai-tool-schema.js"; export * from "../providers/schema-keyword-strip.js"; export * from "../providers/tool-schema-json-projection.js"; diff --git a/packages/ai/src/providers/openai-tool-schema-compat.ts b/packages/ai/src/providers/openai-tool-schema-compat.ts new file mode 100644 index 00000000000..d8868e0128c --- /dev/null +++ b/packages/ai/src/providers/openai-tool-schema-compat.ts @@ -0,0 +1,245 @@ +import type { TSchema } from "typebox"; + +type NormalizeOpenAIStrictCompatOptions = { + promoteEmptyObject: boolean; +}; + +const OPENAI_STRICT_COMPAT_SCHEMA_MAP_KEYS = new Set([ + "$defs", + "definitions", + "dependentSchemas", + // Draft-07 dependencies mix schema values with property-name arrays. The + // recursive helpers leave scalar array entries untouched. + "dependencies", + "patternProperties", + "properties", +]); + +// Annotation-only keywords whose null values can be dropped without changing +// what the schema accepts; null constraint keywords must stay so projection +// quarantines the tool instead of widening it. +const OPENAI_NULLABLE_ANNOTATION_KEYS = new Set([ + "default", + "description", + "examples", + "format", + "title", +]); + +const OPENAI_STRICT_COMPAT_SCHEMA_NESTED_KEYS = new Set([ + "additionalItems", + "additionalProperties", + "allOf", + "anyOf", + "contains", + "contentSchema", + "else", + "if", + "items", + "not", + "oneOf", + "prefixItems", + "propertyNames", + "then", + "unevaluatedItems", + "unevaluatedProperties", +]); + +function normalizeOpenAIStrictCompatSchemaMap(schema: unknown): unknown { + if (!schema || typeof schema !== "object" || Array.isArray(schema)) { + return schema; + } + + let changed = false; + const normalized: Record = {}; + for (const [key, value] of Object.entries(schema as Record)) { + const next = normalizeOpenAIStrictCompatSchemaRecursive(value, { + promoteEmptyObject: false, + }); + normalized[key] = next; + changed ||= next !== value; + } + return changed ? normalized : schema; +} + +function normalizeOpenAIStrictCompatSchemaRecursive( + schema: unknown, + options: NormalizeOpenAIStrictCompatOptions, +): unknown { + if (Array.isArray(schema)) { + let changed = false; + const normalized = schema.map((entry) => { + const next = normalizeOpenAIStrictCompatSchemaRecursive(entry, { + promoteEmptyObject: false, + }); + changed ||= next !== entry; + return next; + }); + return changed ? normalized : schema; + } + if (!schema || typeof schema !== "object") { + return schema; + } + + const record = schema as Record; + let changed = false; + let hadNullType = false; + const normalized: Record = {}; + for (const [key, value] of Object.entries(record)) { + // Repair only null-valued entries that carry no constraint semantics. + // Null constraints stay invalid so projection quarantines the tool. + if (value === null && OPENAI_NULLABLE_ANNOTATION_KEYS.has(key)) { + changed = true; + continue; + } + if (value === null && key === "type") { + hadNullType = true; + changed = true; + continue; + } + const next = OPENAI_STRICT_COMPAT_SCHEMA_MAP_KEYS.has(key) + ? normalizeOpenAIStrictCompatSchemaMap(value) + : OPENAI_STRICT_COMPAT_SCHEMA_NESTED_KEYS.has(key) + ? normalizeOpenAIStrictCompatSchemaRecursive(value, { + promoteEmptyObject: false, + }) + : value; + normalized[key] = next; + changed ||= next !== value; + } + + if (Object.keys(normalized).length === 0) { + if (!options.promoteEmptyObject) { + return schema; + } + return { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + }; + } + + const hasObjectShapeHints = + (normalized.properties && + typeof normalized.properties === "object" && + !Array.isArray(normalized.properties)) || + Array.isArray(normalized.required); + const hasArrayShapeHints = "items" in normalized; + if (!("type" in normalized) && hasObjectShapeHints !== hasArrayShapeHints) { + normalized.type = hasObjectShapeHints ? "object" : "array"; + changed = true; + } else if (hadNullType && !("type" in normalized)) { + // Without an unambiguous shape, retain the invalid type so projection + // rejects the tool instead of widening it to an unconstrained schema. + normalized.type = null; + } + if (normalized.type === "object" && !("properties" in normalized)) { + normalized.properties = {}; + changed = true; + } + + const hasEmptyProperties = + normalized.properties && + typeof normalized.properties === "object" && + !Array.isArray(normalized.properties) && + Object.keys(normalized.properties as Record).length === 0; + + if (normalized.type === "object" && !Array.isArray(normalized.required) && hasEmptyProperties) { + normalized.required = []; + changed = true; + } + if ( + normalized.type === "object" && + hasEmptyProperties && + !("additionalProperties" in normalized) + ) { + normalized.additionalProperties = false; + changed = true; + } + + return changed ? normalized : schema; +} + +/** Repairs recoverable OpenAI tool-schema shapes before canonical normalization. */ +export function normalizeOpenAIStrictCompatSchema(schema: unknown): TSchema { + return normalizeOpenAIStrictCompatSchemaRecursive(schema, { + promoteEmptyObject: true, + }) as TSchema; +} + +/** Finds schema paths that violate OpenAI strict tool-schema requirements. */ +export function findOpenAIStrictSchemaViolations( + schema: unknown, + path: string, + options?: { requireObjectRoot?: boolean }, +): string[] { + if (Array.isArray(schema)) { + if (options?.requireObjectRoot) { + return [`${path}.type`]; + } + return schema.flatMap((item, index) => + findOpenAIStrictSchemaViolations(item, `${path}[${index}]`), + ); + } + if (!schema || typeof schema !== "object") { + return options?.requireObjectRoot ? [`${path}.type`] : []; + } + + const record = schema as Record; + const violations: string[] = []; + for (const key of ["anyOf", "oneOf", "allOf"] as const) { + if (key in record) { + violations.push(`${path}.${key}`); + } + } + if (Array.isArray(record.type)) { + violations.push(`${path}.type`); + } + + const properties = + record.properties && typeof record.properties === "object" && !Array.isArray(record.properties) + ? (record.properties as Record) + : undefined; + + if (record.type === "object") { + if (record.additionalProperties !== false) { + violations.push(`${path}.additionalProperties`); + } + const required = Array.isArray(record.required) + ? record.required.filter((entry): entry is string => typeof entry === "string") + : undefined; + if (!required) { + violations.push(`${path}.required`); + } else if (properties) { + const requiredSet = new Set(required); + for (const key of Object.keys(properties)) { + if (!requiredSet.has(key)) { + violations.push(`${path}.required.${key}`); + } + } + } + } + + // Schema maps contain user-chosen names. Walk their values as schemas, but + // never interpret map keys such as `$defs.anyOf` as schema keywords. + for (const key of OPENAI_STRICT_COMPAT_SCHEMA_MAP_KEYS) { + const schemaMap = record[key]; + if (!schemaMap || typeof schemaMap !== "object" || Array.isArray(schemaMap)) { + continue; + } + for (const [entryKey, value] of Object.entries(schemaMap as Record)) { + violations.push(...findOpenAIStrictSchemaViolations(value, `${path}.${key}.${entryKey}`)); + } + } + // Only recurse through JSON Schema applicators. Annotation payloads such as + // examples/default may contain arbitrary objects that are not schemas. + for (const key of OPENAI_STRICT_COMPAT_SCHEMA_NESTED_KEYS) { + const value = record[key]; + if (value && typeof value === "object") { + violations.push(...findOpenAIStrictSchemaViolations(value, `${path}.${key}`)); + } + } + + return violations; +} diff --git a/packages/ai/src/providers/openai-tool-schema.test.ts b/packages/ai/src/providers/openai-tool-schema.test.ts index fa0e32a53bc..305bef130bb 100644 --- a/packages/ai/src/providers/openai-tool-schema.test.ts +++ b/packages/ai/src/providers/openai-tool-schema.test.ts @@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it } from "vitest"; import { projectOpenAITools } from "./openai-tool-projection.js"; import { clearOpenAIToolSchemaCacheForTest, + findOpenAIStrictSchemaViolations, findOpenAIStrictToolProjectionDiagnostics, isStrictOpenAIJsonSchemaCompatible, normalizeOpenAIStrictToolParameters, @@ -63,6 +64,54 @@ describe("OpenAI strict tool schema normalization", () => { ).toBe(false); }); + it("walks named schema maps without treating definition names as keywords", () => { + const schema = { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + $defs: { + anyOf: { type: "string" }, + }, + examples: [{ anyOf: [{ type: "string" }] }], + }; + + expect( + findOpenAIStrictSchemaViolations(schema, "parameters", { requireObjectRoot: true }), + ).toEqual([]); + }); + + it("walks legacy and content schema applicators", () => { + const nestedObject = { + type: "object", + properties: { value: { type: "string" } }, + }; + const schema = { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + dependencies: { + mode: ["payload"], + payload: nestedObject, + }, + additionalItems: nestedObject, + contentSchema: nestedObject, + }; + + expect( + findOpenAIStrictSchemaViolations(schema, "parameters", { requireObjectRoot: true }), + ).toEqual([ + "parameters.dependencies.payload.additionalProperties", + "parameters.dependencies.payload.required", + "parameters.additionalItems.additionalProperties", + "parameters.additionalItems.required", + "parameters.contentSchema.additionalProperties", + "parameters.contentSchema.required", + ]); + expect(isStrictOpenAIJsonSchemaCompatible(schema)).toBe(false); + }); + it("normalizes truly empty MCP tool schema {} for strict mode", () => { const schema = {}; const normalized = normalizeStrictOpenAIJsonSchema(schema) as Record; diff --git a/packages/ai/src/providers/openai-tool-schema.ts b/packages/ai/src/providers/openai-tool-schema.ts index e9679186d54..8ce3fa8d00d 100644 --- a/packages/ai/src/providers/openai-tool-schema.ts +++ b/packages/ai/src/providers/openai-tool-schema.ts @@ -9,6 +9,9 @@ import { type ToolSchemaModelCompat, } from "./agent-tools-parameter-schema.js"; import type { OpenAIToolProjection } from "./openai-tool-projection.js"; +import { findOpenAIStrictSchemaViolations } from "./openai-tool-schema-compat.js"; + +export { findOpenAIStrictSchemaViolations } from "./openai-tool-schema-compat.js"; /** * OpenAI strict-tool-schema normalization and diagnostics. @@ -188,7 +191,7 @@ export function findOpenAIStrictToolProjectionDiagnostics( violations: [...diagnostic.violations], })), ...projection.tools.flatMap((tool) => { - const violations = findStrictOpenAIJsonSchemaViolations( + const violations = findOpenAIStrictSchemaViolations( normalizeStrictOpenAIJsonSchema(tool.parameters), `${tool.name}.parameters`, ); @@ -246,72 +249,6 @@ function isStrictOpenAIJsonSchemaCompatibleRecursive(schema: unknown): boolean { }); } -function findStrictOpenAIJsonSchemaViolations(schema: unknown, path: string): string[] { - if (Array.isArray(schema)) { - return schema.flatMap((entry, index) => - findStrictOpenAIJsonSchemaViolations(entry, `${path}[${index}]`), - ); - } - if (!schema || typeof schema !== "object") { - return []; - } - - const record = schema as Record; - const violations: string[] = []; - for (const key of ["anyOf", "oneOf", "allOf"] as const) { - if (key in record) { - violations.push(`${path}.${key}`); - } - } - if (Array.isArray(record.type)) { - violations.push(`${path}.type`); - } - if (record.type === "object") { - if (record.additionalProperties !== false) { - violations.push(`${path}.additionalProperties`); - } - const properties = - record.properties && - typeof record.properties === "object" && - !Array.isArray(record.properties) - ? (record.properties as Record) - : {}; - const required = Array.isArray(record.required) - ? record.required.filter((entry): entry is string => typeof entry === "string") - : undefined; - if (!required) { - violations.push(`${path}.required`); - } else { - const requiredSet = new Set(required); - for (const key of Object.keys(properties)) { - if (!requiredSet.has(key)) { - violations.push(`${path}.required.${key}`); - } - } - } - } - - if ( - record.properties && - typeof record.properties === "object" && - !Array.isArray(record.properties) - ) { - for (const [key, value] of Object.entries(record.properties)) { - violations.push(...findStrictOpenAIJsonSchemaViolations(value, `${path}.properties.${key}`)); - } - } - for (const [key, value] of Object.entries(record)) { - if (key === "properties") { - continue; - } - if (value && typeof value === "object") { - violations.push(...findStrictOpenAIJsonSchemaViolations(value, `${path}.${key}`)); - } - } - - return violations; -} - /** Resolves strict mode for the projected tools that will be emitted in the request payload. */ export function resolveOpenAIProjectedToolsStrictToolFlag( projection: OpenAIToolProjection, diff --git a/packages/media-core/src/read-byte-stream-with-limit.test.ts b/packages/media-core/src/read-byte-stream-with-limit.test.ts index 464f827a1ca..c9b27b9c359 100644 --- a/packages/media-core/src/read-byte-stream-with-limit.test.ts +++ b/packages/media-core/src/read-byte-stream-with-limit.test.ts @@ -15,6 +15,8 @@ describe("readByteStreamWithLimit", () => { it("throws and destroys node streams after overflow", async () => { const stream = Readable.from([Buffer.alloc(4), Buffer.alloc(4)]); const destroySpy = vi.spyOn(stream, "destroy"); + const errorSpy = vi.fn(); + stream.on("error", errorSpy); await expect( readByteStreamWithLimit(stream, { @@ -22,6 +24,7 @@ describe("readByteStreamWithLimit", () => { onOverflow: ({ size, maxBytes }) => new Error(`too large ${size}/${maxBytes}`), }), ).rejects.toThrow("too large 8/7"); - expect(destroySpy).toHaveBeenCalled(); + expect(destroySpy).toHaveBeenCalledWith(); + expect(errorSpy).not.toHaveBeenCalled(); }); }); diff --git a/packages/media-core/src/read-byte-stream-with-limit.ts b/packages/media-core/src/read-byte-stream-with-limit.ts index 16d683e2771..97ac5621ad0 100644 --- a/packages/media-core/src/read-byte-stream-with-limit.ts +++ b/packages/media-core/src/read-byte-stream-with-limit.ts @@ -28,20 +28,23 @@ function normalizeByteChunk(chunk: unknown): Buffer { function destroyReadableOnOverflow(stream: unknown, err: Error): void { const readable = stream as { - destroy?: (error?: Error) => unknown; + destroy?: () => unknown; cancel?: (reason?: unknown) => unknown; }; // Stop upstream producers immediately after overflow; otherwise large media // streams can continue buffering after the caller has already failed. if (typeof readable.destroy === "function") { try { - readable.destroy(err); + // The helper already throws the overflow error to its caller. Passing the + // same error to destroy() also emits it on Node streams, which can become + // an unrelated uncaught exception after the awaited rejection settles. + readable.destroy(); } catch {} return; } if (typeof readable.cancel === "function") { try { - void readable.cancel(err); + void Promise.resolve(readable.cancel(err)).catch(() => undefined); } catch {} } } diff --git a/scripts/check-control-ui-performance.mjs b/scripts/check-control-ui-performance.mjs index 1d3819eb96c..2bc5c605b5c 100644 --- a/scripts/check-control-ui-performance.mjs +++ b/scripts/check-control-ui-performance.mjs @@ -156,7 +156,11 @@ function formatViolation(violation) { violation.unit === "bytes" ? formatControlUiPerformanceBytes(violation.limit) : String(violation.limit); - return `${violation.metric}: ${actual} exceeds ${limit}`; + const exactBytes = + violation.unit === "bytes" && actual === limit + ? ` (${violation.actual} B vs ${violation.limit} B)` + : ""; + return `${violation.metric}: ${actual} exceeds ${limit}${exactBytes}`; } export function formatControlUiPerformanceReport( diff --git a/src/agents/provider-http-errors.test.ts b/src/agents/provider-http-errors.test.ts index 6cb79f45680..276c69084d4 100644 --- a/src/agents/provider-http-errors.test.ts +++ b/src/agents/provider-http-errors.test.ts @@ -176,6 +176,56 @@ describe("provider error utils", () => { } satisfies Partial); }); + it("propagates a bounded error-body timeout instead of hanging normalization", async () => { + vi.useFakeTimers(); + try { + const cancel = vi.fn(); + const response = new Response( + new ReadableStream({ + pull() { + return new Promise(() => {}); + }, + cancel, + }), + { status: 503 }, + ); + const assertion = expect( + assertOkOrThrowHttpError(response, "Provider API error", { + bodyTimeoutMs: () => 50, + onBodyTimeout: ({ timeoutMs }) => new Error(`provider body timed out ${timeoutMs}`), + }), + ).rejects.toThrow("provider body timed out 50"); + + await vi.advanceTimersByTimeAsync(50); + await assertion; + expect(cancel).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it("propagates an already-expired lazy error-body deadline", async () => { + const cancel = vi.fn(); + const response = new Response( + new ReadableStream({ + pull() { + return new Promise(() => {}); + }, + cancel, + }), + { status: 503 }, + ); + + await expect( + assertOkOrThrowHttpError(response, "Provider API error", { + bodyTimeoutMs: () => { + throw new Error("provider deadline already expired"); + }, + }), + ).rejects.toThrow("provider deadline already expired"); + expect(cancel).toHaveBeenCalledTimes(1); + }); + it("releases provider error body reader locks after bounded reads complete", async () => { const releaseLock = vi.fn(); const cancel = vi.fn(async () => undefined); diff --git a/src/agents/provider-http-errors.ts b/src/agents/provider-http-errors.ts index 0fa813d519d..16a35424430 100644 --- a/src/agents/provider-http-errors.ts +++ b/src/agents/provider-http-errors.ts @@ -7,7 +7,11 @@ import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; export { asFiniteNumber } from "../../packages/normalization-core/src/number-coercion.js"; import { normalizeOptionalString as trimToUndefined } from "../../packages/normalization-core/src/string-coerce.js"; -import { readResponseTextPrefix, readResponseWithLimit } from "../infra/http-body.js"; +import { + readResponseTextPrefix, + readResponseWithLimit, + type ReadResponseTextPrefixOptions, +} from "../infra/http-body.js"; import { redactSensitiveText } from "../logging/redact.js"; export { asBoolean } from "../utils/boolean.js"; export { normalizeOptionalString as trimToUndefined } from "../../packages/normalization-core/src/string-coerce.js"; @@ -17,6 +21,30 @@ const PROVIDER_BINARY_RESPONSE_MAX_BYTES = 16 * 1024 * 1024; const PROVIDER_JSON_RESPONSE_MAX_BYTES = 16 * 1024 * 1024; const PROVIDER_TEXT_RESPONSE_MAX_BYTES = 16 * 1024 * 1024; +/** Shared timeout and byte-limit options for provider response consumption. */ +type ProviderResponseReadOptions = ReadResponseTextPrefixOptions & { + maxBytes?: number; +}; + +/** Options for bounded provider error-body normalization. */ +type ProviderHttpErrorOptions = { + statusPrefix?: string; + bodyTimeoutMs?: ReadResponseTextPrefixOptions["timeoutMs"]; + onBodyTimeout?: NonNullable; +}; + +class ProviderErrorBodyTimeout extends Error { + readonly timeoutError: unknown; + + constructor(timeoutError: unknown) { + super(timeoutError instanceof Error ? timeoutError.message : String(timeoutError), { + cause: timeoutError, + }); + this.name = "ProviderErrorBodyTimeout"; + this.timeoutError = timeoutError; + } +} + /** Returns a plain object view for provider JSON payloads when one exists. */ export function asObject(value: unknown): Record | undefined { return typeof value === "object" && value !== null && !Array.isArray(value) @@ -38,21 +66,23 @@ function redactProviderErrorBody(body: string): string { export async function readResponseTextLimited( response: Response, limitBytes = 16 * 1024, + options?: ReadResponseTextPrefixOptions, ): Promise { if (limitBytes <= 0) { return ""; } - return (await readResponseTextPrefix(response, limitBytes)).text; + return (await readResponseTextPrefix(response, limitBytes, options)).text; } /** Reads a successful provider text response under a byte cap. */ export async function readProviderTextResponse( response: Response, label: string, - opts?: { maxBytes?: number }, + opts?: ProviderResponseReadOptions, ): Promise { const maxBytes = opts?.maxBytes ?? PROVIDER_TEXT_RESPONSE_MAX_BYTES; const bytes = await readResponseWithLimit(response, maxBytes, { + ...opts, onOverflow: ({ maxBytes: maxBytesLocal }) => new Error(`${label}: text response exceeds ${maxBytesLocal} bytes`), }); @@ -131,8 +161,35 @@ type ProviderHttpErrorInfo = { }; /** Extracts normalized provider error metadata while keeping the raw body bounded and redacted. */ -async function extractProviderErrorInfo(response: Response): Promise { - const rawBody = trimToUndefined(await readResponseTextLimited(response).catch(() => "")); +async function extractProviderErrorInfo( + response: Response, + options?: ProviderHttpErrorOptions, +): Promise { + const bodyTimeoutMs = options?.bodyTimeoutMs; + const rawBody = trimToUndefined( + await readResponseTextLimited(response, 16 * 1024, { + timeoutMs: + typeof bodyTimeoutMs === "function" + ? () => { + try { + return bodyTimeoutMs(); + } catch (error) { + throw new ProviderErrorBodyTimeout(error); + } + } + : bodyTimeoutMs, + onTimeout: (params) => + new ProviderErrorBodyTimeout( + options?.onBodyTimeout?.(params) ?? + new Error(`Provider error body timed out after ${params.timeoutMs}ms`), + ), + }).catch((error: unknown) => { + if (error instanceof ProviderErrorBodyTimeout) { + throw error.timeoutError; + } + return ""; + }), + ); const requestId = extractProviderRequestId(response); if (!rawBody) { return requestId ? { requestId } : {}; @@ -221,9 +278,9 @@ export function formatProviderHttpErrorMessage(params: { export async function createProviderHttpError( response: Response, label: string, - options?: { statusPrefix?: string }, + options?: ProviderHttpErrorOptions, ): Promise { - const info = await extractProviderErrorInfo(response); + const info = await extractProviderErrorInfo(response, options); return new ProviderHttpError( formatProviderHttpErrorMessage({ label, @@ -246,19 +303,24 @@ export async function createProviderHttpError( export async function assertOkOrThrowProviderError( response: Response, label: string, + options?: Omit, ): Promise { if (response.ok) { return; } - throw await createProviderHttpError(response, label); + throw await createProviderHttpError(response, label, options); } /** Throws a normalized generic HTTP error when a fetch response is not OK. */ -export async function assertOkOrThrowHttpError(response: Response, label: string): Promise { +export async function assertOkOrThrowHttpError( + response: Response, + label: string, + options?: Omit, +): Promise { if (response.ok) { return; } - throw await createProviderHttpError(response, label, { statusPrefix: "HTTP " }); + throw await createProviderHttpError(response, label, { ...options, statusPrefix: "HTTP " }); } /** @@ -270,10 +332,11 @@ export async function assertOkOrThrowHttpError(response: Response, label: string export async function readProviderJsonResponse( response: Response, label: string, - opts?: { maxBytes?: number }, + opts?: ProviderResponseReadOptions, ): Promise { const maxBytes = opts?.maxBytes ?? PROVIDER_JSON_RESPONSE_MAX_BYTES; const bytes = await readResponseWithLimit(response, maxBytes, { + ...opts, onOverflow: ({ maxBytes: maxBytesLocal }) => new Error(`${label}: JSON response exceeds ${maxBytesLocal} bytes`), }); @@ -288,8 +351,9 @@ export async function readProviderJsonResponse( export async function readProviderJsonObjectResponse( response: Response, label: string, + opts?: ProviderResponseReadOptions, ): Promise> { - const payload = await readProviderJsonResponse(response, label); + const payload = await readProviderJsonResponse(response, label, opts); const object = asObject(payload); if (!object) { throw new Error(`${label}: malformed JSON response`); @@ -302,8 +366,9 @@ export async function readProviderJsonArrayFieldResponse( response: Response, label: string, field: string, + opts?: ProviderResponseReadOptions, ): Promise { - const payload = await readProviderJsonObjectResponse(response, label); + const payload = await readProviderJsonObjectResponse(response, label, opts); const value = payload[field]; if (!Array.isArray(value)) { throw new Error(`${label}: malformed JSON response`); @@ -340,13 +405,12 @@ export async function readProviderBinaryResponse( response: Response, label: string, kind = "binary", - opts?: { - maxBytes?: number; - }, + opts?: ProviderResponseReadOptions, ): Promise { assertProviderBinaryResponseContent(response, label, kind); const maxBytes = opts?.maxBytes ?? PROVIDER_BINARY_RESPONSE_MAX_BYTES; const bytes = await readResponseWithLimit(response, maxBytes, { + ...opts, onOverflow: ({ maxBytes: maxBytesLocal }) => new Error(`${label}: ${kind} response exceeds ${maxBytesLocal} bytes`), }); diff --git a/src/cli/config-cli.ts b/src/cli/config-cli.ts index fda6841cc3b..4f45d0f2a7c 100644 --- a/src/cli/config-cli.ts +++ b/src/cli/config-cli.ts @@ -1,5 +1,6 @@ // Config CLI command implementation for get/set/unset/patch/validate and secret refs. import fs from "node:fs"; +import { readByteStreamWithLimit } from "@openclaw/media-core/read-byte-stream-with-limit"; import { expectDefined } from "@openclaw/normalization-core"; import { isRecord as isPlainRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; @@ -1478,25 +1479,20 @@ function configPatchModeError(message: string): Error { } async function readStdinText(): Promise { - const chunks: string[] = []; - let bytes = 0; if (process.stdin.isTTY) { throw configPatchModeError( "--stdin refuses to read from an interactive terminal; pipe input or use --file .", ); } process.stdin.setEncoding("utf8"); - for await (const chunk of process.stdin) { - const text = String(chunk); - bytes += Buffer.byteLength(text, "utf8"); - if (bytes > CONFIG_PATCH_STDIN_MAX_BYTES) { - throw configPatchModeError( - `--stdin input exceeds ${CONFIG_PATCH_STDIN_MAX_BYTES} bytes; use --file for larger patches.`, - ); - } - chunks.push(text); - } - return chunks.join(""); + const bytes = await readByteStreamWithLimit(process.stdin, { + maxBytes: CONFIG_PATCH_STDIN_MAX_BYTES, + onOverflow: ({ maxBytes }) => + configPatchModeError( + `--stdin input exceeds ${maxBytes} bytes; use --file for larger patches.`, + ), + }); + return bytes.toString("utf8"); } async function readConfigPatchInput(opts: ConfigPatchOptions): Promise { diff --git a/src/cli/exec-approvals-cli.ts b/src/cli/exec-approvals-cli.ts index b6586dfd380..2f4c72f17d3 100644 --- a/src/cli/exec-approvals-cli.ts +++ b/src/cli/exec-approvals-cli.ts @@ -1,5 +1,6 @@ // CLI for reading and mutating exec approval allowlists locally, via gateway, or via node. import fs from "node:fs/promises"; +import { readByteStreamWithLimit } from "@openclaw/media-core/read-byte-stream-with-limit"; import { expectDefined } from "@openclaw/normalization-core"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; @@ -91,17 +92,11 @@ async function readStdin( stream: NodeJS.ReadableStream = process.stdin, maxBytes = EXEC_APPROVALS_STDIN_MAX_BYTES, ): Promise { - const chunks: Buffer[] = []; - let total = 0; - for await (const chunk of stream) { - const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - total += buffer.byteLength; - if (total > maxBytes) { - throw new Error(`Exec approvals stdin exceeds ${maxBytes} bytes.`); - } - chunks.push(buffer); - } - return Buffer.concat(chunks, total).toString("utf8"); + const bytes = await readByteStreamWithLimit(stream, { + maxBytes, + onOverflow: ({ maxBytes: limit }) => new Error(`Exec approvals stdin exceeds ${limit} bytes.`), + }); + return bytes.toString("utf8"); } async function resolveTargetNodeId(opts: ExecApprovalsCliOpts): Promise { diff --git a/src/commands/models/auth.ts b/src/commands/models/auth.ts index 5e66f8d61b9..5f9c10558c3 100644 --- a/src/commands/models/auth.ts +++ b/src/commands/models/auth.ts @@ -7,6 +7,7 @@ import { select as clackSelect, text as clackText, } from "@clack/prompts"; +import { readByteStreamWithLimit } from "@openclaw/media-core/read-byte-stream-with-limit"; import { expectDefined } from "@openclaw/normalization-core"; import { resolveExpiresAtMsFromDurationMs } from "@openclaw/normalization-core/number-coercion"; import { @@ -133,17 +134,11 @@ const select = async (params: Parameters>[0]) => const MODELS_AUTH_STDIN_MAX_BYTES = 1024 * 1024; async function readPipedStdin(): Promise { - const chunks: Buffer[] = []; - let totalBytes = 0; - for await (const chunk of process.stdin) { - const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk)); - totalBytes += buffer.byteLength; - if (totalBytes > MODELS_AUTH_STDIN_MAX_BYTES) { - throw new Error(`Piped auth input exceeds ${MODELS_AUTH_STDIN_MAX_BYTES} bytes.`); - } - chunks.push(buffer); - } - return Buffer.concat(chunks, totalBytes).toString("utf8"); + const bytes = await readByteStreamWithLimit(process.stdin, { + maxBytes: MODELS_AUTH_STDIN_MAX_BYTES, + onOverflow: ({ maxBytes }) => new Error(`Piped auth input exceeds ${maxBytes} bytes.`), + }); + return bytes.toString("utf8"); } async function readPastedSecret(params: { diff --git a/src/infra/http-body.response.test.ts b/src/infra/http-body.response.test.ts index c3a9ffe76dd..c0b154d4c6b 100644 --- a/src/infra/http-body.response.test.ts +++ b/src/infra/http-body.response.test.ts @@ -265,6 +265,49 @@ describe("readResponseWithLimit", () => { } }); + it("resolves a lazy overall timeout immediately before reading", async () => { + vi.useFakeTimers(); + try { + const cancel = vi.fn(); + const timeoutMs = vi.fn(() => 75); + const response = new Response(makeTricklingStream(40, cancel)); + const assertion = expect( + readResponseWithLimit(response, 1024, { + timeoutMs, + onTimeout: ({ timeoutMs: resolved }) => new Error(`lazy overall ${resolved}`), + }), + ).rejects.toThrow("lazy overall 75"); + + expect(timeoutMs).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(80); + await assertion; + expect(cancel).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it("cancels the body when a lazy timeout resolver reports an expired deadline", async () => { + const cancel = vi.fn(); + const response = new Response( + new ReadableStream({ + pull() { + return new Promise(() => {}); + }, + cancel, + }), + ); + + await expect( + readResponseWithLimit(response, 1024, { + timeoutMs: () => { + throw new Error("deadline expired"); + }, + }), + ).rejects.toThrow("deadline expired"); + expect(cancel).toHaveBeenCalledWith(expect.objectContaining({ message: "deadline expired" })); + }); + it("cancels a getReader-less body when its overall timeout expires", async () => { vi.useFakeTimers(); try { diff --git a/src/infra/http-body.ts b/src/infra/http-body.ts index caf05dc6eed..6c11ba20502 100644 --- a/src/infra/http-body.ts +++ b/src/infra/http-body.ts @@ -140,7 +140,8 @@ type ReadResponsePrefixResult = { export type ReadResponseTextPrefixOptions = { chunkTimeoutMs?: number; onIdleTimeout?: (params: { chunkTimeoutMs: number }) => Error; - timeoutMs?: number; + /** Static timeout or lazy resolver evaluated immediately before body consumption. */ + timeoutMs?: number | (() => number); onTimeout?: (params: { timeoutMs: number }) => Error; }; @@ -215,10 +216,17 @@ async function readResponsePrefix( options?: ReadResponsePrefixOptions, ): Promise { validateMaxBytes(maxBytes); + let timeoutMs: number | undefined; + try { + timeoutMs = typeof options?.timeoutMs === "function" ? options.timeoutMs() : options?.timeoutMs; + } catch (error) { + await response.body?.cancel(error).catch(() => undefined); + throw error; + } const body = response.body; if (!body || typeof body.getReader !== "function") { return await withResponseBodyTimeout({ - timeoutMs: options?.timeoutMs, + timeoutMs, onTimeout: options?.onTimeout, cancel: async (error) => await body?.cancel(error), read: async () => { @@ -237,7 +245,7 @@ async function readResponsePrefix( const reader = body.getReader(); return await withResponseBodyTimeout({ - timeoutMs: options?.timeoutMs, + timeoutMs, onTimeout: options?.onTimeout, cancel: async (error) => await reader.cancel(error), read: async () => await readResponsePrefixFromReader(reader, maxBytes, options), diff --git a/src/media-understanding/shared.test.ts b/src/media-understanding/shared.test.ts index bd7d1839d3e..7a932912dc7 100644 --- a/src/media-understanding/shared.test.ts +++ b/src/media-understanding/shared.test.ts @@ -34,7 +34,6 @@ vi.mock("../infra/net/proxy-env.js", async () => { import { createProviderOperationDeadline, - createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, fetchWithTimeoutGuarded, pollProviderOperationJson, @@ -69,6 +68,21 @@ function getFirstGuardedFetchCall() { return request as Record; } +function createTricklingResponse(status = 200): Response { + const chunk = new TextEncoder().encode("{"); + return new Response( + new ReadableStream({ + async pull(controller) { + controller.enqueue(chunk); + await new Promise((resolve) => { + setTimeout(resolve, 20); + }); + }, + }), + { status, headers: { "content-type": "application/json" } }, + ); +} + describe("provider operation deadlines", () => { it("keeps default per-call timeouts when no operation timeout is configured", () => { const deadline = createProviderOperationDeadline({ @@ -126,6 +140,22 @@ describe("provider operation deadlines", () => { expect(resolveProviderOperationTimeoutMs({ deadline, defaultTimeoutMs: 60_000 })).toBe(1_750); }); + it("resolves a lazy total timeout once when the deadline starts", () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const resolveTimeoutMs = vi.fn(() => 5_000); + + const deadline = createProviderOperationDeadline({ + label: "generated media download", + timeoutMs: resolveTimeoutMs, + }); + vi.setSystemTime(4_250); + + expect(resolveTimeoutMs).toHaveBeenCalledTimes(1); + expect(resolveProviderOperationTimeoutMs({ deadline, defaultTimeoutMs: 60_000 })).toBe(1_750); + expect(resolveTimeoutMs).toHaveBeenCalledTimes(1); + }); + it("throws once the operation deadline has expired", () => { vi.useFakeTimers(); vi.setSystemTime(1_000); @@ -211,6 +241,34 @@ describe("provider operation deadlines", () => { expect(fetchFn).toHaveBeenCalledTimes(2); }); + it("bounds a trickling poll response body with the operation deadline", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const fetchFn = vi.fn().mockResolvedValue(createTricklingResponse()); + const result = pollProviderOperationJson<{ status?: string }>({ + url: "https://api.example.com/v1/videos/task-1", + headers: new Headers(), + deadline: createProviderOperationDeadline({ + label: "video generation task task-1", + timeoutMs: 100, + }), + defaultTimeoutMs: 5_000, + fetchFn, + maxAttempts: 3, + pollIntervalMs: 1_000, + requestFailedMessage: "status failed", + timeoutMessage: "task timed out", + isComplete: (payload) => payload.status === "completed", + }); + const assertion = expect(result).rejects.toThrow( + "video generation task task-1 timed out after 100ms", + ); + + await vi.advanceTimersByTimeAsync(110); + await assertion; + expect(fetchFn).toHaveBeenCalledTimes(1); + }); + it("passes guarded request policy through provider status polling", async () => { const release = vi.fn(async () => {}); fetchWithSsrFGuardMock.mockResolvedValueOnce({ @@ -432,6 +490,7 @@ describe("provider operation deadlines", () => { const response = await fetchProviderDownloadResponse({ url: "https://cdn.example.com/video.mp4", init: { method: "GET" }, + // Keep the shipped timeoutMs call shape working for external plugins. timeoutMs: 5_000, fetchFn, provider: "test-video", @@ -444,6 +503,26 @@ describe("provider operation deadlines", () => { expect(sleep).toHaveBeenCalledWith(0, undefined); }); + it("bounds a trickling generated-asset error body", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const fetchFn = vi.fn().mockResolvedValue(createTricklingResponse(503)); + const result = fetchProviderDownloadResponse({ + url: "https://cdn.example.com/video.mp4", + init: { method: "GET" }, + deadline: createProviderOperationDeadline({ label: "download failed", timeoutMs: 100 }), + fetchFn, + provider: "test-video", + requestFailedMessage: "download failed", + retry: { attempts: 1 }, + }); + const assertion = expect(result).rejects.toThrow("download failed timed out after 100ms"); + + await vi.advanceTimersByTimeAsync(110); + await assertion; + expect(fetchFn).toHaveBeenCalledTimes(1); + }); + it("recomputes remaining download timeout before retry attempts", async () => { vi.useFakeTimers(); vi.setSystemTime(1_000); @@ -461,7 +540,7 @@ describe("provider operation deadlines", () => { fetchProviderDownloadResponse({ url: "https://cdn.example.com/video.mp4", init: { method: "GET" }, - timeoutMs: createProviderOperationTimeoutResolver({ deadline, defaultTimeoutMs: 5_000 }), + deadline, fetchFn, provider: "test-video", requestFailedMessage: "download failed", diff --git a/src/media-understanding/shared.ts b/src/media-understanding/shared.ts index c260cdc62f2..a4d3ed07893 100644 --- a/src/media-understanding/shared.ts +++ b/src/media-understanding/shared.ts @@ -108,25 +108,22 @@ type GuardedProviderRequestParams = { mode?: GuardedFetchMode; }; -/** Creates a timer-safe absolute operation deadline from an optional total timeout. */ +/** Creates a timer-safe absolute deadline, resolving a lazy total timeout exactly once. */ export function createProviderOperationDeadline(params: { - timeoutMs?: number; + timeoutMs?: ProviderOperationTimeoutMs; label: string; }): ProviderOperationDeadline { - if ( - typeof params.timeoutMs !== "number" || - !Number.isFinite(params.timeoutMs) || - params.timeoutMs <= 0 - ) { + const timeoutMs = typeof params.timeoutMs === "function" ? params.timeoutMs() : params.timeoutMs; + if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { return { label: params.label }; } - const timeoutMs = resolveTimerTimeoutMs(params.timeoutMs, 1); + const resolvedTimeoutMs = resolveTimerTimeoutMs(timeoutMs, 1); const deadlineAtMs = - resolveExpiresAtMsFromDurationMs(timeoutMs) ?? resolveDateTimestampMs(Date.now()); + resolveExpiresAtMsFromDurationMs(resolvedTimeoutMs) ?? resolveDateTimestampMs(Date.now()); return { deadlineAtMs, label: params.label, - timeoutMs, + timeoutMs: resolvedTimeoutMs, }; } @@ -142,11 +139,42 @@ export function resolveProviderOperationTimeoutMs(params: { } const remainingMs = deadlineAtMs - Date.now(); if (remainingMs <= 0) { - throw new Error(`${params.deadline.label} timed out after ${params.deadline.timeoutMs}ms`); + throw createProviderOperationTimeoutError(params.deadline); } return Math.max(1, Math.min(defaultTimeoutMs, remainingMs)); } +/** Builds the canonical error for an exhausted provider operation deadline. */ +function createProviderOperationTimeoutError(deadline: ProviderOperationDeadline): Error { + const timeoutLabel = + typeof deadline.timeoutMs === "number" ? ` after ${deadline.timeoutMs}ms` : ""; + return new Error(`${deadline.label} timed out${timeoutLabel}`); +} + +/** Resolves a static or lazy request timeout with a validated fallback. */ +function resolveProviderRequestTimeoutMs(params: { + timeoutMs?: ProviderOperationTimeoutMs; + defaultTimeoutMs: number; +}): number { + const resolved = typeof params.timeoutMs === "function" ? params.timeoutMs() : params.timeoutMs; + const fallback = resolveTimerTimeoutMs(params.defaultTimeoutMs, DEFAULT_GUARDED_HTTP_TIMEOUT_MS); + if (typeof resolved !== "number" || !Number.isFinite(resolved) || resolved <= 0) { + return fallback; + } + return resolveTimerTimeoutMs(resolved, fallback); +} + +/** Returns lazy body-read options tied to the same absolute provider operation deadline. */ +function createProviderOperationBodyReadOptions(params: { + deadline: ProviderOperationDeadline; + defaultTimeoutMs: number; +}) { + return { + timeoutMs: createProviderOperationTimeoutResolver(params), + onTimeout: () => createProviderOperationTimeoutError(params.deadline), + }; +} + /** Returns a lazy timeout resolver for code paths that retry or poll multiple HTTP calls. */ export function createProviderOperationTimeoutResolver(params: { deadline: ProviderOperationDeadline; @@ -170,7 +198,7 @@ export async function waitProviderOperationPollInterval(params: { } const remainingMs = deadlineAtMs - Date.now(); if (remainingMs <= 0) { - throw new Error(`${params.deadline.label} timed out after ${params.deadline.timeoutMs}ms`); + throw createProviderOperationTimeoutError(params.deadline); } await new Promise((resolve) => { setTimeout(resolve, Math.min(pollIntervalMs, remainingMs)); @@ -192,6 +220,10 @@ export async function pollProviderOperationJson( getFailureMessage?: (payload: TPayload) => string | undefined; } & GuardedProviderRequestParams, ): Promise { + const bodyReadOptions = createProviderOperationBodyReadOptions({ + deadline: params.deadline, + defaultTimeoutMs: params.defaultTimeoutMs, + }); for (let attempt = 0; attempt < params.maxAttempts; attempt += 1) { const init = { method: "GET", @@ -217,6 +249,7 @@ export async function pollProviderOperationJson( return (await readProviderJsonObjectResponse( result.response, params.requestFailedMessage, + bodyReadOptions, )) as TPayload; } finally { await result.release(); @@ -232,6 +265,7 @@ export async function pollProviderOperationJson( requestFailedMessage: params.requestFailedMessage, }), params.requestFailedMessage, + bodyReadOptions, )) as TPayload); if (params.isComplete(payload)) { return payload; @@ -263,34 +297,65 @@ export async function fetchProviderOperationResponse(params: { stage: params.stage, retry: params.retry, operation: async () => { + const timeoutMs = resolveProviderRequestTimeoutMs({ + timeoutMs: params.timeoutMs, + defaultTimeoutMs: DEFAULT_GUARDED_HTTP_TIMEOUT_MS, + }); + const requestDeadline = createProviderOperationDeadline({ + timeoutMs, + label: params.requestFailedMessage ?? `${params.provider ?? "provider"} ${params.stage}`, + }); const response = await fetchWithTimeout( params.url, params.init ?? {}, - resolveProviderOperationRequestTimeoutMs(params.timeoutMs), + timeoutMs, params.fetchFn, ); if (params.requestFailedMessage) { - await assertOkOrThrowHttpError(response, params.requestFailedMessage); + await assertOkOrThrowHttpError(response, params.requestFailedMessage, { + bodyTimeoutMs: createProviderOperationTimeoutResolver({ + deadline: requestDeadline, + defaultTimeoutMs: timeoutMs, + }), + onBodyTimeout: () => createProviderOperationTimeoutError(requestDeadline), + }); } return response; }, }); } +/** + * Fetches generated-asset response headers and bounded error details under an absolute deadline. + * Successful-body readers must reuse the same deadline so header time cannot reset the budget. + */ export async function fetchProviderDownloadResponse(params: { url: string; init?: RequestInit; + deadline?: ProviderOperationDeadline; + /** @deprecated Pass `deadline` so successful-body reads can reuse the same total budget. */ timeoutMs?: ProviderOperationTimeoutMs; fetchFn: typeof fetch; provider?: string; requestFailedMessage: string; retry?: TransientProviderRetryConfig; }): Promise { + // timeoutMs is a shipped Plugin SDK contract. Normalize it at this boundary; + // new callers pass the deadline through to their successful-body reader. + const deadline = + params.deadline ?? + createProviderOperationDeadline({ + timeoutMs: params.timeoutMs, + label: params.requestFailedMessage, + }); return await fetchProviderOperationResponse({ stage: "download", url: params.url, init: params.init, - timeoutMs: params.timeoutMs, + timeoutMs: createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_GUARDED_HTTP_TIMEOUT_MS, + }), fetchFn: params.fetchFn, provider: params.provider, requestFailedMessage: params.requestFailedMessage, @@ -298,16 +363,6 @@ export async function fetchProviderDownloadResponse(params: { }); } -function resolveProviderOperationRequestTimeoutMs( - timeoutMs: ProviderOperationTimeoutMs | undefined, -): number { - const resolved = typeof timeoutMs === "function" ? timeoutMs() : timeoutMs; - if (typeof resolved !== "number" || !Number.isFinite(resolved) || resolved <= 0) { - return DEFAULT_GUARDED_HTTP_TIMEOUT_MS; - } - return resolved; -} - function resolveGuardedHttpTimeoutMs(timeoutMs: number | undefined): number { if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { return DEFAULT_GUARDED_HTTP_TIMEOUT_MS; @@ -549,16 +604,30 @@ async function fetchGuardedProviderOperationResponse(params: { stage: params.stage, retry: params.retry, operation: async () => { + const timeoutMs = resolveProviderRequestTimeoutMs({ + timeoutMs: params.timeoutMs, + defaultTimeoutMs: DEFAULT_GUARDED_HTTP_TIMEOUT_MS, + }); + const requestDeadline = createProviderOperationDeadline({ + timeoutMs, + label: params.requestFailedMessage ?? `${params.provider ?? "provider"} ${params.stage}`, + }); const result = await fetchWithTimeoutGuarded( params.url, params.init, - resolveProviderOperationRequestTimeoutMs(params.timeoutMs), + timeoutMs, params.fetchFn, params.guardedOptions, ); try { if (params.requestFailedMessage) { - await assertOkOrThrowHttpError(result.response, params.requestFailedMessage); + await assertOkOrThrowHttpError(result.response, params.requestFailedMessage, { + bodyTimeoutMs: createProviderOperationTimeoutResolver({ + deadline: requestDeadline, + defaultTimeoutMs: timeoutMs, + }), + onBodyTimeout: () => createProviderOperationTimeoutError(requestDeadline), + }); } return result; } catch (error) { diff --git a/src/music-generation/provider-assets.ts b/src/music-generation/provider-assets.ts index 30fe2c39adc..3a104883d37 100644 --- a/src/music-generation/provider-assets.ts +++ b/src/music-generation/provider-assets.ts @@ -4,7 +4,11 @@ import { extensionForMime } from "@openclaw/media-core/mime"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { readResponseWithLimit } from "../infra/http-body.js"; -import { fetchProviderDownloadResponse } from "../media-understanding/shared.js"; +import { + createProviderOperationDeadline, + createProviderOperationTimeoutResolver, + fetchProviderDownloadResponse, +} from "../media-understanding/shared.js"; import type { GeneratedMusicAsset } from "./types.js"; /** @@ -98,10 +102,18 @@ export async function downloadGeneratedMusicAsset(params: { index?: number; maxBytes?: number; }): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs, + label: `${params.provider} generated music download`, + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: params.timeoutMs, + }); const response = await fetchProviderDownloadResponse({ url: params.candidate.url, init: { method: "GET" }, - timeoutMs: params.timeoutMs, + deadline, fetchFn: params.fetchFn, provider: params.provider, requestFailedMessage: params.requestFailedMessage, @@ -114,6 +126,11 @@ export async function downloadGeneratedMusicAsset(params: { const maxBytes = params.maxBytes ?? maxBytesForKind("audio"); return { buffer: await readResponseWithLimit(response, maxBytes, { + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + new Error( + `${params.provider} generated music download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, + ), onOverflow: ({ maxBytes: maxBytesLocal }) => new Error(`${params.provider} generated music download exceeds ${maxBytesLocal} bytes`), }), diff --git a/src/plugin-sdk/provider-tools.test.ts b/src/plugin-sdk/provider-tools.test.ts index 18311eccc98..a0054b18927 100644 --- a/src/plugin-sdk/provider-tools.test.ts +++ b/src/plugin-sdk/provider-tools.test.ts @@ -437,6 +437,49 @@ describe("buildProviderToolCompatFamilyHooks", () => { } }); + it("repairs legacy and content schema applicators without changing property dependencies", () => { + expect( + normalizeOpenAIParameters({ + type: "object", + properties: {}, + required: [], + additionalProperties: false, + dependencies: { + mode: ["payload"], + payload: { type: "object" }, + }, + additionalItems: { type: "object" }, + contentSchema: { type: "object" }, + }), + ).toEqual({ + type: "object", + properties: {}, + required: [], + additionalProperties: false, + dependencies: { + mode: ["payload"], + payload: { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + }, + }, + additionalItems: { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + }, + contentSchema: { + type: "object", + properties: {}, + required: [], + additionalProperties: false, + }, + }); + }); + it("does not tighten or warn for permissive object schemas that use strict:false", () => { const hooks = buildProviderToolCompatFamilyHooks("openai"); const permissiveParameters = { diff --git a/src/plugin-sdk/provider-tools.ts b/src/plugin-sdk/provider-tools.ts index 2cc937d61d2..ce58dad6857 100644 --- a/src/plugin-sdk/provider-tools.ts +++ b/src/plugin-sdk/provider-tools.ts @@ -1,6 +1,8 @@ import { cleanSchemaForGemini, + findOpenAIStrictSchemaViolations, GEMINI_UNSUPPORTED_SCHEMA_KEYWORDS, + normalizeOpenAIStrictCompatSchema, stripUnsupportedSchemaKeywords, } from "@openclaw/ai/internal/openai"; // Provider tool helpers expose shared tool-call payload contracts for provider plugins. @@ -11,7 +13,12 @@ import type { ProviderToolSchemaDiagnostic, } from "./plugin-entry.js"; -export { cleanSchemaForGemini, GEMINI_UNSUPPORTED_SCHEMA_KEYWORDS, stripUnsupportedSchemaKeywords }; +export { + cleanSchemaForGemini, + findOpenAIStrictSchemaViolations, + GEMINI_UNSUPPORTED_SCHEMA_KEYWORDS, + stripUnsupportedSchemaKeywords, +}; /** * Finds unsupported JSON-schema keywords and reports their nested schema paths. @@ -126,12 +133,6 @@ export function normalizeOpenAIToolSchemas( }); } -function normalizeOpenAIStrictCompatSchema(schema: unknown): TSchema { - return normalizeOpenAIStrictCompatSchemaRecursive(schema, { - promoteEmptyObject: true, - }) as TSchema; -} - function shouldApplyOpenAIToolCompat(ctx: ProviderNormalizeToolSchemasContext): boolean { const provider = (ctx.model?.provider ?? ctx.provider ?? "").trim().toLowerCase(); const api = (ctx.model?.api ?? ctx.modelApi ?? "").trim().toLowerCase(); @@ -161,246 +162,6 @@ function isOpenAICodexBaseUrl(baseUrl: string): boolean { return /^https:\/\/chatgpt\.com\/backend-api(?:\/|$)/i.test(baseUrl); } -type NormalizeOpenAIStrictCompatOptions = { - promoteEmptyObject: boolean; -}; - -const OPENAI_STRICT_COMPAT_SCHEMA_MAP_KEYS = new Set([ - "$defs", - "definitions", - "dependentSchemas", - "patternProperties", - "properties", -]); - -// Annotation-only keywords whose null values can be dropped without changing -// what the schema accepts; null constraint keywords must stay so projection -// quarantines the tool instead of widening it. -const OPENAI_NULLABLE_ANNOTATION_KEYS = new Set([ - "default", - "description", - "examples", - "format", - "title", -]); - -const OPENAI_STRICT_COMPAT_SCHEMA_NESTED_KEYS = new Set([ - "additionalProperties", - "allOf", - "anyOf", - "contains", - "else", - "if", - "items", - "not", - "oneOf", - "prefixItems", - "propertyNames", - "then", - "unevaluatedItems", - "unevaluatedProperties", -]); - -function normalizeOpenAIStrictCompatSchemaMap(schema: unknown): unknown { - if (!schema || typeof schema !== "object" || Array.isArray(schema)) { - return schema; - } - - let changed = false; - const normalized: Record = {}; - for (const [key, value] of Object.entries(schema as Record)) { - const next = normalizeOpenAIStrictCompatSchemaRecursive(value, { - promoteEmptyObject: false, - }); - normalized[key] = next; - changed ||= next !== value; - } - return changed ? normalized : schema; -} - -function normalizeOpenAIStrictCompatSchemaRecursive( - schema: unknown, - options: NormalizeOpenAIStrictCompatOptions, -): unknown { - if (Array.isArray(schema)) { - let changed = false; - const normalized = schema.map((entry) => { - const next = normalizeOpenAIStrictCompatSchemaRecursive(entry, { - promoteEmptyObject: false, - }); - changed ||= next !== entry; - return next; - }); - return changed ? normalized : schema; - } - if (!schema || typeof schema !== "object") { - return schema; - } - - const record = schema as Record; - let changed = false; - let hadNullType = false; - const normalized: Record = {}; - for (const [key, value] of Object.entries(record)) { - // Repair only null-valued entries that carry no constraint semantics: - // annotation keywords, and `type: null` when the shape inference below can - // supply the real type. Null constraint keywords (additionalProperties, - // required, enum, ...) stay put so projection quarantines instead of - // silently accepting arguments the plugin never declared. - if (value === null && OPENAI_NULLABLE_ANNOTATION_KEYS.has(key)) { - changed = true; - continue; - } - if (value === null && key === "type") { - hadNullType = true; - changed = true; - continue; - } - const next = OPENAI_STRICT_COMPAT_SCHEMA_MAP_KEYS.has(key) - ? normalizeOpenAIStrictCompatSchemaMap(value) - : OPENAI_STRICT_COMPAT_SCHEMA_NESTED_KEYS.has(key) - ? normalizeOpenAIStrictCompatSchemaRecursive(value, { - promoteEmptyObject: false, - }) - : value; - normalized[key] = next; - changed ||= next !== value; - } - - if (Object.keys(normalized).length === 0) { - if (!options.promoteEmptyObject) { - return schema; - } - return { - type: "object", - properties: {}, - required: [], - additionalProperties: false, - }; - } - - const hasObjectShapeHints = - (normalized.properties && - typeof normalized.properties === "object" && - !Array.isArray(normalized.properties)) || - Array.isArray(normalized.required); - const hasArrayShapeHints = "items" in normalized; - if (!("type" in normalized) && hasObjectShapeHints !== hasArrayShapeHints) { - normalized.type = hasObjectShapeHints ? "object" : "array"; - changed = true; - } else if (hadNullType && !("type" in normalized)) { - // Inference failed (no hints, or ambiguous object+array hints): restore the - // invalid `type: null` so downstream projection quarantines the tool - // instead of shipping an unconstrained schema. - normalized.type = null; - } - if (normalized.type === "object" && !("properties" in normalized)) { - normalized.properties = {}; - changed = true; - } - - const hasEmptyProperties = - normalized.properties && - typeof normalized.properties === "object" && - !Array.isArray(normalized.properties) && - Object.keys(normalized.properties as Record).length === 0; - - if (normalized.type === "object" && !Array.isArray(normalized.required) && hasEmptyProperties) { - normalized.required = []; - changed = true; - } - - if ( - normalized.type === "object" && - hasEmptyProperties && - !("additionalProperties" in normalized) - ) { - normalized.additionalProperties = false; - changed = true; - } - - return changed ? normalized : schema; -} - -/** - * Finds schema paths that violate OpenAI strict tool-schema requirements. - */ -export function findOpenAIStrictSchemaViolations( - /** JSON schema node to inspect recursively. */ - schema: unknown, - /** Dot/bracket path prefix used in returned diagnostics. */ - path: string, - /** Strictness controls for the current schema position. */ - options?: { requireObjectRoot?: boolean }, -): string[] { - if (Array.isArray(schema)) { - if (options?.requireObjectRoot) { - return [`${path}.type`]; - } - return schema.flatMap((item, index) => - findOpenAIStrictSchemaViolations(item, `${path}[${index}]`), - ); - } - if (!schema || typeof schema !== "object") { - if (options?.requireObjectRoot) { - return [`${path}.type`]; - } - return []; - } - - const record = schema as Record; - const violations: string[] = []; - for (const key of ["anyOf", "oneOf", "allOf"] as const) { - if (Array.isArray(record[key])) { - violations.push(`${path}.${key}`); - } - } - if (Array.isArray(record.type)) { - violations.push(`${path}.type`); - } - - const properties = - record.properties && typeof record.properties === "object" && !Array.isArray(record.properties) - ? (record.properties as Record) - : undefined; - - if (record.type === "object") { - if (record.additionalProperties !== false) { - violations.push(`${path}.additionalProperties`); - } - const required = Array.isArray(record.required) - ? record.required.filter((entry): entry is string => typeof entry === "string") - : undefined; - if (!required) { - violations.push(`${path}.required`); - } else if (properties) { - const requiredSet = new Set(required); - for (const key of Object.keys(properties)) { - if (!requiredSet.has(key)) { - violations.push(`${path}.required.${key}`); - } - } - } - } - - if (properties) { - for (const [key, value] of Object.entries(properties)) { - violations.push(...findOpenAIStrictSchemaViolations(value, `${path}.properties.${key}`)); - } - } - - for (const [key, value] of Object.entries(record)) { - if (key === "properties") { - continue; - } - if (value && typeof value === "object") { - violations.push(...findOpenAIStrictSchemaViolations(value, `${path}.${key}`)); - } - } - - return violations; -} - /** * Reports OpenAI strict-schema diagnostics for transports that enforce them before dispatch. */ diff --git a/src/plugin-sdk/test-helpers/provider-http-mocks.ts b/src/plugin-sdk/test-helpers/provider-http-mocks.ts index c1df54b5a2a..d86fae65268 100644 --- a/src/plugin-sdk/test-helpers/provider-http-mocks.ts +++ b/src/plugin-sdk/test-helpers/provider-http-mocks.ts @@ -213,6 +213,15 @@ function resolveMockProviderTimeoutMs( return typeof timeoutMs === "function" ? timeoutMs() : (timeoutMs ?? 60_000); } +function resolveMockProviderDownloadTimeoutMs(params: FetchProviderDownloadResponseParams) { + if (!params.deadline) { + return resolveMockProviderTimeoutMs(params.timeoutMs); + } + return params.deadline.deadlineAtMs === undefined + ? (params.deadline.timeoutMs ?? 60_000) + : Math.max(1, params.deadline.deadlineAtMs - Date.now()); +} + providerHttpMocks.fetchProviderOperationResponseMock.mockImplementation( async (params: FetchProviderOperationResponseParams) => { const response = await providerHttpMocks.fetchWithTimeoutMock( @@ -233,7 +242,7 @@ providerHttpMocks.fetchProviderDownloadResponseMock.mockImplementation( const response = await providerHttpMocks.fetchWithTimeoutMock( params.url, params.init ?? {}, - resolveMockProviderTimeoutMs(params.timeoutMs), + resolveMockProviderDownloadTimeoutMs(params), params.fetchFn, ); await providerHttpMocks.assertOkOrThrowHttpErrorMock(response, params.requestFailedMessage); @@ -280,15 +289,34 @@ vi.mock("openclaw/plugin-sdk/provider-http", () => ({ timeoutMs, }: { label: string; - timeoutMs?: number; - }) => ({ - label, - timeoutMs, - }), + timeoutMs?: number | (() => number); + }) => { + const resolvedTimeoutMs = typeof timeoutMs === "function" ? timeoutMs() : timeoutMs; + return { + label, + timeoutMs: resolvedTimeoutMs, + deadlineAtMs: + typeof resolvedTimeoutMs === "number" ? Date.now() + resolvedTimeoutMs : undefined, + }; + }, createProviderOperationTimeoutResolver: - ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) => - () => + ({ + deadline, defaultTimeoutMs, + }: { + deadline: { deadlineAtMs?: number; label: string; timeoutMs?: number }; + defaultTimeoutMs: number; + }) => + () => { + if (typeof deadline.deadlineAtMs !== "number") { + return defaultTimeoutMs; + } + const remainingMs = deadline.deadlineAtMs - Date.now(); + if (remainingMs <= 0) { + throw new Error(`${deadline.label} timed out after ${deadline.timeoutMs}ms`); + } + return Math.min(defaultTimeoutMs, remainingMs); + }, executeProviderOperationWithRetry: providerHttpMocks.executeProviderOperationWithRetryMock, fetchProviderDownloadResponse: providerHttpMocks.fetchProviderDownloadResponseMock, fetchProviderOperationResponse: providerHttpMocks.fetchProviderOperationResponseMock, diff --git a/test/scripts/control-ui-performance.test.ts b/test/scripts/control-ui-performance.test.ts index 463ac622d0a..cc52fdcc8c6 100644 --- a/test/scripts/control-ui-performance.test.ts +++ b/test/scripts/control-ui-performance.test.ts @@ -102,6 +102,49 @@ describe("Control UI performance budgets", () => { ); }); + it("includes exact bytes when rounded violation values collide", () => { + const metrics = { + schemaVersion: 1 as const, + startup: { + js: { requests: 1, rawBytes: 100, gzipBytes: 43_009, brotliBytes: 30 }, + css: { requests: 1, rawBytes: 50, gzipBytes: 15, brotliBytes: 12 }, + assets: [], + }, + total: { + js: { requests: 1, rawBytes: 100, gzipBytes: 43_009, brotliBytes: 30 }, + css: { requests: 1, rawBytes: 50, gzipBytes: 15, brotliBytes: 12 }, + }, + largest: { + js: { + file: "assets/index-a.js", + type: "js", + rawBytes: 100, + gzipBytes: 43_009, + brotliBytes: 30, + }, + css: { + file: "assets/index-c.css", + type: "css", + rawBytes: 50, + gzipBytes: 15, + brotliBytes: 12, + }, + }, + } satisfies ReturnType; + const budgets = { + startupJsRequests: 1, + startupCssRequests: 1, + startupJsGzipBytes: 43_008, + startupCssGzipBytes: 20, + largestJsGzipBytes: 43_008, + largestCssGzipBytes: 20, + }; + + expect(formatControlUiPerformanceReport(metrics, budgets)).toContain( + "startup JS gzip: 42.0 KiB exceeds 42.0 KiB (43009 B vs 43008 B)", + ); + }); + it("fails when a compressed sidecar is missing", () => { const { distDir } = createDistFixture(); fs.writeFileSync(