fix(openrouter): bound music SSE buffering (#101488)

* fix(openrouter): bound music SSE buffering Co-Authored-By: Claude <noreply@anthropic.com> * fix(openrouter): align music stream bounds with media config Co-authored-by: mikasa0818 <0668001030@xydigit.com> --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-07-21 10:16:44 +00:00 · 2026-07-09 22:10:44 +01:00
parent ae11ea5ba9
commit ce6cb3b250
3 changed files with 248 additions and 22 deletions
@@ -172,6 +172,8 @@ and also exposes `google/lyria-3-clip-preview`. OpenClaw sends `modalities:
 ["text", "audio"]`, streams the response, collects the audio chunks, and saves
 the result as generated media for channel delivery. Lyria models accept one
 reference image through the shared `music_generate image=...` parameter.
+Streaming audio, transcript retention, and the derived SSE event envelope are
+bounded by `agents.defaults.mediaMaxMb` (the default audio cap is 16 MB).

 ## Text-to-speech

@@ -39,16 +39,27 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importOriginal) => {
  };
 });

-function sseResponse(lines: string[], options?: { releaseLock?: () => void }): Response {
+function sseResponse(
+  lines: Array<string | Uint8Array>,
+  options?: { cancel?: () => void; releaseLock?: () => void },
+): Response {
  const encoder = new TextEncoder();
-  if (!options?.releaseLock) {
+  const encodeLine = (line: string | Uint8Array) =>
+    typeof line === "string" ? encoder.encode(line) : line;
+  if (!options?.releaseLock && !options?.cancel) {
+    let index = 0;
    return new Response(
      new ReadableStream({
-        start(controller) {
-          for (const line of lines) {
-            controller.enqueue(encoder.encode(line));
+        pull(controller) {
+          const line = lines[index++];
+          if (line === undefined) {
+            controller.close();
+            return;
          }
-          controller.close();
+          controller.enqueue(encodeLine(line));
+        },
+        cancel() {
+          options?.cancel?.();
        },
      }),
      { status: 200, headers: { "content-type": "text/event-stream" } },
@@ -57,13 +68,13 @@ function sseResponse(lines: string[], options?: { releaseLock?: () => void }): R

  const chunks: Array<ReadableStreamReadResult<Uint8Array>> = lines.map((line) => ({
    done: false,
-    value: encoder.encode(line),
+    value: encodeLine(line),
  }));
  chunks.push({ done: true, value: undefined });
  const reader = {
    read: async () => chunks.shift() ?? { done: true, value: undefined },
-    cancel: async () => undefined,
-    releaseLock: options.releaseLock,
+    cancel: async () => options?.cancel?.(),
+    releaseLock: options?.releaseLock ?? (() => {}),
  } as ReadableStreamDefaultReader<Uint8Array>;

  return {
@@ -164,8 +175,8 @@ describe("openrouter music generation provider", () => {
    postJsonRequestMock.mockResolvedValue({
      response: sseResponse([
        `data: ${JSON.stringify({ choices: [{ delta: { audio: { transcript: "line " } } }] })}\n`,
-        `data: ${JSON.stringify({ choices: [{ delta: { audio: { data: audioBase64.slice(0, 4) } } }] })}\n`,
-        `data: ${JSON.stringify({ choices: [{ delta: { audio: { data: audioBase64.slice(4), transcript: "two" } } }] })}\n`,
+        `data: ${JSON.stringify({ choices: [{ delta: { audio: { data: audioBase64.slice(0, 5) } } }] })}\n`,
+        `data: ${JSON.stringify({ choices: [{ delta: { audio: { data: audioBase64.slice(5), transcript: "two" } } }] })}\n`,
        "data: [DONE]\n",
      ]),
      release,
@@ -352,4 +363,100 @@ describe("openrouter music generation provider", () => {
      }),
    ).rejects.toThrow("OpenRouter music generation stream ended before completion");
  });
+
+  it("surfaces and cancels OpenRouter mid-stream errors", async () => {
+    const cancel = vi.fn();
+    postJsonRequestMock.mockResolvedValue({
+      response: sseResponse(
+        [
+          `data: ${JSON.stringify({
+            error: { code: "provider_error", message: "provider disconnected" },
+            choices: [{ delta: {}, finish_reason: "error" }],
+          })}\n`,
+        ],
+        { cancel },
+      ),
+      release: vi.fn(async () => {}),
+    });
+
+    await expect(
+      buildOpenRouterMusicGenerationProvider().generateMusic({
+        provider: "openrouter",
+        model: "google/lyria-3-clip-preview",
+        prompt: "surface provider failure",
+        cfg: {},
+      }),
+    ).rejects.toThrow("OpenRouter music generation failed: provider disconnected");
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it("accepts valid SSE events above the old fixed two-megabyte cap", async () => {
+    const audio = Buffer.alloc(1_600_000, 0x61);
+    postJsonRequestMock.mockResolvedValue({
+      response: sseResponse([
+        `data: ${JSON.stringify({ choices: [{ delta: { audio: { data: audio.toString("base64") } } }] })}\n`,
+        "data: [DONE]\n",
+      ]),
+      release: vi.fn(async () => {}),
+    });
+
+    const result = await buildOpenRouterMusicGenerationProvider().generateMusic({
+      provider: "openrouter",
+      model: "google/lyria-3-clip-preview",
+      prompt: "large valid audio event",
+      cfg: { agents: { defaults: { mediaMaxMb: 2 } } },
+    });
+
+    expect(result.tracks[0]?.buffer).toEqual(audio);
+  });
+
+  it("rejects OpenRouter music SSE events outside the configured media envelope", async () => {
+    const maxBytes = 8;
+    const maxEventBytes = Math.ceil(maxBytes / 3) * 4 + maxBytes + 64 * 1024;
+    const cancel = vi.fn();
+    postJsonRequestMock.mockResolvedValue({
+      response: sseResponse([new Uint8Array(maxEventBytes + 1)], { cancel }),
+      release: vi.fn(async () => {}),
+    });
+
+    await expect(
+      buildOpenRouterMusicGenerationProvider().generateMusic({
+        provider: "openrouter",
+        model: "google/lyria-3-clip-preview",
+        prompt: "unterminated event",
+        cfg: { agents: { defaults: { mediaMaxMb: maxBytes / (1024 * 1024) } } },
+      }),
+    ).rejects.toThrow(
+      `OpenRouter music generation SSE event exceeded ${maxEventBytes} bytes for a ${maxBytes}-byte media limit`,
+    );
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
+  it.each([
+    {
+      label: "audio",
+      delta: { audio: { data: Buffer.from("123456789").toString("base64") } },
+    },
+    {
+      label: "transcript",
+      delta: { audio: { transcript: "123456789" } },
+    },
+  ])("rejects $label beyond agents.defaults.mediaMaxMb", async ({ label, delta }) => {
+    postJsonRequestMock.mockResolvedValue({
+      response: sseResponse([
+        `data: ${JSON.stringify({ choices: [{ delta }] })}\n`,
+        "data: [DONE]\n",
+      ]),
+      release: vi.fn(async () => {}),
+    });
+
+    await expect(
+      buildOpenRouterMusicGenerationProvider().generateMusic({
+        provider: "openrouter",
+        model: "google/lyria-3-clip-preview",
+        prompt: `oversized ${label}`,
+        cfg: { agents: { defaults: { mediaMaxMb: 8 / (1024 * 1024) } } },
+      }),
+    ).rejects.toThrow(`OpenRouter music generation ${label} exceeded 8 bytes`);
+  });
 });
@@ -1,5 +1,6 @@
 // Openrouter provider module implements model/runtime integration.
 import { toImageDataUrl } from "openclaw/plugin-sdk/image-generation";
+import { maxBytesForKind } from "openclaw/plugin-sdk/media-runtime";
 import type {
  MusicGenerationProvider,
  MusicGenerationRequest,
@@ -22,6 +23,8 @@ import { OPENROUTER_BASE_URL } from "./provider-catalog.js";
 const DEFAULT_OPENROUTER_MUSIC_MODEL = "google/lyria-3-pro-preview";
 const OPENROUTER_CLIP_MUSIC_MODEL = "google/lyria-3-clip-preview";
 const DEFAULT_TIMEOUT_MS = 180_000;
+const MB = 1024 * 1024;
+const OPENROUTER_SSE_ENVELOPE_OVERHEAD_BYTES = 64 * 1024;
 const OPENROUTER_MUSIC_MODELS = [
  DEFAULT_OPENROUTER_MUSIC_MODEL,
  OPENROUTER_CLIP_MUSIC_MODEL,
@@ -32,6 +35,15 @@ type OpenRouterAudioStreamResult = {
  transcript: string;
 };

+type OpenRouterAudioStreamAccumulator = {
+  audioBuffers: Buffer[];
+  audioBytes: number;
+  audioBase64Remainder: string;
+  transcriptChunks: string[];
+  transcriptBytes: number;
+  maxBytes: number;
+};
+
 function resolveOpenRouterMusicModel(model: string | undefined): string {
  return normalizeOptionalString(model) ?? DEFAULT_OPENROUTER_MUSIC_MODEL;
 }
@@ -112,10 +124,77 @@ function readDeltaAudio(part: unknown): { data?: string; transcript?: string } |
  };
 }

-function processOpenRouterSseLine(
-  line: string,
-  result: { audioBuffers: Buffer[]; transcriptChunks: string[] },
-): boolean {
+function resolveGeneratedMusicMaxBytes(req: MusicGenerationRequest): number {
+  const configured = req.cfg.agents?.defaults?.mediaMaxMb;
+  if (typeof configured === "number" && Number.isFinite(configured) && configured > 0) {
+    return Math.floor(configured * MB);
+  }
+  return maxBytesForKind("audio");
+}
+
+function resolveOpenRouterSseEventMaxBytes(maxBytes: number): number {
+  const maxBase64Bytes = Math.ceil(maxBytes / 3) * 4;
+  return maxBase64Bytes + maxBytes + OPENROUTER_SSE_ENVELOPE_OVERHEAD_BYTES;
+}
+
+function createOpenRouterMusicTooLargeError(kind: "audio" | "transcript", maxBytes: number) {
+  return new Error(`OpenRouter music generation ${kind} exceeded ${maxBytes} bytes`);
+}
+
+function appendDecodedOpenRouterMusicAudio(
+  result: OpenRouterAudioStreamAccumulator,
+  base64: string,
+): void {
+  if (!base64) {
+    return;
+  }
+  const decodedBytes = Buffer.byteLength(base64, "base64");
+  if (decodedBytes > result.maxBytes - result.audioBytes) {
+    throw createOpenRouterMusicTooLargeError("audio", result.maxBytes);
+  }
+  const buffer = Buffer.from(base64, "base64");
+  const nextBytes = result.audioBytes + buffer.byteLength;
+  if (nextBytes > result.maxBytes) {
+    throw createOpenRouterMusicTooLargeError("audio", result.maxBytes);
+  }
+  result.audioBytes = nextBytes;
+  result.audioBuffers.push(buffer);
+}
+
+function appendOpenRouterMusicAudio(result: OpenRouterAudioStreamAccumulator, data: string): void {
+  // OpenRouter defines delta.audio.data as slices of one base64 stream. Keep
+  // only the incomplete quartet so arbitrary provider chunk boundaries decode safely.
+  const combined = result.audioBase64Remainder + data;
+  const completeLength = combined.length - (combined.length % 4);
+  result.audioBase64Remainder = combined.slice(completeLength);
+  appendDecodedOpenRouterMusicAudio(result, combined.slice(0, completeLength));
+}
+
+function flushOpenRouterMusicAudio(result: OpenRouterAudioStreamAccumulator): void {
+  appendDecodedOpenRouterMusicAudio(result, result.audioBase64Remainder);
+  result.audioBase64Remainder = "";
+}
+
+function appendOpenRouterMusicTranscript(
+  result: OpenRouterAudioStreamAccumulator,
+  transcript: string,
+): void {
+  const nextBytes = result.transcriptBytes + Buffer.byteLength(transcript, "utf8");
+  if (nextBytes > result.maxBytes) {
+    throw createOpenRouterMusicTooLargeError("transcript", result.maxBytes);
+  }
+  result.transcriptBytes = nextBytes;
+  result.transcriptChunks.push(transcript);
+}
+
+function readOpenRouterStreamError(part: unknown): string | undefined {
+  if (!isRecord(part) || !isRecord(part.error)) {
+    return undefined;
+  }
+  return normalizeOptionalString(part.error.message) ?? "unknown provider stream error";
+}
+
+function processOpenRouterSseLine(line: string, result: OpenRouterAudioStreamAccumulator): boolean {
  if (!line.startsWith("data:")) {
    return false;
  }
@@ -126,12 +205,17 @@ function processOpenRouterSseLine(
  if (data === "[DONE]") {
    return true;
  }
-  const audio = readDeltaAudio(JSON.parse(data));
+  const payload: unknown = JSON.parse(data);
+  const streamError = readOpenRouterStreamError(payload);
+  if (streamError) {
+    throw new Error(`OpenRouter music generation failed: ${streamError}`);
+  }
+  const audio = readDeltaAudio(payload);
  if (audio?.data) {
-    result.audioBuffers.push(Buffer.from(audio.data, "base64"));
+    appendOpenRouterMusicAudio(result, audio.data);
  }
  if (audio?.transcript) {
-    result.transcriptChunks.push(audio.transcript);
+    appendOpenRouterMusicTranscript(result, audio.transcript);
  }
  return false;
 }
@@ -171,14 +255,24 @@ async function readOpenRouterStreamChunk(
 async function readOpenRouterAudioStream(
  response: Response,
  deadline: ProviderOperationDeadline,
+  maxBytes: number,
 ): Promise<OpenRouterAudioStreamResult> {
  if (!response.body) {
    throw new Error("OpenRouter music generation response missing stream body");
  }
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
-  const result = { audioBuffers: [] as Buffer[], transcriptChunks: [] as string[] };
+  const result = {
+    audioBuffers: [] as Buffer[],
+    audioBytes: 0,
+    audioBase64Remainder: "",
+    transcriptChunks: [] as string[],
+    transcriptBytes: 0,
+    maxBytes,
+  };
+  const maxEventBytes = resolveOpenRouterSseEventMaxBytes(maxBytes);
  let buffer = "";
+  let pendingBytes = 0;
  let doneSeen = false;
  try {
    for (;;) {
@@ -186,14 +280,23 @@ async function readOpenRouterAudioStream(
      if (done) {
        break;
      }
+      for (const byte of value) {
+        pendingBytes = byte === 0x0a ? 0 : pendingBytes + 1;
+        if (pendingBytes > maxEventBytes) {
+          throw new Error(
+            `OpenRouter music generation SSE event exceeded ${maxEventBytes} bytes for a ${maxBytes}-byte media limit`,
+          );
+        }
+      }
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split(/\r?\n/u);
      buffer = lines.pop() ?? "";
      for (const line of lines) {
        if (processOpenRouterSseLine(line.trim(), result)) {
+          flushOpenRouterMusicAudio(result);
          await reader.cancel();
          return {
-            audioBuffer: Buffer.concat(result.audioBuffers),
+            audioBuffer: Buffer.concat(result.audioBuffers, result.audioBytes),
            transcript: result.transcriptChunks.join(""),
          };
        }
@@ -201,6 +304,12 @@ async function readOpenRouterAudioStream(
    }
    resolveOpenRouterStreamRemainingMs(deadline);
    buffer += decoder.decode();
+    pendingBytes = Buffer.byteLength(buffer, "utf8");
+    if (pendingBytes > maxEventBytes) {
+      throw new Error(
+        `OpenRouter music generation SSE event exceeded ${maxEventBytes} bytes for a ${maxBytes}-byte media limit`,
+      );
+    }
    if (buffer.trim()) {
      for (const line of buffer.split(/\r?\n/u)) {
        if (processOpenRouterSseLine(line.trim(), result)) {
@@ -211,10 +320,14 @@ async function readOpenRouterAudioStream(
    if (!doneSeen) {
      throw new Error("OpenRouter music generation stream ended before completion");
    }
+    flushOpenRouterMusicAudio(result);
    return {
-      audioBuffer: Buffer.concat(result.audioBuffers),
+      audioBuffer: Buffer.concat(result.audioBuffers, result.audioBytes),
      transcript: result.transcriptChunks.join(""),
    };
+  } catch (error) {
+    await reader.cancel().catch(() => {});
+    throw error;
  } finally {
    try {
      reader.releaseLock();
@@ -310,7 +423,11 @@ export function buildOpenRouterMusicGenerationProvider(): MusicGenerationProvide

      try {
        await assertOkOrThrowHttpError(response, "OpenRouter music generation failed");
-        const streamResult = await readOpenRouterAudioStream(response, streamDeadline);
+        const streamResult = await readOpenRouterAudioStream(
+          response,
+          streamDeadline,
+          resolveGeneratedMusicMaxBytes(req),
+        );
        if (streamResult.audioBuffer.byteLength === 0) {
          throw new Error("OpenRouter music generation response missing audio data");
        }