diff --git a/extensions/openai/video-generation-provider.test.ts b/extensions/openai/video-generation-provider.test.ts index 3876e3b87d0c..f73e1f87bf79 100644 --- a/extensions/openai/video-generation-provider.test.ts +++ b/extensions/openai/video-generation-provider.test.ts @@ -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 | undefined] { + const call = fetchWithTimeoutGuardedMock.mock.calls[index] as + | [string, RequestInit | undefined, number, unknown, Record | undefined] + | undefined; + if (!call) { + throw new Error(`expected fetchWithTimeoutGuarded call ${index}`); + } + return call; +} + +function pollProviderOperationRequest(index = 0): Record { + const request = pollProviderOperationJsonMock.mock.calls[index]?.[0] as + | Record + | undefined; + if (!request) { + throw new Error(`expected pollProviderOperationJson call ${index}`); + } + return request; +} + function providerHttpConfigRequest(): Record { 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(); diff --git a/extensions/openai/video-generation-provider.ts b/extensions/openai/video-generation-provider.ts index e73b1ac98eca..ff5dc989cb00 100644 --- a/extensions/openai/video-generation-provider.ts +++ b/extensions/openai/video-generation-provider.ts @@ -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[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 { +async function pollOpenAIVideo( + params: { + videoId: string; + headers: Headers; + timeoutMs?: number; + baseUrl: string; + fetchFn: typeof fetch; + } & OpenAIVideoRequestPolicy, +): Promise { 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 { +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 { 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], diff --git a/src/media-understanding/shared.test.ts b/src/media-understanding/shared.test.ts index 283eacbe22f4..9ce32876cbe7 100644 --- a/src/media-understanding/shared.test.ts +++ b/src/media-understanding/shared.test.ts @@ -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() diff --git a/src/media-understanding/shared.ts b/src/media-understanding/shared.ts index cbdc2987c964..a24ea72f9901 100644 --- a/src/media-understanding/shared.ts +++ b/src/media-understanding/shared.ts @@ -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(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 { +export async function pollProviderOperationJson( + 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 { 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[4]>; +type GuardedProviderRequestOptions = NonNullable[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, }); diff --git a/src/plugin-sdk/test-helpers/provider-http-mocks.ts b/src/plugin-sdk/test-helpers/provider-http-mocks.ts index 5bc1864838ae..89504810613a 100644 --- a/src/plugin-sdk/test-helpers/provider-http-mocks.ts +++ b/src/plugin-sdk/test-helpers/provider-http-mocks.ts @@ -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[0]; type PostMultipartRequestParams = Parameters[0]; +type FetchWithTimeoutGuardedParams = Parameters; type FetchProviderOperationResponseParams = Parameters[0]; type FetchProviderDownloadResponseParams = Parameters[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>; assertOkOrThrowProviderErrorMock: Mock<(response: Response, label: string) => Promise>; @@ -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();