fix(openrouter): release music stream readers

vincentkoc · vincentkoc · commit f5419b5bb0bb · 2026-06-19T09:04:11.000+02:00
diff --git a/extensions/openrouter/music-generation-provider.test.ts b/extensions/openrouter/music-generation-provider.test.ts
@@ -39,19 +39,69 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importOriginal) => {
   };
 });
 
-function sseResponse(lines: string[]): Response {
+function sseResponse(lines: string[], options?: { releaseLock?: () => void }): Response {
   const encoder = new TextEncoder();
-  return new Response(
-    new ReadableStream({
-      start(controller) {
-        for (const line of lines) {
-          controller.enqueue(encoder.encode(line));
-        }
-        controller.close();
-      },
-    }),
-    { status: 200, headers: { "content-type": "text/event-stream" } },
-  );
+  if (!options?.releaseLock) {
+    return new Response(
+      new ReadableStream({
+        start(controller) {
+          for (const line of lines) {
+            controller.enqueue(encoder.encode(line));
+          }
+          controller.close();
+        },
+      }),
+      { status: 200, headers: { "content-type": "text/event-stream" } },
+    );
+  }
+
+  const chunks: Array<ReadableStreamReadResult<Uint8Array>> = lines.map((line) => ({
+    done: false,
+    value: encoder.encode(line),
+  }));
+  chunks.push({ done: true, value: undefined });
+  const reader = {
+    read: async () => chunks.shift() ?? { done: true, value: undefined },
+    cancel: async () => undefined,
+    releaseLock: options.releaseLock,
+  } as ReadableStreamDefaultReader<Uint8Array>;
+
+  return {
+    ok: true,
+    status: 200,
+    headers: new Headers({ "content-type": "text/event-stream" }),
+    body: {
+      getReader: () => reader,
+    },
+  } as Response;
+}
+
+function sseResponseLines(params: {
+  audio?: string;
+  transcript?: string;
+  done?: boolean;
+}): string[] {
+  const lines: string[] = [];
+  if (params.audio || params.transcript) {
+    lines.push(
+      `data: ${JSON.stringify({
+        choices: [
+          {
+            delta: {
+              audio: {
+                ...(params.audio ? { data: params.audio } : {}),
+                ...(params.transcript ? { transcript: params.transcript } : {}),
+              },
+            },
+          },
+        ],
+      })}\n`,
+    );
+  }
+  if (params.done) {
+    lines.push("data: [DONE]\n");
+  }
+  return lines;
 }
 
 function stalledSseResponse(line: string): Response {
@@ -150,6 +200,32 @@ describe("openrouter music generation provider", () => {
     expect(release).toHaveBeenCalledOnce();
   });
 
+  it("releases OpenRouter audio stream readers after completion", async () => {
+    const releaseLock = vi.fn();
+    postJsonRequestMock.mockResolvedValue({
+      response: sseResponse(
+        sseResponseLines({
+          audio: Buffer.from("wav-bytes").toString("base64"),
+          done: true,
+        }),
+        { releaseLock },
+      ),
+      release: vi.fn(async () => {}),
+    });
+
+    await expect(
+      buildOpenRouterMusicGenerationProvider().generateMusic({
+        provider: "openrouter",
+        model: "google/lyria-3-pro-preview",
+        prompt: "release stream reader",
+        cfg: {},
+      }),
+    ).resolves.toMatchObject({
+      tracks: [{ mimeType: "audio/wav" }],
+    });
+    expect(releaseLock).toHaveBeenCalledTimes(1);
+  });
+
   it("decodes independently padded OpenRouter audio chunks", async () => {
     postJsonRequestMock.mockResolvedValue({
       response: sseResponse([
diff --git a/extensions/openrouter/music-generation-provider.ts b/extensions/openrouter/music-generation-provider.ts
@@ -179,40 +179,46 @@ async function readOpenRouterAudioStream(
   const result = { audioBuffers: [] as Buffer[], transcriptChunks: [] as string[] };
   let buffer = "";
   let doneSeen = false;
-  for (;;) {
-    const { value, done } = await readOpenRouterStreamChunk(reader, deadline);
-    if (done) {
-      break;
-    }
-    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)) {
-        await reader.cancel();
-        return {
-          audioBuffer: Buffer.concat(result.audioBuffers),
-          transcript: result.transcriptChunks.join(""),
-        };
+  try {
+    for (;;) {
+      const { value, done } = await readOpenRouterStreamChunk(reader, deadline);
+      if (done) {
+        break;
+      }
+      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)) {
+          await reader.cancel();
+          return {
+            audioBuffer: Buffer.concat(result.audioBuffers),
+            transcript: result.transcriptChunks.join(""),
+          };
+        }
       }
     }
-  }
-  resolveOpenRouterStreamRemainingMs(deadline);
-  buffer += decoder.decode();
-  if (buffer.trim()) {
-    for (const line of buffer.split(/\r?\n/u)) {
-      if (processOpenRouterSseLine(line.trim(), result)) {
-        doneSeen = true;
+    resolveOpenRouterStreamRemainingMs(deadline);
+    buffer += decoder.decode();
+    if (buffer.trim()) {
+      for (const line of buffer.split(/\r?\n/u)) {
+        if (processOpenRouterSseLine(line.trim(), result)) {
+          doneSeen = true;
+        }
       }
     }
+    if (!doneSeen) {
+      throw new Error("OpenRouter music generation stream ended before completion");
+    }
+    return {
+      audioBuffer: Buffer.concat(result.audioBuffers),
+      transcript: result.transcriptChunks.join(""),
+    };
+  } finally {
+    try {
+      reader.releaseLock();
+    } catch {}
   }
-  if (!doneSeen) {
-    throw new Error("OpenRouter music generation stream ended before completion");
-  }
-  return {
-    audioBuffer: Buffer.concat(result.audioBuffers),
-    transcript: result.transcriptChunks.join(""),
-  };
 }
 
 export function buildOpenRouterMusicGenerationProvider(): MusicGenerationProvider {
