Merge pull request #130 from MiniMax-AI/fix/stream-audio-decode

fix: decode SSE audio stream for music and speech

Author: RyanLee-Dev

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

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

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

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

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