From dd00b38304d5ed0184fcdb9f94101a49f742f4a4 Mon Sep 17 00:00:00 2001
From: yuanhe <yuanhe@minimaxi.com>
Date: Mon, 11 May 2026 11:47:45 +0800
Subject: [PATCH] fix: decode SSE audio stream properly for music and speech

Music streaming was writing raw SSE text instead of decoded audio bytes.
Two root causes:
- extra_info filter skipped all music events (null !== undefined is true)
- AbortSignal.timeout aborted long-running streaming responses

Extracted pipeAudioStream() to shared util; uses status===2 to skip
the terminal summary event. Disabled fetch timeout for streaming.
---
 src/client/http.ts                |  2 +-
 src/commands/music/cover.ts       | 10 ++--------
 src/commands/music/generate.ts    | 12 +++---------
 src/commands/speech/synthesize.ts | 11 ++---------
 src/utils/audio-stream.ts         | 29 +++++++++++++++++++++++++++++
 5 files changed, 37 insertions(+), 27 deletions(-)
 create mode 100644 src/utils/audio-stream.ts

diff --git a/src/client/http.ts b/src/client/http.ts
index 7e327d6..c77ba79 100644
--- a/src/client/http.ts
+++ b/src/client/http.ts
@@ -62,7 +62,7 @@ export async function request(config: Config, opts: RequestOpts): Promise<Respon
         ? (opts.body as FormData)
         : JSON.stringify(opts.body)
       : undefined,
-    signal: AbortSignal.timeout(timeoutMs),
+    signal: opts.stream ? undefined : AbortSignal.timeout(timeoutMs),
   });
 
   if (config.verbose) {
diff --git a/src/commands/music/cover.ts b/src/commands/music/cover.ts
index 610f575..b7c8460 100644
--- a/src/commands/music/cover.ts
+++ b/src/commands/music/cover.ts
@@ -7,6 +7,7 @@ import { musicEndpoint } from '../../client/endpoints';
 import { formatOutput, detectOutputFormat } from '../../output/formatter';
 import { saveAudioOutput } from '../../output/audio';
 import { MUSIC_FORMATS, formatList, validateAudioFormat } from '../../utils/audio-formats';
+import { pipeAudioStream } from '../../utils/audio-stream';
 import type { Config } from '../../config/schema';
 import type { GlobalFlags } from '../../types/flags';
 import type { MusicRequest, MusicResponse } from '../../types/api';
@@ -109,14 +110,7 @@ export default defineCommand({
 
     if (flags.stream) {
       const res = await request(config, { url, method: 'POST', body, stream: true });
-      const reader = res.body?.getReader();
-      if (!reader) throw new CLIError('No response body', ExitCode.GENERAL);
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        process.stdout.write(value);
-      }
-      reader.releaseLock();
+      await pipeAudioStream(res);
       return;
     }
 
diff --git a/src/commands/music/generate.ts b/src/commands/music/generate.ts
index 3a4bfe3..79d0806 100644
--- a/src/commands/music/generate.ts
+++ b/src/commands/music/generate.ts
@@ -3,10 +3,11 @@ import { CLIError } from '../../errors/base';
 import { ExitCode } from '../../errors/codes';
 import { request, requestJson } from '../../client/http';
 import { musicEndpoint } from '../../client/endpoints';
-import { formatOutput, detectOutputFormat, dryRun } from '../../output/formatter';
+import { detectOutputFormat, dryRun } from '../../output/formatter';
 import { saveAudioOutput } from '../../output/audio';
 import { readTextFromPathOrStdin } from '../../utils/fs';
 import { MUSIC_FORMATS, formatList, validateAudioFormat } from '../../utils/audio-formats';
+import { pipeAudioStream } from '../../utils/audio-stream';
 import type { Config } from '../../config/schema';
 import type { GlobalFlags } from '../../types/flags';
 import type { MusicRequest, MusicResponse } from '../../types/api';
@@ -173,14 +174,7 @@ export default defineCommand({
 
     if (flags.stream) {
       const res = await request(config, { url, method: 'POST', body, stream: true });
-      const reader = res.body?.getReader();
-      if (!reader) throw new CLIError('No response body', ExitCode.GENERAL);
-      while (true) {
-        const { done, value } = await reader.read();
-        if (done) break;
-        process.stdout.write(value);
-      }
-      reader.releaseLock();
+      await pipeAudioStream(res);
       return;
     }
 
diff --git a/src/commands/speech/synthesize.ts b/src/commands/speech/synthesize.ts
index 0eb8244..8f968ea 100644
--- a/src/commands/speech/synthesize.ts
+++ b/src/commands/speech/synthesize.ts
@@ -3,12 +3,12 @@ import { CLIError } from '../../errors/base';
 import { ExitCode } from '../../errors/codes';
 import { request, requestJson } from '../../client/http';
 import { speechEndpoint } from '../../client/endpoints';
-import { parseSSE } from '../../client/stream';
 import { detectOutputFormat, formatOutput, dryRun } from '../../output/formatter';
 import { saveAudioOutput } from '../../output/audio';
 import { writeFileSync } from 'fs';
 import { readTextFromPathOrStdin } from '../../utils/fs';
 import { T2A_FORMATS, formatList, validateAudioFormat, validateT2AStreaming, t2aDefaultSampleRate } from '../../utils/audio-formats';
+import { pipeAudioStream } from '../../utils/audio-stream';
 import type { Config } from '../../config/schema';
 import type { GlobalFlags } from '../../types/flags';
 import type { SpeechRequest, SpeechResponse } from '../../types/api';
@@ -105,14 +105,7 @@ export default defineCommand({
 
     if (flags.stream) {
       const res = await request(config, { url, method: 'POST', body, stream: true });
-      for await (const event of parseSSE(res)) {
-        if (!event.data || event.data === '[DONE]') break;
-        const parsed = JSON.parse(event.data);
-        const audioHex = parsed?.data?.audio;
-        if (audioHex) {
-          process.stdout.write(Buffer.from(audioHex, 'hex'));
-        }
-      }
+      await pipeAudioStream(res);
       return;
     }
 
diff --git a/src/utils/audio-stream.ts b/src/utils/audio-stream.ts
new file mode 100644
index 0000000..1ff57c0
--- /dev/null
+++ b/src/utils/audio-stream.ts
@@ -0,0 +1,29 @@
+import { parseSSE } from '../client/stream';
+
+interface SseAudioPayload {
+  data?: { audio?: string; status?: number };
+}
+
+export async function pipeAudioStream(response: Response): Promise<void> {
+  process.stdout.on('error', (err: NodeJS.ErrnoException) => {
+    if (err.code === 'EPIPE') process.exit(0);
+    throw err;
+  });
+
+  for await (const event of parseSSE(response)) {
+    if (!event.data || event.data === '[DONE]') break;
+
+    let parsed: SseAudioPayload;
+    try { parsed = JSON.parse(event.data); } catch { continue; }
+
+    if (parsed.data?.status === 2) continue;
+
+    const hex = parsed.data?.audio;
+    if (!hex) continue;
+
+    const chunk = Buffer.from(hex, 'hex');
+    if (!process.stdout.write(chunk)) {
+      await new Promise<void>(r => process.stdout.once('drain', r));
+    }
+  }
+}