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

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<string, unknown> = {
-          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<string, unknown> = {
+        guardrailIdentifier: guardrailConfig.guardrailIdentifier,
+        guardrailVersion: guardrailConfig.guardrailVersion,
+      };
+      if (guardrailConfig.streamProcessingMode) {
+        gc.streamProcessingMode = guardrailConfig.streamProcessingMode;
+      }
+      if (guardrailConfig.trace) {
+        gc.trace = guardrailConfig.trace;
+      }
+      payload.guardrailConfig = gc;
+    });
   };
 }
 

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. */

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();
   }

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<GeneratedVideoAsset> {
-  const deadline = createProviderOperationDeadline({
-    timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-    label: "BytePlus generated video download",
-  });
-  const timeoutMs = createProviderOperationTimeoutResolver({
-    deadline,
-    defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-  });
-  const response = await fetchProviderDownloadResponse({
+  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. */

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<typeof streamWithPayloadPatch>[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/"),
+    },
+  );
 }

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(

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<string, unknown>);
-              }
-            },
-            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<string, unknown>);
+            }
+          },
+          payload,
+        ),
+    });
   };
 }
 

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();
   }

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);
   };
 }
 

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<string, unknown>): 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 {

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<GeneratedMusicAsset> {
-  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 {

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<GeneratedVideoAsset> {
-  const deadline = createProviderOperationDeadline({
-    timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-    label: "MiniMax generated video download",
-  });
-  const timeoutMs = createProviderOperationTimeoutResolver({
-    deadline,
-    defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-  });
-  const { response, release } = await fetchMinimaxResponse({
-    stage: "download",
+  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: {

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<string, unknown>).num_ctx = numCtx;
-    });
+  return createPayloadPatchStreamWrapper(baseFn, ({ payload }) => {
+    if (!payload.options || typeof payload.options !== "object") {
+      payload.options = {};
+    }
+    (payload.options as Record<string, unknown>).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 {

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 }),
+    },
+  );
 }

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();
   }

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<GeneratedVideoAsset> {
   const url = new URL(`${params.baseUrl}/videos/${params.videoId}/content`);
   url.searchParams.set("variant", "video");
-  const deadline = createProviderOperationDeadline({
-    timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-    label: "OpenAI generated video download",
-  });
-  const timeoutMs = createProviderOperationTimeoutResolver({
-    deadline,
-    defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-  });
-  const { response, release } = await fetchOpenAIVideoDownload({
+  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 {

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(

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<GeneratedVideoAsset[]> {
   const videos: GeneratedVideoAsset[] = [];
   for (const [index, url] of params.urls.entries()) {
-    const deadline = createProviderOperationDeadline({
-      timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-      label: "Runway generated video download",
-    });
-    const timeoutMs = createProviderOperationTimeoutResolver({
-      deadline,
-      defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-    });
-    const response = await fetchProviderDownloadResponse({
-      url,
-      init: { method: "GET" },
-      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;
 }

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<GeneratedVideoAsset> {
-  const deadline = createProviderOperationDeadline({
-    timeoutMs: params.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-    label: "Together generated video download",
-  });
-  const timeoutMs = createProviderOperationTimeoutResolver({
-    deadline,
-    defaultTimeoutMs: deadline.timeoutMs ?? DEFAULT_TIMEOUT_MS,
-  });
-  const response = await fetchProviderDownloadResponse({
+  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 {

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();
   }

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<GeneratedVideoAsset> {
-  const deadline = createProviderOperationDeadline({
-    timeoutMs: params.timeoutMs ?? params.defaultTimeoutMs,
-    label: "xAI generated video download",
-  });
-  const timeoutMs = createProviderOperationTimeoutResolver({
-    deadline,
-    defaultTimeoutMs: deadline.timeoutMs ?? params.defaultTimeoutMs,
-  });
-  const { response, release } = await fetchXaiVideoResponse({
+  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();
-  }
 }

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<Uint8Array>({ 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<Uint8Array>({
       start(controller) {

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<Uint8Array> {
-  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`);

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<Uint8Array>({ cancel }), {
+            headers: { "content-type": "application/json" },
+          }),
+          release,
+        }),
+      }),
+    ).rejects.toThrow("Example generated video download: malformed video response");
+
+    expect(cancel).toHaveBeenCalledTimes(1);
+    expect(release).toHaveBeenCalledTimes(1);
+  });
+});

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<void>;
+};
+
+type GeneratedVideoResponseFactory = (params: {
+  deadline: ProviderOperationDeadline;
+  timeoutMs: () => number;
+}) => Promise<GeneratedVideoResponseHandle>;
+
+/** 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<string, unknown>;
+  fetchResponse?: GeneratedVideoResponseFactory;
+}): Promise<GeneratedVideoAsset> {
+  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?.();
+  }
+}

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);
+  });
+});

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<void>;
+  mimeType?: string;
+};
+
+type GeneratedMusicResponseFactory = (params: {
+  deadline: ProviderOperationDeadline;
+  timeoutMs: () => number;
+}) => Promise<GeneratedMusicResponseHandle>;
+
 /**
  * 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<GeneratedMusicAsset> {
   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?.();
+  }
 }

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";