From bc0c3fcd634b66164fff359901bd66a93008fc59 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 7 Aug 2026 15:45:53 -0700 Subject: [PATCH] refactor(providers): shared generated-media download guard and binary-response adoption (#120351) * fix(agents): cancel rejected provider binary bodies * refactor(providers): share generated media downloads * refactor(providers): adopt payload patch stream wrapper * fix(providers): preserve generated media helper contracts --- .../amazon-bedrock/register.sync.runtime.ts | 45 ++++---- extensions/anthropic/stream-wrappers.ts | 78 ++++++------- extensions/azure-speech/tts.ts | 24 +--- .../byteplus/video-generation-provider.ts | 62 ++++------ extensions/deepinfra/cache-wrapper.ts | 26 ++--- extensions/fireworks/stream.ts | 21 ++-- extensions/github-copilot/stream.ts | 39 +++---- extensions/gradium/tts.ts | 26 ++--- extensions/kimi-coding/stream.ts | 26 +++-- extensions/meta/stream.ts | 23 ++-- .../minimax/music-generation-provider.ts | 109 +++++++----------- .../minimax/video-generation-provider.ts | 53 ++++----- extensions/ollama/src/stream-compat.ts | 25 ++-- extensions/openai/native-web-search.ts | 24 ++-- extensions/openai/tts.ts | 26 ++--- .../openai/video-generation-provider.ts | 78 +++++-------- extensions/opencode-go/stream.ts | 17 +-- .../runway/video-generation-provider.ts | 77 +++++-------- .../together/video-generation-provider.ts | 48 +++----- extensions/xai/tts.ts | 26 ++--- extensions/xai/video-generation-transport.ts | 60 ++++------ src/agents/provider-http-errors.test.ts | 13 +++ src/agents/provider-http-errors.ts | 16 ++- src/media-generation/provider-assets.test.ts | 60 ++++++++++ src/media-generation/provider-assets.ts | 88 ++++++++++++++ src/music-generation/provider-assets.test.ts | 31 ++++- src/music-generation/provider-assets.ts | 80 +++++++++---- src/plugin-sdk/media-generation-runtime.ts | 1 + 28 files changed, 617 insertions(+), 585 deletions(-) create mode 100644 src/media-generation/provider-assets.test.ts create mode 100644 src/media-generation/provider-assets.ts diff --git a/extensions/amazon-bedrock/register.sync.runtime.ts b/extensions/amazon-bedrock/register.sync.runtime.ts index 2a04b1b1d2a5..75b4b2ece4b1 100644 --- a/extensions/amazon-bedrock/register.sync.runtime.ts +++ b/extensions/amazon-bedrock/register.sync.runtime.ts @@ -19,7 +19,7 @@ import { resolveClaudeOpus5ModelIdentity, resolveClaudeSonnet5ModelIdentity, } from "openclaw/plugin-sdk/provider-model-shared"; -import { streamWithPayloadPatch } from "openclaw/plugin-sdk/provider-stream-shared"; +import { createPayloadPatchStreamWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; import { refreshAwsSharedConfigCacheForBedrock } from "./aws-credential-refresh.js"; import { supportsBedrockPromptCaching } from "./bedrock-options.js"; import { loadBedrockControlPlaneSdk, runBedrockControlPlaneRequest } from "./control-plane.js"; @@ -137,14 +137,13 @@ function createBedrockServiceTierWrapper( underlying: StreamFn, serviceTier: BedrockServiceTier, ): StreamFn { - return (model, context, options) => { - if (model.api !== "bedrock-converse-stream") { - return underlying(model, context, options); - } - return streamWithPayloadPatch(underlying, model, context, options, (payloadObj) => { - payloadObj.serviceTier ??= { type: serviceTier }; - }); - }; + return createPayloadPatchStreamWrapper( + underlying, + ({ payload }) => { + payload.serviceTier ??= { type: serviceTier }; + }, + { shouldPatch: ({ model }) => model.api === "bedrock-converse-stream" }, + ); } function createGuardrailWrapStreamFn( @@ -164,21 +163,19 @@ function createGuardrailWrapStreamFn( if (!inner) { return inner; } - return (model, context, options) => { - return streamWithPayloadPatch(inner, model, context, options, (payload) => { - const gc: Record = { - guardrailIdentifier: guardrailConfig.guardrailIdentifier, - guardrailVersion: guardrailConfig.guardrailVersion, - }; - if (guardrailConfig.streamProcessingMode) { - gc.streamProcessingMode = guardrailConfig.streamProcessingMode; - } - if (guardrailConfig.trace) { - gc.trace = guardrailConfig.trace; - } - payload.guardrailConfig = gc; - }); - }; + return createPayloadPatchStreamWrapper(inner, ({ payload }) => { + const gc: Record = { + guardrailIdentifier: guardrailConfig.guardrailIdentifier, + guardrailVersion: guardrailConfig.guardrailVersion, + }; + if (guardrailConfig.streamProcessingMode) { + gc.streamProcessingMode = guardrailConfig.streamProcessingMode; + } + if (guardrailConfig.trace) { + gc.trace = guardrailConfig.trace; + } + payload.guardrailConfig = gc; + }); }; } diff --git a/extensions/anthropic/stream-wrappers.ts b/extensions/anthropic/stream-wrappers.ts index a4c98a9f5f28..dcfa49bc3805 100644 --- a/extensions/anthropic/stream-wrappers.ts +++ b/extensions/anthropic/stream-wrappers.ts @@ -16,8 +16,8 @@ import { applyAnthropicPayloadPolicyToParams, composeProviderStreamWrappers, createAnthropicThinkingPrefillPayloadWrapper, + createPayloadPatchStreamWrapper, resolveAnthropicPayloadPolicy, - streamWithPayloadPatch, } from "openclaw/plugin-sdk/provider-stream-shared"; import { createSubsystemLogger } from "openclaw/plugin-sdk/runtime-env"; import { @@ -186,6 +186,10 @@ export function createAnthropicFastModeWrapper( enabled: DynamicFastMode, ): StreamFn { const underlying = baseStreamFn ?? streamSimple; + const fastPayloadWrapper = createPayloadPatchStreamWrapper(underlying, ({ payload }) => { + delete payload.service_tier; + payload.speed = "fast"; + }); return (model, context, options) => { const resolved = typeof enabled === "function" ? enabled() : enabled; if (resolved === undefined) { @@ -199,19 +203,10 @@ export function createAnthropicFastModeWrapper( ) { return underlying(model, context, options); } - return streamWithPayloadPatch( - underlying, - applyAnthropicFastModePricing(model), - context, - { - ...options, - headers: mergeAnthropicBetaHeader(options?.headers, [ANTHROPIC_FAST_MODE_BETA]), - }, - (payloadObj) => { - delete payloadObj.service_tier; - payloadObj.speed = "fast"; - }, - ); + return fastPayloadWrapper(applyAnthropicFastModePricing(model), context, { + ...options, + headers: mergeAnthropicBetaHeader(options?.headers, [ANTHROPIC_FAST_MODE_BETA]), + }); } return createAnthropicServiceTierWrapper(underlying, resolveAnthropicFastServiceTier(resolved))( model, @@ -226,31 +221,36 @@ export function createAnthropicServiceTierWrapper( baseStreamFn: StreamFn | undefined, serviceTier: AnthropicServiceTier, ): StreamFn { - const underlying = baseStreamFn ?? streamSimple; - return (model, context, options) => { - // Opus 5 and Sonnet 5 do not support Priority Tier; omit service_tier entirely. - if ( - isAnthropicOAuthApiKey(options?.apiKey) || - resolveClaudeOpus5ModelIdentity(model) !== undefined || - resolveClaudeSonnet5ModelIdentity(model) !== undefined - ) { - return underlying(model, context, options); - } - - const payloadPolicy = resolveAnthropicPayloadPolicy({ - provider: readStringValue(model.provider), - api: readStringValue(model.api), - baseUrl: readStringValue(model.baseUrl), - serviceTier, - }); - if (!payloadPolicy.allowsServiceTier) { - return underlying(model, context, options); - } - - return streamWithPayloadPatch(underlying, model, context, options, (payloadObj) => - applyAnthropicPayloadPolicyToParams(payloadObj, payloadPolicy, new Set()), - ); - }; + return createPayloadPatchStreamWrapper( + baseStreamFn, + ({ payload, model }) => { + const payloadPolicy = resolveAnthropicPayloadPolicy({ + provider: readStringValue(model.provider), + api: readStringValue(model.api), + baseUrl: readStringValue(model.baseUrl), + serviceTier, + }); + applyAnthropicPayloadPolicyToParams(payload, payloadPolicy, new Set()); + }, + { + shouldPatch: ({ model, options }) => { + // Opus 5 and Sonnet 5 do not support Priority Tier; omit service_tier entirely. + if ( + isAnthropicOAuthApiKey(options?.apiKey) || + resolveClaudeOpus5ModelIdentity(model) !== undefined || + resolveClaudeSonnet5ModelIdentity(model) !== undefined + ) { + return false; + } + return resolveAnthropicPayloadPolicy({ + provider: readStringValue(model.provider), + api: readStringValue(model.api), + baseUrl: readStringValue(model.baseUrl), + serviceTier, + }).allowsServiceTier; + }, + }, + ); } /** Wrap a stream function to strip trailing assistant prefill before thinking requests. */ diff --git a/extensions/azure-speech/tts.ts b/extensions/azure-speech/tts.ts index 739092ef2679..91da9e046176 100644 --- a/extensions/azure-speech/tts.ts +++ b/extensions/azure-speech/tts.ts @@ -4,10 +4,9 @@ */ import { assertOkOrThrowProviderError, - assertProviderBinaryResponseContent, + readProviderBinaryResponse, readProviderJsonResponse, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import type { SpeechVoiceOption } from "openclaw/plugin-sdk/speech-core"; import { asObject, trimToUndefined } from "openclaw/plugin-sdk/speech-core"; import { @@ -222,26 +221,13 @@ export async function azureSpeechTTS(params: { try { await assertOkOrThrowProviderError(response, "Azure Speech TTS API error"); - try { - assertProviderBinaryResponseContent(response, "Azure Speech TTS API error", "audio"); - } catch (error) { - // A debug-capture clone can keep the tee open, so waiting for cancel would hang - // before the rejected response and its dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - const audio = await readResponseWithLimit( - response, - params.maxBytes ?? DEFAULT_AZURE_SPEECH_MAX_BYTES, - { + return Buffer.from( + await readProviderBinaryResponse(response, "Azure Speech TTS API error", "audio", { + maxBytes: params.maxBytes ?? DEFAULT_AZURE_SPEECH_MAX_BYTES, onOverflow: ({ maxBytes }) => new Error(`Azure Speech TTS audio response exceeds ${maxBytes} bytes`), - }, + }), ); - if (audio.byteLength === 0) { - throw new Error("Azure Speech TTS API error: malformed audio response"); - } - return audio; } finally { await release(); } diff --git a/extensions/byteplus/video-generation-provider.ts b/extensions/byteplus/video-generation-provider.ts index 1222b1e42e49..fac46c64e308 100644 --- a/extensions/byteplus/video-generation-provider.ts +++ b/extensions/byteplus/video-generation-provider.ts @@ -2,13 +2,14 @@ * BytePlus Seedance video generation provider implementation. */ import { toImageDataUrl } from "openclaw/plugin-sdk/image-generation"; -import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; -import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; +import { + downloadGeneratedVideoAsset, + resolveGeneratedMediaMaxBytes, +} from "openclaw/plugin-sdk/media-generation-runtime"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { assertOkOrThrowHttpError, - assertProviderBinaryResponseContent, createProviderOperationDeadline, createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, @@ -19,7 +20,6 @@ import { resolveProviderHttpRequestConfig, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { asSafeIntegerInRange, isRecord, @@ -181,49 +181,27 @@ async function downloadBytePlusVideo(params: { fetchFn: typeof fetch; maxBytes: number; }): Promise { - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, - label: "BytePlus generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, - }); - const response = await fetchProviderDownloadResponse({ + return await downloadGeneratedVideoAsset({ url: params.url, - init: { method: "GET" }, - deadline, + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, fetchFn: params.fetchFn, provider: "byteplus", + label: "BytePlus generated video download", requestFailedMessage: "BytePlus generated video download failed", + maxBytes: params.maxBytes, + validateBinaryResponse: true, + fetchResponse: async ({ deadline }) => ({ + response: await fetchProviderDownloadResponse({ + url: params.url, + init: { method: "GET" }, + deadline, + fetchFn: params.fetchFn, + provider: "byteplus", + requestFailedMessage: "BytePlus generated video download failed", + }), + }), }); - try { - assertProviderBinaryResponseContent(response, "BytePlus generated video download", "video"); - } catch (error) { - // A rejected binary response still owns a live socket until its unread body is canceled. - // A debug-capture clone can keep the tee open, so waiting for cancel would hang - // before the rejected response and its dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - 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`), - }); - if (buffer.byteLength === 0) { - throw new Error("BytePlus generated video download: malformed video response"); - } - return { - buffer, - mimeType, - fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - }; } /** Builds the BytePlus video generation provider registered by the plugin. */ diff --git a/extensions/deepinfra/cache-wrapper.ts b/extensions/deepinfra/cache-wrapper.ts index 8322512e361e..a026355b53f7 100644 --- a/extensions/deepinfra/cache-wrapper.ts +++ b/extensions/deepinfra/cache-wrapper.ts @@ -1,26 +1,22 @@ // Deepinfra plugin module implements cache wrapper behavior. +import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; import { applyAnthropicEphemeralCacheControlMarkers, - streamWithPayloadPatch, -} from "openclaw/plugin-sdk/provider-stream"; - -// StreamFn isn't re-exported via the plugin SDK; derive it from a helper that -// accepts it so we stay on the SDK boundary. -type StreamFn = Parameters[0]; + createPayloadPatchStreamWrapper, +} from "openclaw/plugin-sdk/provider-stream-shared"; // Inject Anthropic ephemeral cache_control markers for anthropic/* models on // DeepInfra. The OpenRouter equivalent short-circuits on a provider/endpoint // check, so DeepInfra advertises isCacheTtlEligible but the payload patch // never fires. Gating on the model id instead fixes that. export function createDeepInfraAnthropicCacheWrapper(baseStreamFn: StreamFn): StreamFn { - return ((model, context, options) => { - const modelIdRaw = (model as { id?: unknown }).id; - const modelId = typeof modelIdRaw === "string" ? modelIdRaw.toLowerCase() : ""; - if (!modelId.startsWith("anthropic/")) { - return baseStreamFn(model, context, options); - } - return streamWithPayloadPatch(baseStreamFn, model, context, options, (payload) => { + return createPayloadPatchStreamWrapper( + baseStreamFn, + ({ payload }) => { applyAnthropicEphemeralCacheControlMarkers(payload); - }); - }) as StreamFn; + }, + { + shouldPatch: ({ model }) => model.id.toLowerCase().startsWith("anthropic/"), + }, + ); } diff --git a/extensions/fireworks/stream.ts b/extensions/fireworks/stream.ts index 31ffa4ef064e..abe7453ff0eb 100644 --- a/extensions/fireworks/stream.ts +++ b/extensions/fireworks/stream.ts @@ -1,9 +1,8 @@ // Fireworks plugin module implements stream behavior. import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; -import { streamSimple } from "openclaw/plugin-sdk/llm"; import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; import { normalizeProviderId } from "openclaw/plugin-sdk/provider-model-shared"; -import { streamWithPayloadPatch } from "openclaw/plugin-sdk/provider-stream-shared"; +import { createPayloadPatchStreamWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; import { isFireworksKimiModelId } from "./model-id.js"; function isFireworksProviderId(providerId: string): boolean { @@ -12,16 +11,14 @@ function isFireworksProviderId(providerId: string): boolean { } function createFireworksKimiThinkingDisabledWrapper(baseStreamFn: StreamFn | undefined): StreamFn { - const underlying = baseStreamFn ?? streamSimple; - return (model, context, options) => - streamWithPayloadPatch(underlying, model, context, options, (payloadObj) => { - // Fireworks Kimi can emit chain-of-thought in visible `content` unless - // the Anthropic-style thinking toggle is explicitly disabled. - payloadObj.thinking = { type: "disabled" }; - delete payloadObj.reasoning; - delete payloadObj.reasoning_effort; - delete payloadObj.reasoningEffort; - }); + return createPayloadPatchStreamWrapper(baseStreamFn, ({ payload }) => { + // Fireworks Kimi can emit chain-of-thought in visible `content` unless + // the Anthropic-style thinking toggle is explicitly disabled. + payload.thinking = { type: "disabled" }; + delete payload.reasoning; + delete payload.reasoning_effort; + delete payload.reasoningEffort; + }); } export function wrapFireworksProviderStream( diff --git a/extensions/github-copilot/stream.ts b/extensions/github-copilot/stream.ts index 2cca85c2f539..63286a356da1 100644 --- a/extensions/github-copilot/stream.ts +++ b/extensions/github-copilot/stream.ts @@ -4,7 +4,7 @@ import type { Context } from "openclaw/plugin-sdk/llm"; import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; import { applyAnthropicEphemeralCacheControlMarkers, - streamWithPayloadPatch, + createPayloadPatchStreamWrapper, } from "openclaw/plugin-sdk/provider-stream-shared"; import { sanitizeCopilotReplayResponsePayload } from "./connection-bound-ids.js"; import { stripCopilotAssistantThinkingMessages } from "./replay-policy.js"; @@ -192,32 +192,29 @@ export function wrapCopilotAnthropicStream( return undefined; } const underlying = baseStreamFn; + const payloadWrapper = createPayloadPatchStreamWrapper(underlying, ({ payload }) => + patchCopilotAnthropicPayload(payload), + ); return (model, context, options) => { if (model.provider !== "github-copilot" || model.api !== "anthropic-messages") { return underlying(model, context, options); } const originalOnPayload = options?.onPayload; - return streamWithPayloadPatch( - underlying, - model, - context, - { - ...options, - headers: buildCopilotRequestHeaders(context, options?.headers), - onPayload: (payload, payloadModel) => - patchOnPayloadResult( - originalOnPayload?.(payload, payloadModel), - (replacement) => { - if (replacement && typeof replacement === "object") { - patchCopilotAnthropicPayload(replacement as Record); - } - }, - payload, - ), - }, - patchCopilotAnthropicPayload, - ); + return payloadWrapper(model, context, { + ...options, + headers: buildCopilotRequestHeaders(context, options?.headers), + onPayload: (payload, payloadModel) => + patchOnPayloadResult( + originalOnPayload?.(payload, payloadModel), + (replacement) => { + if (replacement && typeof replacement === "object") { + patchCopilotAnthropicPayload(replacement as Record); + } + }, + payload, + ), + }); }; } diff --git a/extensions/gradium/tts.ts b/extensions/gradium/tts.ts index 42f7201620a3..8148fa2728a3 100644 --- a/extensions/gradium/tts.ts +++ b/extensions/gradium/tts.ts @@ -1,9 +1,8 @@ // Gradium plugin module implements tts behavior. import { assertOkOrThrowProviderError, - assertProviderBinaryResponseContent, + readProviderBinaryResponse, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { fetchWithSsrFGuard } from "openclaw/plugin-sdk/ssrf-runtime"; import { GRADIUM_API_HOSTNAME, normalizeGradiumBaseUrl } from "./shared.js"; @@ -57,22 +56,13 @@ export async function gradiumTTS(params: { try { await assertOkOrThrowProviderError(response, "Gradium API error"); - try { - assertProviderBinaryResponseContent(response, "Gradium API error", "audio"); - } catch (error) { - // A debug-capture clone can keep the tee open, so waiting for cancel would hang - // before the rejected response and its dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - const audio = await readResponseWithLimit(response, maxBytes, { - onOverflow: ({ maxBytes: maxBytesLocal }) => - new Error(`Gradium TTS audio response exceeds ${maxBytesLocal} bytes`), - }); - if (audio.byteLength === 0) { - throw new Error("Gradium API error: malformed audio response"); - } - return audio; + return Buffer.from( + await readProviderBinaryResponse(response, "Gradium API error", "audio", { + maxBytes, + onOverflow: ({ maxBytes: maxBytesLocal }) => + new Error(`Gradium TTS audio response exceeds ${maxBytesLocal} bytes`), + }), + ); } finally { await release(); } diff --git a/extensions/kimi-coding/stream.ts b/extensions/kimi-coding/stream.ts index 274f29af362b..4a139b989393 100644 --- a/extensions/kimi-coding/stream.ts +++ b/extensions/kimi-coding/stream.ts @@ -7,8 +7,8 @@ import { } from "openclaw/plugin-sdk/llm"; import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; import { + createPayloadPatchStreamWrapper, normalizeOpenAICompatibleReasoningReplay, - streamWithPayloadPatch, } from "openclaw/plugin-sdk/provider-stream-shared"; import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; import { isKimiK3ModelId } from "./provider-policy-api.js"; @@ -372,15 +372,9 @@ function createKimiThinkingWrapper( k3ThinkingConfig: { type: "disabled" } | { type: "adaptive"; effort: KimiK3ThinkingEffort }, ): StreamFn { const underlying = baseStreamFn ?? streamSimple; - return (model, context, options) => { - const runtimeModel = - model.api === "anthropic-messages" && isKimiK3ModelId(model.id) - ? { - ...model, - compat: { ...model.compat, allowEmptySignature: true }, - } - : model; - return streamWithPayloadPatch(underlying, runtimeModel, context, options, (payloadObj) => { + const payloadWrapper = createPayloadPatchStreamWrapper( + underlying, + ({ payload: payloadObj, model }) => { if (model.api === "anthropic-messages" && isKimiK3ModelId(model.id)) { const outputConfig = payloadObj.output_config; if (k3ThinkingConfig.type === "disabled") { @@ -428,7 +422,17 @@ function createKimiThinkingWrapper( delete payloadObj.reasoning_effort; delete payloadObj.reasoningEffort; stripAnthropicCacheControlMarkers(payloadObj); - }); + }, + ); + return (model, context, options) => { + const runtimeModel = + model.api === "anthropic-messages" && isKimiK3ModelId(model.id) + ? { + ...model, + compat: { ...model.compat, allowEmptySignature: true }, + } + : model; + return payloadWrapper(runtimeModel, context, options); }; } diff --git a/extensions/meta/stream.ts b/extensions/meta/stream.ts index dcab032ee53f..962ce41d06a8 100644 --- a/extensions/meta/stream.ts +++ b/extensions/meta/stream.ts @@ -1,8 +1,7 @@ // Meta plugin module implements stream behavior. import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; -import { streamSimple } from "openclaw/plugin-sdk/llm"; import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; -import { streamWithPayloadPatch } from "openclaw/plugin-sdk/provider-stream-shared"; +import { createPayloadPatchStreamWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; const META_REASONING_ENCRYPTED_CONTENT_INCLUDE = "reasoning.encrypted_content"; @@ -19,17 +18,15 @@ function ensureMetaResponsesReplayFields(payloadObj: Record): v } function createMetaResponsesWrapper(baseStreamFn: StreamFn | undefined): StreamFn { - const underlying = baseStreamFn ?? streamSimple; - return (model, context, options) => - streamWithPayloadPatch(underlying, model, context, options, (payloadObj) => { - if (model.provider !== "meta" || model.api !== "openai-responses") { - return; - } - if (!model.reasoning) { - return; - } - ensureMetaResponsesReplayFields(payloadObj); - }); + return createPayloadPatchStreamWrapper(baseStreamFn, ({ payload, model }) => { + if (model.provider !== "meta" || model.api !== "openai-responses") { + return; + } + if (!model.reasoning) { + return; + } + ensureMetaResponsesReplayFields(payload); + }); } export function wrapMetaProviderStream(ctx: ProviderWrapStreamFnContext): StreamFn | undefined { diff --git a/extensions/minimax/music-generation-provider.ts b/extensions/minimax/music-generation-provider.ts index 243d7eb564b5..77e952b3d844 100644 --- a/extensions/minimax/music-generation-provider.ts +++ b/extensions/minimax/music-generation-provider.ts @@ -1,17 +1,16 @@ // Minimax provider module implements model/runtime integration. import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; -import type { - GeneratedMusicAsset, - MusicGenerationProvider, +import { + downloadGeneratedMusicAsset, + type GeneratedMusicAsset, + type MusicGenerationProvider, } from "openclaw/plugin-sdk/music-generation"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { assertOkOrThrowHttpError, - assertProviderBinaryResponseContent, createProviderOperationDeadline, - createProviderOperationTimeoutResolver, executeProviderOperationWithRetry, fetchWithTimeoutGuarded, postJsonRequest, @@ -90,72 +89,46 @@ async function downloadTrackFromUrl(params: { maxBytes: number; policy: MinimaxRequestPolicy; }): Promise { - const deadline = createProviderOperationDeadline({ + return await downloadGeneratedMusicAsset({ + candidate: { url: params.url }, 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", - operation: async () => { - const guardedResult = await fetchWithTimeoutGuarded( - params.url, - { method: "GET" }, - timeoutMs(), - params.fetchFn, - resolveMinimaxGuardedRequestOptions(params.policy), - ); - try { - await assertOkOrThrowHttpError( - guardedResult.response, - "MiniMax generated music download failed", - ); - } catch (error) { - await guardedResult.release(); - throw error; - } - return guardedResult; + fetchFn: params.fetchFn, + provider: "MiniMax", + requestFailedMessage: "MiniMax generated music download failed", + maxBytes: params.maxBytes, + validateBinaryResponse: true, + includeSourceUrl: false, + fetchResponse: async ({ timeoutMs }) => { + const result = await executeProviderOperationWithRetry({ + provider: "minimax", + stage: "download", + operation: async () => { + const guardedResult = await fetchWithTimeoutGuarded( + params.url, + { method: "GET" }, + timeoutMs(), + params.fetchFn, + resolveMinimaxGuardedRequestOptions(params.policy), + ); + try { + await assertOkOrThrowHttpError( + guardedResult.response, + "MiniMax generated music download failed", + ); + } catch (error) { + await guardedResult.release(); + throw error; + } + return guardedResult; + }, + }); + return { + ...result, + mimeType: + normalizeOptionalString(result.response.headers.get("content-type")) ?? "audio/mpeg", + }; }, }); - try { - try { - assertProviderBinaryResponseContent( - result.response, - "MiniMax generated music download", - "audio", - ); - } catch (error) { - // Release the unread response before its guarded dispatcher is closed. - await result.response.body?.cancel().catch(() => undefined); - throw error; - } - const mimeType = - normalizeOptionalString(result.response.headers.get("content-type")) ?? "audio/mpeg"; - const ext = extensionForMime(mimeType)?.replace(/^\./u, "") || "mp3"; - const 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`), - }); - if (buffer.byteLength === 0) { - throw new Error("MiniMax generated music download: malformed audio response"); - } - return { - buffer, - mimeType, - fileName: `track-1.${ext}`, - }; - } finally { - await result.release(); - } } function resolveBodyReadTimeoutMs(deadline: ProviderOperationDeadline): number { diff --git a/extensions/minimax/video-generation-provider.ts b/extensions/minimax/video-generation-provider.ts index 428c7e3739bf..262f0ed2af80 100644 --- a/extensions/minimax/video-generation-provider.ts +++ b/extensions/minimax/video-generation-provider.ts @@ -1,6 +1,9 @@ // Minimax provider module implements model/runtime integration. import { toImageDataUrl } from "openclaw/plugin-sdk/image-generation"; -import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; +import { + downloadGeneratedVideoAsset, + resolveGeneratedMediaMaxBytes, +} from "openclaw/plugin-sdk/media-generation-runtime"; import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; @@ -258,42 +261,26 @@ async function downloadVideoFromUrl(params: { maxBytes: number; policy: MinimaxRequestPolicy; }): Promise { - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, - label: "MiniMax generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, - }); - const { response, release } = await fetchMinimaxResponse({ - stage: "download", + return await downloadGeneratedVideoAsset({ url: params.url, - init: { method: "GET" }, - timeoutMs, + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, fetchFn: params.fetchFn, + provider: "minimax", + label: "MiniMax generated video download", requestFailedMessage: "MiniMax generated video download failed", - policy: params.policy, + maxBytes: params.maxBytes, + fetchResponse: async ({ timeoutMs }) => + await fetchMinimaxResponse({ + stage: "download", + url: params.url, + init: { method: "GET" }, + timeoutMs, + fetchFn: params.fetchFn, + requestFailedMessage: "MiniMax generated video download failed", + policy: params.policy, + }), }); - 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`), - }); - return { - buffer, - mimeType, - fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - }; - } finally { - await release(); - } } async function downloadVideoFromFileId(params: { diff --git a/extensions/ollama/src/stream-compat.ts b/extensions/ollama/src/stream-compat.ts index 45b27ec7984b..4b2493749d4b 100644 --- a/extensions/ollama/src/stream-compat.ts +++ b/extensions/ollama/src/stream-compat.ts @@ -1,5 +1,4 @@ import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; -import { streamSimple } from "openclaw/plugin-sdk/llm"; import type { OpenClawConfig, ProviderRuntimeModel, @@ -11,8 +10,8 @@ import { } from "openclaw/plugin-sdk/provider-model-shared"; import { createMoonshotThinkingWrapper, + createPayloadPatchStreamWrapper, resolveMoonshotThinkingType, - streamWithPayloadPatch, } from "openclaw/plugin-sdk/provider-stream-shared"; import { isLoopbackHost } from "openclaw/plugin-sdk/ssrf-runtime"; import { shouldWrapOllamaCompatMoonshotThinking } from "./model-behavior.js"; @@ -98,25 +97,21 @@ export function shouldInjectOllamaCompatNumCtx(params: { } export function wrapOllamaCompatNumCtx(baseFn: StreamFn | undefined, numCtx: number): StreamFn { - const streamFn = baseFn ?? streamSimple; - return (model, context, options) => - streamWithPayloadPatch(streamFn, model, context, options, (payloadRecord) => { - if (!payloadRecord.options || typeof payloadRecord.options !== "object") { - payloadRecord.options = {}; - } - (payloadRecord.options as Record).num_ctx = numCtx; - }); + return createPayloadPatchStreamWrapper(baseFn, ({ payload }) => { + if (!payload.options || typeof payload.options !== "object") { + payload.options = {}; + } + (payload.options as Record).num_ctx = numCtx; + }); } function createOllamaThinkingWrapper( baseFn: StreamFn | undefined, think: OllamaThinkValue, ): StreamFn { - const streamFn = baseFn ?? streamSimple; - return (model, context, options) => - streamWithPayloadPatch(streamFn, model, context, options, (payloadRecord) => { - payloadRecord.think = think; - }); + return createPayloadPatchStreamWrapper(baseFn, ({ payload }) => { + payload.think = think; + }); } function resolveOllamaThinkValue(thinkingLevel: unknown): OllamaThinkValue | undefined { diff --git a/extensions/openai/native-web-search.ts b/extensions/openai/native-web-search.ts index e1058d04caae..b002c25e8fad 100644 --- a/extensions/openai/native-web-search.ts +++ b/extensions/openai/native-web-search.ts @@ -1,9 +1,8 @@ // Openai plugin module implements native web search behavior. import type { StreamFn } from "openclaw/plugin-sdk/agent-core"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { streamSimple } from "openclaw/plugin-sdk/llm"; import { normalizeProviderId } from "openclaw/plugin-sdk/provider-model-shared"; -import { streamWithPayloadPatch } from "openclaw/plugin-sdk/provider-stream-shared"; +import { createPayloadPatchStreamWrapper } from "openclaw/plugin-sdk/provider-stream-shared"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { isOpenAIApiBaseUrl } from "./base-url.js"; @@ -91,16 +90,15 @@ export function createOpenAINativeWebSearchWrapper( nativeWebSearchAllowedByToolPolicy?: boolean; }, ): StreamFn { - const underlying = baseStreamFn ?? streamSimple; - return (model, context, options) => { - if (!shouldEnableOpenAINativeWebSearch({ config: params.config, model })) { - return underlying(model, context, options); - } - if (params.nativeWebSearchAllowedByToolPolicy === false) { - return underlying(model, context, options); - } - return streamWithPayloadPatch(underlying, model, context, options, (payload) => { + return createPayloadPatchStreamWrapper( + baseStreamFn, + ({ payload }) => { patchOpenAINativeWebSearchPayload(payload); - }); - }; + }, + { + shouldPatch: ({ model }) => + params.nativeWebSearchAllowedByToolPolicy !== false && + shouldEnableOpenAINativeWebSearch({ config: params.config, model }), + }, + ); } diff --git a/extensions/openai/tts.ts b/extensions/openai/tts.ts index d2e0248462f5..f4ef94faed02 100644 --- a/extensions/openai/tts.ts +++ b/extensions/openai/tts.ts @@ -1,14 +1,13 @@ // Openai plugin module implements tts behavior. import { assertOkOrThrowProviderError, - assertProviderBinaryResponseContent, + readProviderBinaryResponse, resolveProviderRequestHeaders, } from "openclaw/plugin-sdk/provider-http"; import { captureHttpExchange, isDebugProxyGlobalFetchPatchInstalled, } from "openclaw/plugin-sdk/proxy-capture"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { fetchWithSsrFGuard, ssrfPolicyFromHttpBaseUrlAllowedHostname, @@ -188,22 +187,13 @@ export async function openaiTTS(params: { await assertOkOrThrowProviderError(response, "OpenAI TTS API error"); - try { - assertProviderBinaryResponseContent(response, "OpenAI TTS API error", "audio"); - } catch (error) { - // Capture may clone and tee this response; awaiting cancellation would - // deadlock before the rejected response and dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - const audio = await readResponseWithLimit(response, maxBytes, { - onOverflow: ({ maxBytes: maxBytesLocal }) => - new Error(`OpenAI TTS audio response exceeds ${maxBytesLocal} bytes`), - }); - if (audio.byteLength === 0) { - throw new Error("OpenAI TTS API error: malformed audio response"); - } - return audio; + return Buffer.from( + await readProviderBinaryResponse(response, "OpenAI TTS API error", "audio", { + maxBytes, + onOverflow: ({ maxBytes: maxBytesLocal }) => + new Error(`OpenAI TTS audio response exceeds ${maxBytesLocal} bytes`), + }), + ); } finally { await release(); } diff --git a/extensions/openai/video-generation-provider.ts b/extensions/openai/video-generation-provider.ts index 39f014a1db40..90c2558a08da 100644 --- a/extensions/openai/video-generation-provider.ts +++ b/extensions/openai/video-generation-provider.ts @@ -1,11 +1,13 @@ // Openai provider module implements model/runtime integration. -import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; +import { + downloadGeneratedVideoAsset, + resolveGeneratedMediaMaxBytes, +} from "openclaw/plugin-sdk/media-generation-runtime"; import { extensionForMime, type MediaKind } from "openclaw/plugin-sdk/media-mime"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { assertOkOrThrowHttpError, - assertProviderBinaryResponseContent, createProviderOperationDeadline, createProviderOperationTimeoutResolver, executeProviderOperationWithRetry, @@ -19,7 +21,6 @@ import { sanitizeConfiguredModelProviderRequest, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { GeneratedVideoAsset, @@ -238,57 +239,32 @@ async function downloadOpenAIVideo( ): Promise { const url = new URL(`${params.baseUrl}/videos/${params.videoId}/content`); url.searchParams.set("variant", "video"); - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, - label: "OpenAI generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, - }); - const { response, release } = await fetchOpenAIVideoDownload({ + return await downloadGeneratedVideoAsset({ url: url.toString(), - init: { - method: "GET", - headers: new Headers({ - ...Object.fromEntries(params.headers.entries()), - Accept: "application/binary", - }), - }, - deadline, + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, fetchFn: params.fetchFn, - allowPrivateNetwork: params.allowPrivateNetwork, - dispatcherPolicy: params.dispatcherPolicy, + provider: "openai", + label: "OpenAI generated video download", + requestFailedMessage: "OpenAI video download failed", + maxBytes: params.maxBytes, + validateBinaryResponse: true, + fetchResponse: async ({ deadline }) => + await fetchOpenAIVideoDownload({ + url: url.toString(), + init: { + method: "GET", + headers: new Headers({ + ...Object.fromEntries(params.headers.entries()), + Accept: "application/binary", + }), + }, + deadline, + fetchFn: params.fetchFn, + allowPrivateNetwork: params.allowPrivateNetwork, + dispatcherPolicy: params.dispatcherPolicy, + }), }); - try { - try { - assertProviderBinaryResponseContent(response, deadline.label, "video"); - } catch (error) { - // Capture can tee this unread body; awaiting its cancellation deadlocks before release. - void response.body?.cancel().catch(() => undefined); - throw error; - } - 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`), - }); - if (buffer.byteLength === 0) { - throw new Error(`${deadline.label}: malformed video response`); - } - return { - buffer, - mimeType, - fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - }; - } finally { - await release(); - } } export function buildOpenAIVideoGenerationProvider(): VideoGenerationProvider { diff --git a/extensions/opencode-go/stream.ts b/extensions/opencode-go/stream.ts index 8afa0e8ad9ee..12fdb1dfceae 100644 --- a/extensions/opencode-go/stream.ts +++ b/extensions/opencode-go/stream.ts @@ -2,7 +2,7 @@ import type { ProviderWrapStreamFnContext } from "openclaw/plugin-sdk/plugin-entry"; import { createDeepSeekV4OpenAICompatibleThinkingWrapper, - streamWithPayloadPatch, + createPayloadPatchStreamWrapper, } from "openclaw/plugin-sdk/provider-stream-shared"; import { isOpencodeGoKimiNoReasoningModelId } from "./provider-catalog.js"; import { isOpencodeGoDeepSeekV4ModelId } from "./provider-policy-api.js"; @@ -35,13 +35,14 @@ function createOpencodeGoKimiNoReasoningWrapper( if (!baseStreamFn) { return undefined; } - const underlying = baseStreamFn; - return (model, context, options) => { - if (model.provider !== "opencode-go" || !isOpencodeGoKimiNoReasoningModelId(model.id)) { - return underlying(model, context, options); - } - return streamWithPayloadPatch(underlying, model, context, options, stripReasoningParams); - }; + return createPayloadPatchStreamWrapper( + baseStreamFn, + ({ payload }) => stripReasoningParams(payload), + { + shouldPatch: ({ model }) => + model.provider === "opencode-go" && isOpencodeGoKimiNoReasoningModelId(model.id), + }, + ); } export function createOpencodeGoWrapper( diff --git a/extensions/runway/video-generation-provider.ts b/extensions/runway/video-generation-provider.ts index fb71db8ec317..e93a932bc05e 100644 --- a/extensions/runway/video-generation-provider.ts +++ b/extensions/runway/video-generation-provider.ts @@ -1,11 +1,12 @@ // Runway provider module implements model/runtime integration. -import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; -import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; +import { + downloadGeneratedVideoAsset, + resolveGeneratedMediaMaxBytes, +} from "openclaw/plugin-sdk/media-generation-runtime"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { assertOkOrThrowHttpError, - assertProviderBinaryResponseContent, createProviderOperationDeadline, createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, @@ -16,7 +17,6 @@ import { resolveProviderHttpRequestConfig, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { isRecord, normalizeLowercaseStringOrEmpty, @@ -300,50 +300,31 @@ async function downloadRunwayVideos(params: { }): Promise { const videos: GeneratedVideoAsset[] = []; for (const [index, url] of params.urls.entries()) { - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, - label: "Runway generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, - }); - const response = await fetchProviderDownloadResponse({ - url, - init: { method: "GET" }, - deadline, - fetchFn: params.fetchFn, - provider: "runway", - requestFailedMessage: "Runway generated video download failed", - }); - try { - assertProviderBinaryResponseContent(response, "Runway generated video download", "video"); - } catch (error) { - // A rejected binary response still owns a live socket until its unread body is canceled. - // A debug-capture clone can keep the tee open, so waiting for cancel would hang - // before the rejected response and its dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - 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`), - }); - if (buffer.byteLength === 0) { - throw new Error("Runway generated video download: malformed video response"); - } - videos.push({ - buffer, - mimeType, - fileName: `video-${index + 1}.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - metadata: { sourceUrl: url }, - }); + videos.push( + await downloadGeneratedVideoAsset({ + url, + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, + fetchFn: params.fetchFn, + provider: "runway", + label: "Runway generated video download", + requestFailedMessage: "Runway generated video download failed", + index, + maxBytes: params.maxBytes, + validateBinaryResponse: true, + metadata: { sourceUrl: url }, + fetchResponse: async ({ deadline }) => ({ + response: await fetchProviderDownloadResponse({ + url, + init: { method: "GET" }, + deadline, + fetchFn: params.fetchFn, + provider: "runway", + requestFailedMessage: "Runway generated video download failed", + }), + }), + }), + ); } return videos; } diff --git a/extensions/together/video-generation-provider.ts b/extensions/together/video-generation-provider.ts index 35c416ccceb5..eac9c30cf491 100644 --- a/extensions/together/video-generation-provider.ts +++ b/extensions/together/video-generation-provider.ts @@ -1,7 +1,9 @@ // Together provider module implements model/runtime integration. import { toImageDataUrl } from "openclaw/plugin-sdk/image-generation"; -import { resolveGeneratedMediaMaxBytes } from "openclaw/plugin-sdk/media-generation-runtime"; -import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; +import { + downloadGeneratedVideoAsset, + resolveGeneratedMediaMaxBytes, +} from "openclaw/plugin-sdk/media-generation-runtime"; import { isProviderApiKeyConfigured } from "openclaw/plugin-sdk/provider-auth"; import { resolveApiKeyForProvider } from "openclaw/plugin-sdk/provider-auth-runtime"; import { @@ -16,7 +18,6 @@ import { resolveProviderHttpRequestConfig, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { asSafeIntegerInRange, normalizeOptionalString, @@ -136,37 +137,26 @@ async function downloadTogetherVideo(params: { fetchFn: typeof fetch; maxBytes: number; }): Promise { - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, - label: "Together generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS, - }); - const response = await fetchProviderDownloadResponse({ + return await downloadGeneratedVideoAsset({ url: params.url, - init: { method: "GET" }, - deadline, + timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS, + defaultTimeoutMs: DEFAULT_TIMEOUT_MS, fetchFn: params.fetchFn, provider: "together", + label: "Together generated video download", requestFailedMessage: "Together generated video download failed", + maxBytes: params.maxBytes, + fetchResponse: async ({ deadline }) => ({ + response: await fetchProviderDownloadResponse({ + url: params.url, + init: { method: "GET" }, + 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`), - }); - return { - buffer, - mimeType, - fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - }; } export function buildTogetherVideoGenerationProvider(): VideoGenerationProvider { diff --git a/extensions/xai/tts.ts b/extensions/xai/tts.ts index 8f984178bd39..6f59f83453c7 100644 --- a/extensions/xai/tts.ts +++ b/extensions/xai/tts.ts @@ -2,11 +2,10 @@ import { canonicalizeBase64 } from "openclaw/plugin-sdk/media-runtime"; import { assertOkOrThrowProviderError, - assertProviderBinaryResponseContent, postJsonRequest, + readProviderBinaryResponse, readProviderJsonResponse, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; import { asObject, trimToUndefined, type SpeechVoiceOption } from "openclaw/plugin-sdk/speech"; import { fetchWithSsrFGuard, @@ -482,22 +481,13 @@ export async function xaiTTS(params: { try { await assertOkOrThrowProviderError(response, "xAI TTS API error"); - try { - assertProviderBinaryResponseContent(response, "xAI TTS API error", "audio"); - } catch (error) { - // A debug-capture clone can keep the tee open, so waiting for cancel would hang - // before the rejected response and its dispatcher can be released. - void response.body?.cancel().catch(() => undefined); - throw error; - } - const audio = await readResponseWithLimit(response, maxBytes, { - onOverflow: ({ maxBytes: maxBytesLocal }) => - new Error(`xAI TTS audio response exceeds ${maxBytesLocal} bytes`), - }); - if (audio.byteLength === 0) { - throw new Error("xAI TTS API error: malformed audio response"); - } - return audio; + return Buffer.from( + await readProviderBinaryResponse(response, "xAI TTS API error", "audio", { + maxBytes, + onOverflow: ({ maxBytes: maxBytesLocal }) => + new Error(`xAI TTS audio response exceeds ${maxBytesLocal} bytes`), + }), + ); } finally { await release(); } diff --git a/extensions/xai/video-generation-transport.ts b/extensions/xai/video-generation-transport.ts index d5065fbe957c..b1ed0a0e8109 100644 --- a/extensions/xai/video-generation-transport.ts +++ b/extensions/xai/video-generation-transport.ts @@ -1,14 +1,10 @@ -import { extensionForMime } from "openclaw/plugin-sdk/media-mime"; +import { downloadGeneratedVideoAsset } from "openclaw/plugin-sdk/media-generation-runtime"; import { assertOkOrThrowHttpError, - createProviderOperationDeadline, - createProviderOperationTimeoutResolver, executeProviderOperationWithRetry, fetchWithTimeoutGuarded, type ProviderOperationTimeoutMs, } from "openclaw/plugin-sdk/provider-http"; -import { readResponseWithLimit } from "openclaw/plugin-sdk/response-limit-runtime"; -import { normalizeOptionalString } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { GeneratedVideoAsset } from "openclaw/plugin-sdk/video-generation"; export type XaiVideoRequestPolicy = { @@ -73,43 +69,27 @@ export async function downloadXaiVideo( maxBytes: number; } & XaiVideoRequestPolicy, ): Promise { - const deadline = createProviderOperationDeadline({ - timeoutMs: params.timeoutMs ?? params.defaultTimeoutMs, - label: "xAI generated video download", - }); - const timeoutMs = createProviderOperationTimeoutResolver({ - deadline, - defaultTimeoutMs: deadline.timeoutMs ?? params.defaultTimeoutMs, - }); - const { response, release } = await fetchXaiVideoResponse({ + return await downloadGeneratedVideoAsset({ url: params.url, - stage: "download", - requestFailedMessage: "xAI generated video download failed", - auditContext: "xai-video-download", - init: { method: "GET" }, - timeoutMs, + timeoutMs: params.timeoutMs ?? params.defaultTimeoutMs, defaultTimeoutMs: params.defaultTimeoutMs, - allowPrivateNetwork: params.allowPrivateNetwork, - dispatcherPolicy: params.dispatcherPolicy, fetchFn: params.fetchFn, + provider: "xai", + label: "xAI generated video download", + requestFailedMessage: "xAI generated video download failed", + maxBytes: params.maxBytes, + fetchResponse: async ({ timeoutMs }) => + await fetchXaiVideoResponse({ + url: params.url, + stage: "download", + requestFailedMessage: "xAI generated video download failed", + auditContext: "xai-video-download", + init: { method: "GET" }, + timeoutMs, + defaultTimeoutMs: params.defaultTimeoutMs, + allowPrivateNetwork: params.allowPrivateNetwork, + dispatcherPolicy: params.dispatcherPolicy, + fetchFn: params.fetchFn, + }), }); - 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`), - }); - return { - buffer, - mimeType, - fileName: `video-1.${extensionForMime(mimeType)?.slice(1) ?? "mp4"}`, - }; - } finally { - await release(); - } } diff --git a/src/agents/provider-http-errors.test.ts b/src/agents/provider-http-errors.test.ts index 2de89a0f1e7a..f5a4ba36f843 100644 --- a/src/agents/provider-http-errors.test.ts +++ b/src/agents/provider-http-errors.test.ts @@ -404,6 +404,19 @@ describe("provider error utils", () => { expect(streamed.getReadCount()).toBeLessThan(20); }); + it("cancels binary response bodies rejected by content-type validation", async () => { + const cancel = vi.fn(); + const response = new Response(new ReadableStream({ cancel }), { + status: 200, + headers: { "content-type": "application/json" }, + }); + + await expect( + readProviderBinaryResponse(response, "Provider TTS failed", "audio"), + ).rejects.toThrow("Provider TTS failed: malformed audio response"); + expect(cancel).toHaveBeenCalledTimes(1); + }); + it("rejects stalled JSON response body after chunk idle timeout", async () => { const stream = new ReadableStream({ start(controller) { diff --git a/src/agents/provider-http-errors.ts b/src/agents/provider-http-errors.ts index d4934bc1c9dd..dd7e7b997530 100644 --- a/src/agents/provider-http-errors.ts +++ b/src/agents/provider-http-errors.ts @@ -24,6 +24,7 @@ const PROVIDER_TEXT_RESPONSE_MAX_BYTES = 16 * 1024 * 1024; /** Shared timeout and byte-limit options for provider response consumption. */ type ProviderResponseReadOptions = ReadResponseTextPrefixOptions & { maxBytes?: number; + onOverflow?: (params: { size: number; maxBytes: number; res: Response }) => Error; }; /** Options for bounded provider error-body normalization. */ @@ -428,12 +429,21 @@ export async function readProviderBinaryResponse( kind = "binary", opts?: ProviderResponseReadOptions, ): Promise { - assertProviderBinaryResponseContent(response, label, kind); + try { + assertProviderBinaryResponseContent(response, label, kind); + } catch (error) { + // A captured response may be teed; do not await cancellation before its + // rejected branch and dispatcher can be released. + void response.body?.cancel().catch(() => undefined); + throw error; + } 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`), + onOverflow: + opts?.onOverflow ?? + (({ maxBytes: maxBytesLocal }) => + new Error(`${label}: ${kind} response exceeds ${maxBytesLocal} bytes`)), }); if (bytes.byteLength === 0) { throw new Error(`${label}: malformed ${kind} response`); diff --git a/src/media-generation/provider-assets.test.ts b/src/media-generation/provider-assets.test.ts new file mode 100644 index 000000000000..cadd42261219 --- /dev/null +++ b/src/media-generation/provider-assets.test.ts @@ -0,0 +1,60 @@ +import { describe, expect, it, vi } from "vitest"; +import { downloadGeneratedVideoAsset } from "./provider-assets.js"; + +describe("downloadGeneratedVideoAsset", () => { + it("preserves indexed filenames, metadata, and caller-owned response cleanup", async () => { + const release = vi.fn(async () => undefined); + const asset = await downloadGeneratedVideoAsset({ + url: "https://cdn.example/video", + timeoutMs: 1_000, + defaultTimeoutMs: 1_000, + fetchFn: fetch, + provider: "example", + label: "Example generated video download", + requestFailedMessage: "Example generated video download failed", + index: 2, + metadata: { sourceUrl: "https://cdn.example/video" }, + fetchResponse: async () => ({ + response: new Response(new Uint8Array([1, 2, 3]), { + headers: { "content-type": "video/webm" }, + }), + release, + }), + }); + + expect(asset).toMatchObject({ + buffer: Buffer.from([1, 2, 3]), + mimeType: "video/webm", + fileName: "video-3.webm", + metadata: { sourceUrl: "https://cdn.example/video" }, + }); + expect(release).toHaveBeenCalledTimes(1); + }); + + it("cancels rejected binary bodies before releasing caller-owned transport", async () => { + const cancel = vi.fn(); + const release = vi.fn(async () => undefined); + + await expect( + downloadGeneratedVideoAsset({ + url: "https://cdn.example/video", + timeoutMs: 1_000, + defaultTimeoutMs: 1_000, + fetchFn: fetch, + provider: "example", + label: "Example generated video download", + requestFailedMessage: "Example generated video download failed", + validateBinaryResponse: true, + fetchResponse: async () => ({ + response: new Response(new ReadableStream({ cancel }), { + headers: { "content-type": "application/json" }, + }), + release, + }), + }), + ).rejects.toThrow("Example generated video download: malformed video response"); + + expect(cancel).toHaveBeenCalledTimes(1); + expect(release).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/media-generation/provider-assets.ts b/src/media-generation/provider-assets.ts new file mode 100644 index 000000000000..d92beebd460b --- /dev/null +++ b/src/media-generation/provider-assets.ts @@ -0,0 +1,88 @@ +// Downloads generated video assets under provider-owned transport policies. +import { maxBytesForKind } from "@openclaw/media-core/constants"; +import { extensionForMime } from "@openclaw/media-core/mime"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { readProviderBinaryResponse } from "../agents/provider-http-errors.js"; +import { readResponseWithLimit } from "../infra/http-body.js"; +import { + createProviderOperationDeadline, + createProviderOperationTimeoutResolver, + fetchProviderDownloadResponse, + type ProviderOperationDeadline, + type ProviderOperationTimeoutMs, +} from "../media-understanding/shared.js"; +import type { GeneratedVideoAsset } from "../video-generation/types.js"; + +type GeneratedVideoResponseHandle = { + response: Response; + release?: () => Promise; +}; + +type GeneratedVideoResponseFactory = (params: { + deadline: ProviderOperationDeadline; + timeoutMs: () => number; +}) => Promise; + +/** Download a generated video URL with size limits and inferred video metadata. */ +export async function downloadGeneratedVideoAsset(params: { + url: string; + timeoutMs: ProviderOperationTimeoutMs; + defaultTimeoutMs: number; + fetchFn: typeof fetch; + provider: string; + label: string; + requestFailedMessage: string; + index?: number; + maxBytes?: number; + validateBinaryResponse?: boolean; + metadata?: Record; + fetchResponse?: GeneratedVideoResponseFactory; +}): Promise { + const deadline = createProviderOperationDeadline({ + timeoutMs: params.timeoutMs, + label: params.label, + }); + const timeoutMs = createProviderOperationTimeoutResolver({ + deadline, + defaultTimeoutMs: deadline.timeoutMs ?? params.defaultTimeoutMs, + }); + const handle = params.fetchResponse + ? await params.fetchResponse({ deadline, timeoutMs }) + : { + response: await fetchProviderDownloadResponse({ + url: params.url, + init: { method: "GET" }, + deadline, + fetchFn: params.fetchFn, + provider: params.provider, + requestFailedMessage: params.requestFailedMessage, + }), + }; + try { + const mimeType = + normalizeOptionalString(handle.response.headers.get("content-type")) ?? "video/mp4"; + const maxBytes = params.maxBytes ?? maxBytesForKind("video"); + const readOptions = { + maxBytes, + timeoutMs, + onTimeout: ({ timeoutMs: bodyTimeoutMs }: { timeoutMs: number }) => + new Error(`${params.label} timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`), + onOverflow: ({ maxBytes: maxBytesLocal }: { maxBytes: number }) => + new Error(`${params.label} exceeds ${maxBytesLocal} bytes`), + }; + const buffer = params.validateBinaryResponse + ? Buffer.from( + await readProviderBinaryResponse(handle.response, params.label, "video", readOptions), + ) + : await readResponseWithLimit(handle.response, maxBytes, readOptions); + const ext = extensionForMime(mimeType)?.replace(/^\./u, "") ?? "mp4"; + return { + buffer, + mimeType, + fileName: `video-${(params.index ?? 0) + 1}.${ext}`, + ...(params.metadata ? { metadata: params.metadata } : {}), + }; + } finally { + await handle.release?.(); + } +} diff --git a/src/music-generation/provider-assets.test.ts b/src/music-generation/provider-assets.test.ts index 0e871452b6ec..9cb3a7fbf0e3 100644 --- a/src/music-generation/provider-assets.test.ts +++ b/src/music-generation/provider-assets.test.ts @@ -1,5 +1,5 @@ -import { describe, expect, it } from "vitest"; -import { generatedMusicAssetFromBase64 } from "./provider-assets.js"; +import { describe, expect, it, vi } from "vitest"; +import { downloadGeneratedMusicAsset, generatedMusicAssetFromBase64 } from "./provider-assets.js"; describe("generatedMusicAssetFromBase64", () => { it.each([ @@ -11,3 +11,30 @@ describe("generatedMusicAssetFromBase64", () => { ); }); }); + +describe("downloadGeneratedMusicAsset", () => { + it("preserves custom response MIME types and optional source metadata", async () => { + const release = vi.fn(async () => undefined); + const asset = await downloadGeneratedMusicAsset({ + candidate: { url: "https://cdn.example/track" }, + timeoutMs: 1_000, + fetchFn: fetch, + provider: "Example", + requestFailedMessage: "Example generated music download failed", + includeSourceUrl: false, + fetchResponse: async () => ({ + response: new Response(new Uint8Array([1, 2, 3])), + mimeType: "application/octet-stream", + release, + }), + }); + + expect(asset).toMatchObject({ + buffer: Buffer.from([1, 2, 3]), + mimeType: "application/octet-stream", + fileName: "track-1.mp3", + }); + expect(asset.metadata).toBeUndefined(); + expect(release).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/music-generation/provider-assets.ts b/src/music-generation/provider-assets.ts index 3dd50eeb1d42..a9d9950ae6c2 100644 --- a/src/music-generation/provider-assets.ts +++ b/src/music-generation/provider-assets.ts @@ -4,14 +4,27 @@ import { maxBytesForKind } from "@openclaw/media-core/constants"; import { extensionForMime } from "@openclaw/media-core/mime"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; +import { readProviderBinaryResponse } from "../agents/provider-http-errors.js"; import { readResponseWithLimit } from "../infra/http-body.js"; import { createProviderOperationDeadline, createProviderOperationTimeoutResolver, fetchProviderDownloadResponse, + type ProviderOperationDeadline, } from "../media-understanding/shared.js"; import type { GeneratedMusicAsset } from "./types.js"; +type GeneratedMusicResponseHandle = { + response: Response; + release?: () => Promise; + mimeType?: string; +}; + +type GeneratedMusicResponseFactory = (params: { + deadline: ProviderOperationDeadline; + timeoutMs: () => number; +}) => Promise; + /** * Asset extraction and download helpers for music generation providers. * @@ -106,6 +119,9 @@ export async function downloadGeneratedMusicAsset(params: { requestFailedMessage: string; index?: number; maxBytes?: number; + validateBinaryResponse?: boolean; + includeSourceUrl?: boolean; + fetchResponse?: GeneratedMusicResponseFactory; }): Promise { const deadline = createProviderOperationDeadline({ timeoutMs: params.timeoutMs, @@ -115,34 +131,48 @@ export async function downloadGeneratedMusicAsset(params: { deadline, defaultTimeoutMs: params.timeoutMs, }); - const response = await fetchProviderDownloadResponse({ - url: params.candidate.url, - init: { method: "GET" }, - deadline, - fetchFn: params.fetchFn, - provider: params.provider, - requestFailedMessage: params.requestFailedMessage, - }); - const mimeType = - normalizeSpecificAudioMimeType(response.headers.get("content-type")) ?? - normalizeSpecificAudioMimeType(params.candidate.mimeType) ?? - "audio/mpeg"; - const ext = extensionForMime(mimeType)?.replace(/^\./u, "") || "mp3"; - const maxBytes = params.maxBytes ?? maxBytesForKind("audio"); - return { - buffer: await readResponseWithLimit(response, maxBytes, { + const handle = params.fetchResponse + ? await params.fetchResponse({ deadline, timeoutMs }) + : { + response: await fetchProviderDownloadResponse({ + url: params.candidate.url, + init: { method: "GET" }, + deadline, + fetchFn: params.fetchFn, + provider: params.provider, + requestFailedMessage: params.requestFailedMessage, + }), + }; + try { + const mimeType = + handle.mimeType ?? + normalizeSpecificAudioMimeType(handle.response.headers.get("content-type")) ?? + normalizeSpecificAudioMimeType(params.candidate.mimeType) ?? + "audio/mpeg"; + const ext = extensionForMime(mimeType)?.replace(/^\./u, "") || "mp3"; + const maxBytes = params.maxBytes ?? maxBytesForKind("audio"); + const readOptions = { + maxBytes, timeoutMs, - onTimeout: ({ timeoutMs: bodyTimeoutMs }) => + onTimeout: ({ timeoutMs: bodyTimeoutMs }: { timeoutMs: number }) => new Error( `${params.provider} generated music download timed out after ${deadline.timeoutMs ?? bodyTimeoutMs}ms`, ), - onOverflow: ({ maxBytes: maxBytesLocal }) => + onOverflow: ({ maxBytes: maxBytesLocal }: { maxBytes: number }) => new Error(`${params.provider} generated music download exceeds ${maxBytesLocal} bytes`), - }), - mimeType, - fileName: params.candidate.fileName ?? `track-${(params.index ?? 0) + 1}.${ext}`, - metadata: { - url: params.candidate.url, - }, - }; + }; + const buffer = params.validateBinaryResponse + ? Buffer.from( + await readProviderBinaryResponse(handle.response, deadline.label, "audio", readOptions), + ) + : await readResponseWithLimit(handle.response, maxBytes, readOptions); + return { + buffer, + mimeType, + fileName: params.candidate.fileName ?? `track-${(params.index ?? 0) + 1}.${ext}`, + ...(params.includeSourceUrl === false ? {} : { metadata: { url: params.candidate.url } }), + }; + } finally { + await handle.release?.(); + } } diff --git a/src/plugin-sdk/media-generation-runtime.ts b/src/plugin-sdk/media-generation-runtime.ts index 503e467a3aaf..36dccf704f47 100644 --- a/src/plugin-sdk/media-generation-runtime.ts +++ b/src/plugin-sdk/media-generation-runtime.ts @@ -2,3 +2,4 @@ export { resolveClosestSize } from "../media-generation/runtime-shared.js"; export { resolveGeneratedMediaMaxBytes } from "../media/configured-max-bytes.js"; +export { downloadGeneratedVideoAsset } from "../media-generation/provider-assets.js";