mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-21 02:06:43 +00:00
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 <huang.ting3@xydigit.com> * 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 <huang.ting3@xydigit.com>
This commit is contained in:
Peter Steinberger
co-authored by
Pick-cat
Pick-cat
parent
e01a880084
commit
40793e6f8b
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<typeof import("openclaw/plugin-sdk/provider-http")>();
|
||||
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<Uint8Array>({
|
||||
async pull(controller) {
|
||||
controller.enqueue(new Uint8Array([1]));
|
||||
await new Promise<void>((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" });
|
||||
|
||||
|
||||
@@ -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<BytePlusTaskResponse>(
|
||||
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<BytePlusTaskResponse>({
|
||||
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<GeneratedVideoAsset> {
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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<string> {
|
||||
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.");
|
||||
}
|
||||
|
||||
@@ -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<string, unknown> | 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 () => {
|
||||
|
||||
@@ -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<GeneratedMusicAsset> {
|
||||
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<Buffer> {
|
||||
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<typeof setTimeout> | undefined;
|
||||
try {
|
||||
const timeoutMs = resolveBodyReadTimeoutMs(deadline);
|
||||
const timeoutPromise = new Promise<never>((_, 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(
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<MinimaxQueryResponse>(
|
||||
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<GeneratedVideoAsset> {
|
||||
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<GeneratedVideoAsset> {
|
||||
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<MinimaxFileRetrieveResponse>(
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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<typeof createProviderOperationDeadline>;
|
||||
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<GeneratedVideoAsset> {
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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<RunwayTaskDetailResponse>(
|
||||
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<RunwayTaskDetailResponse>({
|
||||
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<GeneratedVideoAsset[]> {
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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<number> {
|
||||
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");
|
||||
|
||||
@@ -153,16 +153,29 @@ async function downloadTogetherVideo(params: {
|
||||
fetchFn: typeof fetch;
|
||||
maxBytes: number;
|
||||
}): Promise<GeneratedVideoAsset> {
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
|
||||
+49
-88
@@ -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<typeof resolveApiKeyForProvider>[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<unknown>(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<unknown>({
|
||||
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: {
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<GeneratedVideoAsset> {
|
||||
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`),
|
||||
});
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<string, unknown> = {};
|
||||
for (const [key, value] of Object.entries(schema as Record<string, unknown>)) {
|
||||
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<string, unknown>;
|
||||
let changed = false;
|
||||
let hadNullType = false;
|
||||
const normalized: Record<string, unknown> = {};
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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<string, unknown>)
|
||||
: 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<string, unknown>)) {
|
||||
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;
|
||||
}
|
||||
@@ -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<string, unknown>;
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
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<string, unknown>)
|
||||
: {};
|
||||
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,
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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 {}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -176,6 +176,56 @@ describe("provider error utils", () => {
|
||||
} satisfies Partial<ProviderHttpError>);
|
||||
});
|
||||
|
||||
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<Uint8Array>({
|
||||
pull() {
|
||||
return new Promise<void>(() => {});
|
||||
},
|
||||
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<Uint8Array>({
|
||||
pull() {
|
||||
return new Promise<void>(() => {});
|
||||
},
|
||||
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);
|
||||
|
||||
@@ -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<ReadResponseTextPrefixOptions["onTimeout"]>;
|
||||
};
|
||||
|
||||
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<string, unknown> | 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<string> {
|
||||
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<string> {
|
||||
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<ProviderHttpErrorInfo> {
|
||||
const rawBody = trimToUndefined(await readResponseTextLimited(response).catch(() => ""));
|
||||
async function extractProviderErrorInfo(
|
||||
response: Response,
|
||||
options?: ProviderHttpErrorOptions,
|
||||
): Promise<ProviderHttpErrorInfo> {
|
||||
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<Error> {
|
||||
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<ProviderHttpErrorOptions, "statusPrefix">,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
export async function assertOkOrThrowHttpError(
|
||||
response: Response,
|
||||
label: string,
|
||||
options?: Omit<ProviderHttpErrorOptions, "statusPrefix">,
|
||||
): Promise<void> {
|
||||
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<T>(
|
||||
response: Response,
|
||||
label: string,
|
||||
opts?: { maxBytes?: number },
|
||||
opts?: ProviderResponseReadOptions,
|
||||
): Promise<T> {
|
||||
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<T>(
|
||||
export async function readProviderJsonObjectResponse(
|
||||
response: Response,
|
||||
label: string,
|
||||
opts?: ProviderResponseReadOptions,
|
||||
): Promise<Record<string, unknown>> {
|
||||
const payload = await readProviderJsonResponse<unknown>(response, label);
|
||||
const payload = await readProviderJsonResponse<unknown>(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<unknown[]> {
|
||||
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<Uint8Array> {
|
||||
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`),
|
||||
});
|
||||
|
||||
+9
-13
@@ -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<string> {
|
||||
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 <path>.",
|
||||
);
|
||||
}
|
||||
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 <path> 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 <path> for larger patches.`,
|
||||
),
|
||||
});
|
||||
return bytes.toString("utf8");
|
||||
}
|
||||
|
||||
async function readConfigPatchInput(opts: ConfigPatchOptions): Promise<unknown> {
|
||||
|
||||
@@ -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<string> {
|
||||
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<string | null> {
|
||||
|
||||
@@ -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 <T>(params: Parameters<typeof clackSelect<T>>[0]) =>
|
||||
const MODELS_AUTH_STDIN_MAX_BYTES = 1024 * 1024;
|
||||
|
||||
async function readPipedStdin(): Promise<string> {
|
||||
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: {
|
||||
|
||||
@@ -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<Uint8Array>({
|
||||
pull() {
|
||||
return new Promise<void>(() => {});
|
||||
},
|
||||
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 {
|
||||
|
||||
+11
-3
@@ -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<ReadResponsePrefixResult> {
|
||||
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),
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
}
|
||||
|
||||
function createTricklingResponse(status = 200): Response {
|
||||
const chunk = new TextEncoder().encode("{");
|
||||
return new Response(
|
||||
new ReadableStream<Uint8Array>({
|
||||
async pull(controller) {
|
||||
controller.enqueue(chunk);
|
||||
await new Promise<void>((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<typeof fetch>().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<typeof fetch>().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",
|
||||
|
||||
@@ -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<TPayload>(
|
||||
getFailureMessage?: (payload: TPayload) => string | undefined;
|
||||
} & GuardedProviderRequestParams,
|
||||
): Promise<TPayload> {
|
||||
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<TPayload>(
|
||||
return (await readProviderJsonObjectResponse(
|
||||
result.response,
|
||||
params.requestFailedMessage,
|
||||
bodyReadOptions,
|
||||
)) as TPayload;
|
||||
} finally {
|
||||
await result.release();
|
||||
@@ -232,6 +265,7 @@ export async function pollProviderOperationJson<TPayload>(
|
||||
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<Response> {
|
||||
// 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) {
|
||||
|
||||
@@ -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<GeneratedMusicAsset> {
|
||||
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`),
|
||||
}),
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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<string, unknown> = {};
|
||||
for (const [key, value] of Object.entries(schema as Record<string, unknown>)) {
|
||||
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<string, unknown>;
|
||||
let changed = false;
|
||||
let hadNullType = false;
|
||||
const normalized: Record<string, unknown> = {};
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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<string, unknown>)
|
||||
: 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.
|
||||
*/
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof collectControlUiPerformanceMetrics>;
|
||||
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(
|
||||
|
||||
Reference in New Issue
Block a user