vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-13 09:36:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: thread openai video request policy

Browse Source
This commit is contained in:
Shakker
2026-05-23 00:12:58 +01:00
committed by Shakker
parent cc48c34f91
commit 31b5145594
5 changed files with 310 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
extensions/openai/video-generation-provider.test.ts
View File

@@ -9,6 +9,8 @@ const {
postJsonRequestMock,
postMultipartRequestMock,
fetchWithTimeoutMock,
fetchWithTimeoutGuardedMock,
pollProviderOperationJsonMock,
resolveProviderHttpRequestConfigMock,
sanitizeConfiguredModelProviderRequestMock,
} = getProviderHttpMocks();
@@ -49,6 +51,28 @@ function fetchWithTimeoutCall(index: number): [string, RequestInit | undefined,
return call;
}
function fetchWithTimeoutGuardedCall(
index = 0,
): [string, RequestInit | undefined, number, unknown, Record<string, unknown> | undefined] {
const call = fetchWithTimeoutGuardedMock.mock.calls[index] as
| [string, RequestInit | undefined, number, unknown, Record<string, unknown> | undefined]
| undefined;
if (!call) {
throw new Error(`expected fetchWithTimeoutGuarded call ${index}`);
}
return call;
}
function pollProviderOperationRequest(index = 0): Record<string, unknown> {
const request = pollProviderOperationJsonMock.mock.calls[index]?.[0] as
| Record<string, unknown>
| undefined;
if (!request) {
throw new Error(`expected pollProviderOperationJson call ${index}`);
}
return request;
}
function providerHttpConfigRequest(): Record<string, unknown> {
const [call] = resolveProviderHttpRequestConfigMock.mock.calls;
if (!call) {
@@ -263,6 +287,20 @@ describe("openai video generation provider", () => {
const createRequest = postJsonRequest();
expect(createRequest.url).toBe("http://127.0.0.1:44080/v1/videos");
expect(createRequest.allowPrivateNetwork).toBe(true);
const statusRequest = pollProviderOperationRequest();
expect(statusRequest.url).toBe("http://127.0.0.1:44080/v1/videos/vid_local");
expect(statusRequest.allowPrivateNetwork).toBe(true);
expect(statusRequest.auditContext).toBe("openai-video-status");
const [downloadUrl, downloadInit, downloadTimeout, downloadFetch, downloadOptions] =
fetchWithTimeoutGuardedCall();
expect(downloadUrl).toBe("http://127.0.0.1:44080/v1/videos/vid_local/content?variant=video");
expect(downloadInit?.method).toBe("GET");
expect(downloadTimeout).toBe(120000);
expect(downloadFetch).toBe(fetch);
expect(downloadOptions).toEqual({
ssrfPolicy: { allowPrivateNetwork: true },
auditContext: "openai-video-download",
});
});
it("uses multipart input_reference for video-to-video uploads", async () => {
@@ -305,6 +343,59 @@ describe("openai video generation provider", () => {
expect(createRequest.allowPrivateNetwork).toBe(false);
});
it("honors configured request allowPrivateNetwork for multipart video uploads", async () => {
fetchWithTimeoutMock
.mockResolvedValueOnce({
ok: true,
json: async () => ({
id: "vid_789",
model: "sora-2",
status: "queued",
}),
})
.mockResolvedValueOnce({
json: async () => ({
id: "vid_789",
model: "sora-2",
status: "completed",
}),
})
.mockResolvedValueOnce({
headers: new Headers({ "content-type": "video/mp4" }),
arrayBuffer: async () => Buffer.from("mp4-bytes"),
});
const provider = buildOpenAIVideoGenerationProvider();
await provider.generateVideo({
provider: "openai",
model: "sora-2",
prompt: "Remix this clip",
cfg: {
models: {
providers: {
openai: {
baseUrl: "http://127.0.0.1:44080/v1",
request: { allowPrivateNetwork: true },
models: [],
},
},
},
},
inputVideos: [{ buffer: Buffer.from("mp4-bytes"), mimeType: "video/mp4" }],
});
expect(postJsonRequestMock).not.toHaveBeenCalled();
const createRequest = postMultipartRequest();
expect(createRequest.url).toBe("http://127.0.0.1:44080/v1/videos");
expect(createRequest.body).toBeInstanceOf(FormData);
expect(createRequest.allowPrivateNetwork).toBe(true);
expect(pollProviderOperationRequest().allowPrivateNetwork).toBe(true);
expect(fetchWithTimeoutGuardedCall()[4]).toEqual({
ssrfPolicy: { allowPrivateNetwork: true },
auditContext: "openai-video-download",
});
});
it("rejects multiple reference assets", async () => {
const provider = buildOpenAIVideoGenerationProvider();

116
extensions/openai/video-generation-provider.ts
View File

@@ -6,6 +6,7 @@ import {
createProviderOperationDeadline,
createProviderOperationTimeoutResolver,
fetchProviderDownloadResponse,
fetchWithTimeoutGuarded,
pollProviderOperationJson,
postJsonRequest,
postMultipartRequest,
@@ -30,6 +31,11 @@ const MAX_POLL_ATTEMPTS = 120;
const OPENAI_VIDEO_SECONDS = [4, 8, 12] as const;
const OPENAI_VIDEO_SIZES = ["720x1280", "1280x720", "1024x1792", "1792x1024"] as const;
type OpenAIVideoRequestPolicy = {
allowPrivateNetwork: boolean;
dispatcherPolicy?: Parameters<typeof postJsonRequest>[0]["dispatcherPolicy"];
};
type OpenAIVideoStatus = "queued" | "in_progress" | "completed" | "failed";
type OpenAIVideoResponse = {
@@ -117,13 +123,15 @@ function resolveReferenceAsset(req: VideoGenerationRequest) {
return new File([toBlobBytes(asset.buffer)], fileName, { type: mimeType });
}
async function pollOpenAIVideo(params: {
videoId: string;
headers: Headers;
timeoutMs?: number;
baseUrl: string;
fetchFn: typeof fetch;
}): Promise<OpenAIVideoResponse> {
async function pollOpenAIVideo(
params: {
videoId: string;
headers: Headers;
timeoutMs?: number;
baseUrl: string;
fetchFn: typeof fetch;
} & OpenAIVideoRequestPolicy,
): Promise<OpenAIVideoResponse> {
const deadline = createProviderOperationDeadline({
timeoutMs: params.timeoutMs,
label: `OpenAI video generation task ${params.videoId}`,
@@ -138,6 +146,9 @@ async function pollOpenAIVideo(params: {
pollIntervalMs: POLL_INTERVAL_MS,
requestFailedMessage: "OpenAI video status request failed",
timeoutMessage: `OpenAI video generation task ${params.videoId} did not finish in time`,
allowPrivateNetwork: params.allowPrivateNetwork,
dispatcherPolicy: params.dispatcherPolicy,
auditContext: "openai-video-status",
isComplete: (payload) => payload.status === "completed",
getFailureMessage: (payload) =>
payload.status === "failed"
@@ -146,16 +157,63 @@ async function pollOpenAIVideo(params: {
});
}
async function downloadOpenAIVideo(params: {
videoId: string;
headers: Headers;
timeoutMs?: ProviderOperationTimeoutMs;
baseUrl: string;
fetchFn: typeof fetch;
}): Promise<GeneratedVideoAsset> {
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;
fetchFn: typeof fetch;
} & OpenAIVideoRequestPolicy,
) {
if (!params.allowPrivateNetwork && !params.dispatcherPolicy) {
const response = await fetchProviderDownloadResponse({
url: params.url,
init: params.init,
timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
fetchFn: params.fetchFn,
provider: "openai",
requestFailedMessage: "OpenAI video download failed",
});
return {
response,
release: async () => {},
};
}
const result = await fetchWithTimeoutGuarded(
params.url,
params.init,
resolveOpenAIVideoDownloadTimeoutMs(params.timeoutMs),
params.fetchFn,
{
...(params.allowPrivateNetwork ? { ssrfPolicy: { allowPrivateNetwork: true } } : {}),
...(params.dispatcherPolicy ? { dispatcherPolicy: params.dispatcherPolicy } : {}),
auditContext: "openai-video-download",
},
);
await assertOkOrThrowHttpError(result.response, "OpenAI video download failed");
return result;
}
async function downloadOpenAIVideo(
params: {
videoId: string;
headers: Headers;
timeoutMs?: ProviderOperationTimeoutMs;
baseUrl: string;
fetchFn: typeof fetch;
} & OpenAIVideoRequestPolicy,
): Promise<GeneratedVideoAsset> {
const url = new URL(`${params.baseUrl}/videos/${params.videoId}/content`);
url.searchParams.set("variant", "video");
const response = await fetchProviderDownloadResponse({
const { response, release } = await fetchOpenAIVideoDownload({
url: url.toString(),
init: {
method: "GET",
@@ -164,18 +222,22 @@ async function downloadOpenAIVideo(params: {
Accept: "application/binary",
}),
},
timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
provider: "openai",
requestFailedMessage: "OpenAI video download failed",
allowPrivateNetwork: params.allowPrivateNetwork,
dispatcherPolicy: params.dispatcherPolicy,
});
const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4";
const arrayBuffer = await response.arrayBuffer();
return {
buffer: Buffer.from(arrayBuffer),
mimeType,
fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`,
};
try {
const mimeType = normalizeOptionalString(response.headers.get("content-type")) ?? "video/mp4";
const arrayBuffer = await response.arrayBuffer();
return {
buffer: Buffer.from(arrayBuffer),
mimeType,
fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`,
};
} finally {
await release();
}
}
export function buildOpenAIVideoGenerationProvider(): VideoGenerationProvider {
@@ -351,6 +413,8 @@ export function buildOpenAIVideoGenerationProvider(): VideoGenerationProvider {
}),
baseUrl,
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
});
const video = await downloadOpenAIVideo({
videoId,
@@ -361,6 +425,8 @@ export function buildOpenAIVideoGenerationProvider(): VideoGenerationProvider {
}),
baseUrl,
fetchFn,
allowPrivateNetwork,
dispatcherPolicy,
});
return {
videos: [video],

33
src/media-understanding/shared.test.ts
View File

@@ -155,6 +155,39 @@ describe("provider operation deadlines", () => {
expect(fetchFn).toHaveBeenCalledTimes(2);
});
it("passes guarded request policy through provider status polling", async () => {
const release = vi.fn(async () => {});
fetchWithSsrFGuardMock.mockResolvedValueOnce({
response: new Response(JSON.stringify({ status: "completed" })),
finalUrl: "https://api.example.com/v1/videos/task-1",
release,
});
const result = await pollProviderOperationJson<{ status?: string }>({
url: "https://api.example.com/v1/videos/task-1",
headers: new Headers({ authorization: "Bearer test" }),
deadline: createProviderOperationDeadline({
label: "video generation task task-1",
}),
defaultTimeoutMs: 5_000,
fetchFn: fetch,
maxAttempts: 3,
pollIntervalMs: 1_000,
requestFailedMessage: "status failed",
timeoutMessage: "task timed out",
allowPrivateNetwork: true,
dispatcherPolicy: { mode: "direct" },
auditContext: "provider-video-status",
isComplete: (payload) => payload.status === "completed",
});
expect(result).toEqual({ status: "completed" });
expect(getFirstGuardedFetchCall().policy).toEqual({ allowPrivateNetwork: true });
expect(getFirstGuardedFetchCall().dispatcherPolicy).toEqual({ mode: "direct" });
expect(getFirstGuardedFetchCall().auditContext).toBe("provider-video-status");
expect(release).toHaveBeenCalledTimes(1);
});
it("throws provider failure messages while polling status JSON", async () => {
const fetchFn = vi
.fn<typeof fetch>()

118
src/media-understanding/shared.ts
View File

@@ -17,7 +17,6 @@ import {
buildProviderRequestDispatcherPolicy,
resolveProviderRequestPolicyConfig,
type ModelProviderRequestTransportOverrides,
type ProviderRequestTransportOverrides,
type ResolvedProviderRequestConfig,
} from "../agents/provider-request-config.js";
import type { GuardedFetchMode, GuardedFetchResult } from "../infra/net/fetch-guard.js";
@@ -82,6 +81,15 @@ export type ProviderOperationDeadline = {
export type ProviderOperationTimeoutMs = number | (() => number);
type GuardedProviderRequestParams = {
pinDns?: boolean;
allowPrivateNetwork?: boolean;
ssrfPolicy?: SsrFPolicy;
dispatcherPolicy?: PinnedDispatcherPolicy;
auditContext?: string;
mode?: GuardedFetchMode;
};
export function createProviderOperationDeadline(params: {
timeoutMs?: number;
label: string;
@@ -139,38 +147,61 @@ export async function waitProviderOperationPollInterval(params: {
await new Promise((resolve) => setTimeout(resolve, Math.min(params.pollIntervalMs, remainingMs)));
}
export async function pollProviderOperationJson<TPayload>(params: {
url: string;
headers: Headers;
deadline: ProviderOperationDeadline;
defaultTimeoutMs: number;
fetchFn: typeof fetch;
maxAttempts: number;
pollIntervalMs: number;
requestFailedMessage: string;
timeoutMessage: string;
isComplete: (payload: TPayload) => boolean;
getFailureMessage?: (payload: TPayload) => string | undefined;
}): Promise<TPayload> {
export async function pollProviderOperationJson<TPayload>(
params: {
url: string;
headers: Headers;
deadline: ProviderOperationDeadline;
defaultTimeoutMs: number;
fetchFn: typeof fetch;
maxAttempts: number;
pollIntervalMs: number;
requestFailedMessage: string;
timeoutMessage: string;
isComplete: (payload: TPayload) => boolean;
getFailureMessage?: (payload: TPayload) => string | undefined;
} & GuardedProviderRequestParams,
): Promise<TPayload> {
for (let attempt = 0; attempt < params.maxAttempts; attempt += 1) {
const response = await fetchProviderOperationResponse({
stage: "poll",
url: params.url,
init: {
method: "GET",
headers: params.headers,
},
timeoutMs: createProviderOperationTimeoutResolver({
deadline: params.deadline,
defaultTimeoutMs: params.defaultTimeoutMs,
}),
fetchFn: params.fetchFn,
requestFailedMessage: params.requestFailedMessage,
const init = {
method: "GET",
headers: params.headers,
};
const timeoutMs = createProviderOperationTimeoutResolver({
deadline: params.deadline,
defaultTimeoutMs: params.defaultTimeoutMs,
});
const payload = (await readProviderJsonObjectResponse(
response,
params.requestFailedMessage,
)) as TPayload;
const guardedOptions = resolveGuardedRequestOptions(params);
const payload = guardedOptions
? await (async () => {
const result = await fetchWithTimeoutGuarded(
params.url,
init,
timeoutMs(),
params.fetchFn,
guardedOptions,
);
try {
await assertOkOrThrowHttpError(result.response, params.requestFailedMessage);
return (await readProviderJsonObjectResponse(
result.response,
params.requestFailedMessage,
)) as TPayload;
} finally {
await result.release();
}
})()
: ((await readProviderJsonObjectResponse(
await fetchProviderOperationResponse({
stage: "poll",
url: params.url,
init,
timeoutMs,
fetchFn: params.fetchFn,
requestFailedMessage: params.requestFailedMessage,
}),
params.requestFailedMessage,
)) as TPayload);
if (params.isComplete(payload)) {
return payload;
}
@@ -404,9 +435,9 @@ export async function fetchWithTimeoutGuarded(
});
}
type GuardedPostRequestOptions = NonNullable<Parameters<typeof fetchWithTimeoutGuarded>[4]>;
type GuardedProviderRequestOptions = NonNullable<Parameters<typeof fetchWithTimeoutGuarded>[4]>;
function mergeGuardedPostSsrfPolicy(params: {
function mergeGuardedRequestSsrfPolicy(params: {
ssrfPolicy?: SsrFPolicy;
allowPrivateNetwork?: boolean;
}): SsrFPolicy | undefined {
@@ -419,14 +450,9 @@ function mergeGuardedPostSsrfPolicy(params: {
return { ...params.ssrfPolicy, allowPrivateNetwork: true };
}
function resolveGuardedPostRequestOptions(params: {
pinDns?: boolean;
allowPrivateNetwork?: boolean;
ssrfPolicy?: SsrFPolicy;
dispatcherPolicy?: PinnedDispatcherPolicy;
auditContext?: string;
mode?: GuardedFetchMode;
}): GuardedPostRequestOptions | undefined {
function resolveGuardedRequestOptions(
params: GuardedProviderRequestParams,
): GuardedProviderRequestOptions | undefined {
if (
!params.allowPrivateNetwork &&
!params.ssrfPolicy &&
@@ -437,7 +463,7 @@ function resolveGuardedPostRequestOptions(params: {
) {
return undefined;
}
const ssrfPolicy = mergeGuardedPostSsrfPolicy(params);
const ssrfPolicy = mergeGuardedRequestSsrfPolicy(params);
return {
...(ssrfPolicy ? { ssrfPolicy } : {}),
...(params.pinDns !== undefined ? { pinDns: params.pinDns } : {}),
@@ -485,7 +511,7 @@ export async function postTranscriptionRequest(
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
guardedOptions: resolveGuardedPostRequestOptions(params),
guardedOptions: resolveGuardedRequestOptions(params),
retryStage: params.retryStage,
retry: params.retry,
});
@@ -496,7 +522,7 @@ async function postGuardedRequest(params: {
init: RequestInit;
timeoutMs?: number;
fetchFn: typeof fetch;
guardedOptions?: GuardedPostRequestOptions;
guardedOptions?: GuardedProviderRequestOptions;
retryStage?: ProviderOperationRetryStage;
retry?: TransientProviderRetryConfig;
}) {
@@ -563,7 +589,7 @@ export async function postJsonRequest(
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
guardedOptions: resolveGuardedPostRequestOptions(params),
guardedOptions: resolveGuardedRequestOptions(params),
retryStage: params.retryStage,
retry: params.retry,
});
@@ -598,7 +624,7 @@ export async function postMultipartRequest(
},
timeoutMs: params.timeoutMs,
fetchFn: params.fetchFn,
guardedOptions: resolveGuardedPostRequestOptions(params),
guardedOptions: resolveGuardedRequestOptions(params),
retryStage: params.retryStage,
retry: params.retry,
});

23
src/plugin-sdk/test-helpers/provider-http-mocks.ts
View File

@@ -2,6 +2,7 @@ import { afterEach, vi, type Mock } from "vitest";
import type {
fetchProviderDownloadResponse,
fetchProviderOperationResponse,
fetchWithTimeoutGuarded,
pollProviderOperationJson,
postMultipartRequest,
resolveProviderHttpRequestConfig,
@@ -13,6 +14,7 @@ type ResolveProviderHttpRequestConfigParams = Parameters<
>[0];
type PollProviderOperationJsonParams = Parameters<typeof pollProviderOperationJson>[0];
type PostMultipartRequestParams = Parameters<typeof postMultipartRequest>[0];
type FetchWithTimeoutGuardedParams = Parameters<typeof fetchWithTimeoutGuarded>;
type FetchProviderOperationResponseParams = Parameters<typeof fetchProviderOperationResponse>[0];
type FetchProviderDownloadResponseParams = Parameters<typeof fetchProviderDownloadResponse>[0];
type SanitizeConfiguredModelProviderRequestParams = Parameters<
@@ -33,6 +35,7 @@ interface ProviderHttpMocks {
postJsonRequestMock: AnyMock;
postMultipartRequestMock: AnyMock;
fetchWithTimeoutMock: AnyMock;
fetchWithTimeoutGuardedMock: AnyMock;
pollProviderOperationJsonMock: AnyMock;
assertOkOrThrowHttpErrorMock: Mock<(response: Response, label: string) => Promise<void>>;
assertOkOrThrowProviderErrorMock: Mock<(response: Response, label: string) => Promise<void>>;
@@ -51,6 +54,7 @@ const providerHttpMocks = vi.hoisted(() => ({
postJsonRequestMock: vi.fn(),
postMultipartRequestMock: vi.fn(),
fetchWithTimeoutMock: vi.fn(),
fetchWithTimeoutGuardedMock: vi.fn(),
fetchProviderOperationResponseMock: vi.fn(),
fetchProviderDownloadResponseMock: vi.fn(),
pollProviderOperationJsonMock: vi.fn(),
@@ -68,6 +72,23 @@ const providerHttpMocks = vi.hoisted(() => ({
})),
}));
providerHttpMocks.fetchWithTimeoutGuardedMock.mockImplementation(
async (...args: FetchWithTimeoutGuardedParams) => {
const [url, init, timeoutMs, fetchFn] = args;
const response = await providerHttpMocks.fetchWithTimeoutMock(
url,
init ?? {},
timeoutMs ?? 60_000,
fetchFn,
);
return {
response,
finalUrl: url,
release: async () => {},
};
},
);
providerHttpMocks.postMultipartRequestMock.mockImplementation(
async (params: PostMultipartRequestParams) => {
const response = await providerHttpMocks.fetchWithTimeoutMock(
@@ -173,6 +194,7 @@ vi.mock("openclaw/plugin-sdk/provider-http", () => ({
fetchProviderDownloadResponse: providerHttpMocks.fetchProviderDownloadResponseMock,
fetchProviderOperationResponse: providerHttpMocks.fetchProviderOperationResponseMock,
fetchWithTimeout: providerHttpMocks.fetchWithTimeoutMock,
fetchWithTimeoutGuarded: providerHttpMocks.fetchWithTimeoutGuardedMock,
pollProviderOperationJson: providerHttpMocks.pollProviderOperationJsonMock,
postJsonRequest: providerHttpMocks.postJsonRequestMock,
postMultipartRequest: providerHttpMocks.postMultipartRequestMock,
@@ -195,6 +217,7 @@ export function installProviderHttpMockCleanup(): void {
providerHttpMocks.postJsonRequestMock.mockReset();
providerHttpMocks.postMultipartRequestMock.mockClear();
providerHttpMocks.fetchWithTimeoutMock.mockReset();
providerHttpMocks.fetchWithTimeoutGuardedMock.mockClear();
providerHttpMocks.fetchProviderOperationResponseMock.mockClear();
providerHttpMocks.fetchProviderDownloadResponseMock.mockClear();
providerHttpMocks.pollProviderOperationJsonMock.mockClear();