fix(tts): propagate and validate actual audio MIME (#41043)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: WH-2099 <wh2099@pm.me>

api/constants/tts_auto_play_timeout.py

@@ -1,4 +0,0 @@
-TTS_AUTO_PLAY_TIMEOUT = 5
-
-# sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
-TTS_AUTO_PLAY_YIELD_CPU_TIME = 0.02

api/controllers/service_api/app/audio.py

@@ -151,8 +151,9 @@ class TextApi(Resource):
         tags=["TTS"],
         responses={
             200: (
-                "Returns the generated audio. Generator responses are streamed by the service as `audio/mpeg`; "
-                "otherwise the provider output is returned directly."
+                "Returns the generated audio. The `Content-Type` header reflects the provider audio container, "
+                "verified from the response bytes when recognizable. The binary response can be AAC, FLAC, MP4, "
+                "MP3, Ogg, WAV, or WebM."
             ),
             400: (
                 "- `app_unavailable` : App unavailable or misconfigured.\n"
@@ -165,7 +166,8 @@ class TextApi(Resource):
         },
     )
     @expect_with_user(service_api_ns, TextToAudioPayload)
-    @binary_response(service_api_ns, "audio/mpeg")
+    # OpenAPI 2 cannot express a runtime-selected response MIME type, so document the body as generic binary.
+    @binary_response(service_api_ns, "application/octet-stream")
     @service_api_ns.doc("text_to_audio")
     @service_api_ns.doc(description="Convert text to audio using text-to-speech")
     @service_api_ns.doc(

api/core/app/apps/advanced_chat/generate_task_pipeline.py

@@ -11,7 +11,6 @@ from typing import Any, Union
 from sqlalchemy import select, update
 from sqlalchemy.orm import Session
 
-from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.common.graph_runtime_state_support import GraphRuntimeStateSupport
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
@@ -70,7 +69,7 @@ from core.app.entities.task_entities import (
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
-from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
+from core.base.tts import AppGeneratorTTSPublisher
 from core.db.session_factory import session_factory
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
@@ -349,9 +348,13 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if not publisher:
             return None
         audio_msg = publisher.check_and_get_audio()
-        if audio_msg and isinstance(audio_msg, AudioTrunk) and audio_msg.status != "finish":
-            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
-        return None
+        if audio_msg is None:
+            return None
+        if audio_msg.status == "responding":
+            return MessageAudioStreamResponse(audio=audio_msg.audio, audio_type=audio_msg.audio_type, task_id=task_id)
+        if audio_msg.status in {"finish", "error"}:
+            return None
+        raise RuntimeError(f"TTS publisher returned an unknown status: {audio_msg.status}")
 
     def _wrapper_process_stream_response(
         self, trace_manager: TraceQueueManager | None = None
@@ -362,7 +365,8 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         features_dict = self._workflow_features_dict
 
         if (
-            features_dict.get("text_to_speech")
+            self._base_task_pipeline.stream
+            and features_dict.get("text_to_speech")
             and features_dict["text_to_speech"].get("enabled")
             and features_dict["text_to_speech"].get("autoPlay") == "enabled"
         ):
@@ -370,34 +374,41 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 tenant_id, features_dict["text_to_speech"].get("voice"), features_dict["text_to_speech"].get("language")
             )
 
-        for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
-            while True:
-                audio_response = self._listen_audio_msg(publisher=tts_publisher, task_id=task_id)
-                if audio_response:
+        try:
+            for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
+                while audio_response := self._listen_audio_msg(publisher=tts_publisher, task_id=task_id):
                     yield audio_response
-                else:
-                    break
-            yield response
+                if tts_publisher and isinstance(response, ErrorStreamResponse):
+                    tts_publisher.cancel()
+                    yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                    yield response
+                    return
+                yield response
 
-        start_listener_time = time.time()
-        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
-            try:
-                if not tts_publisher:
-                    break
-                audio_trunk = tts_publisher.check_and_get_audio()
-                if audio_trunk is None:
-                    time.sleep(TTS_AUTO_PLAY_YIELD_CPU_TIME)
+            if tts_publisher is None:
+                return
+
+            tts_publisher.publish(None)
+            while True:
+                audio_trunk = tts_publisher.check_and_get_audio(block=True)
+                assert audio_trunk is not None
+                if audio_trunk.status == "responding":
+                    yield MessageAudioStreamResponse(
+                        audio=audio_trunk.audio, audio_type=audio_trunk.audio_type, task_id=task_id
+                    )
                     continue
-                if audio_trunk.status == "finish":
-                    break
-                else:
-                    start_listener_time = time.time()
-                    yield MessageAudioStreamResponse(audio=audio_trunk.audio, task_id=task_id)
-            except Exception:
-                logger.exception("Failed to listen audio message, task_id: %s", task_id)
-                break
-        if tts_publisher:
-            yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                if audio_trunk.status not in {"finish", "error"}:
+                    raise RuntimeError(f"TTS publisher returned an unknown status: {audio_trunk.status}")
+
+                yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                if audio_trunk.status == "error":
+                    if audio_trunk.error is None:
+                        raise RuntimeError("TTS publisher returned an error terminal without an exception")
+                    yield ErrorStreamResponse(err=audio_trunk.error, task_id=task_id)
+                return
+        finally:
+            if tts_publisher:
+                tts_publisher.cancel()
 
     @contextmanager
     def _database_session(self):
@@ -1042,9 +1053,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                     ):
                         yield from responses
 
-        if tts_publisher:
-            tts_publisher.publish(None)
-
         if self._conversation_name_generate_thread:
             logger.debug("Conversation name generation running as daemon thread")
 

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -6,7 +6,6 @@ from typing import Union
 
 from sqlalchemy.orm import Session, sessionmaker
 
-from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.common.graph_runtime_state_support import GraphRuntimeStateSupport
 from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
@@ -59,7 +58,7 @@ from core.app.entities.task_entities import (
     WorkflowStartStreamResponse,
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
-from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
+from core.base.tts import AppGeneratorTTSPublisher
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.system_variables import build_system_variables
 from extensions.ext_database import db
@@ -246,9 +245,13 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if not publisher:
             return None
         audio_msg = publisher.check_and_get_audio()
-        if audio_msg and isinstance(audio_msg, AudioTrunk) and audio_msg.status != "finish":
-            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
-        return None
+        if audio_msg is None:
+            return None
+        if audio_msg.status == "responding":
+            return MessageAudioStreamResponse(audio=audio_msg.audio, audio_type=audio_msg.audio_type, task_id=task_id)
+        if audio_msg.status in {"finish", "error"}:
+            return None
+        raise RuntimeError(f"TTS publisher returned an unknown status: {audio_msg.status}")
 
     def _wrapper_process_stream_response(
         self, trace_manager: TraceQueueManager | None = None
@@ -259,7 +262,8 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         features_dict = self._workflow_features_dict
 
         if (
-            features_dict.get("text_to_speech")
+            self._base_task_pipeline.stream
+            and features_dict.get("text_to_speech")
             and features_dict["text_to_speech"].get("enabled")
             and features_dict["text_to_speech"].get("autoPlay") == "enabled"
         ):
@@ -267,35 +271,41 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                 tenant_id, features_dict["text_to_speech"].get("voice"), features_dict["text_to_speech"].get("language")
             )
 
-        for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
-            while True:
-                audio_response = self._listen_audio_msg(publisher=tts_publisher, task_id=task_id)
-                if audio_response:
+        try:
+            for response in self._process_stream_response(tts_publisher=tts_publisher, trace_manager=trace_manager):
+                while audio_response := self._listen_audio_msg(publisher=tts_publisher, task_id=task_id):
                     yield audio_response
-                else:
-                    break
-            yield response
+                if tts_publisher and isinstance(response, ErrorStreamResponse):
+                    tts_publisher.cancel()
+                    yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                    yield response
+                    return
+                yield response
 
-        start_listener_time = time.time()
-        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
-            try:
-                if not tts_publisher:
-                    break
-                audio_trunk = tts_publisher.check_and_get_audio()
-                if audio_trunk is None:
-                    # release cpu
-                    # sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
-                    time.sleep(TTS_AUTO_PLAY_YIELD_CPU_TIME)
+            if tts_publisher is None:
+                return
+
+            tts_publisher.publish(None)
+            while True:
+                audio_trunk = tts_publisher.check_and_get_audio(block=True)
+                assert audio_trunk is not None
+                if audio_trunk.status == "responding":
+                    yield MessageAudioStreamResponse(
+                        audio=audio_trunk.audio, audio_type=audio_trunk.audio_type, task_id=task_id
+                    )
                     continue
-                if audio_trunk.status == "finish":
-                    break
-                else:
-                    yield MessageAudioStreamResponse(audio=audio_trunk.audio, task_id=task_id)
-            except Exception:
-                logger.exception("Fails to get audio trunk, task_id: %s", task_id)
-                break
-        if tts_publisher:
-            yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                if audio_trunk.status not in {"finish", "error"}:
+                    raise RuntimeError(f"TTS publisher returned an unknown status: {audio_trunk.status}")
+
+                yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                if audio_trunk.status == "error":
+                    if audio_trunk.error is None:
+                        raise RuntimeError("TTS publisher returned an error terminal without an exception")
+                    yield ErrorStreamResponse(err=audio_trunk.error, task_id=task_id)
+                return
+        finally:
+            if tts_publisher:
+                tts_publisher.cancel()
 
     @contextmanager
     def _database_session(self):
@@ -742,9 +752,6 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
                     ):
                         yield from responses
 
-        if tts_publisher:
-            tts_publisher.publish(None)
-
     def _save_workflow_app_log(self, *, session: Session, workflow_run_id: str | None):
         invoke_from = self._application_generate_entity.invoke_from
         match invoke_from:

api/core/app/entities/task_entities.py

@@ -132,6 +132,7 @@ class MessageAudioStreamResponse(StreamResponse):
 
     event: StreamEvent = StreamEvent.TTS_MESSAGE
     audio: str
+    audio_type: str | None = None
 
 
 class MessageAudioEndStreamResponse(StreamResponse):

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -8,7 +8,6 @@ from typing import Any, cast
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import (
     AgentChatAppGenerateEntity,
@@ -46,7 +45,7 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.app.task_pipeline.message_file_utils import prepare_file_dict
-from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
+from core.base.tts import AppGeneratorTTSPublisher
 from core.db.session_factory import session_factory
 from core.model_manager import ModelInstance
 from core.ops.entities.trace_entity import TraceTaskName
@@ -210,10 +209,13 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
         if publisher is None:
             return None
         audio_msg = publisher.check_and_get_audio()
-        if audio_msg and isinstance(audio_msg, AudioTrunk) and audio_msg.status != "finish":
-            # audio_str = audio_msg.audio.decode('utf-8', errors='ignore')
-            return MessageAudioStreamResponse(audio=audio_msg.audio, task_id=task_id)
-        return None
+        if audio_msg is None:
+            return None
+        if audio_msg.status == "responding":
+            return MessageAudioStreamResponse(audio=audio_msg.audio, audio_type=audio_msg.audio_type, task_id=task_id)
+        if audio_msg.status in {"finish", "error"}:
+            return None
+        raise RuntimeError(f"TTS publisher returned an unknown status: {audio_msg.status}")
 
     def _wrapper_process_stream_response(
         self, trace_manager: TraceQueueManager | None = None
@@ -223,40 +225,47 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
         publisher = None
         text_to_speech_dict = cast(dict[str, Any], self._app_config.app_model_config_dict.get("text_to_speech"))
         if (
-            text_to_speech_dict
+            self.stream
+            and text_to_speech_dict
             and text_to_speech_dict.get("autoPlay") == "enabled"
             and text_to_speech_dict.get("enabled")
         ):
             publisher = AppGeneratorTTSPublisher(
                 tenant_id, text_to_speech_dict.get("voice", ""), text_to_speech_dict.get("language", None)
             )
-        for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
-            while True:
-                audio_response = self._listen_audio_msg(publisher, task_id)
-                if audio_response:
+        try:
+            for response in self._process_stream_response(publisher=publisher, trace_manager=trace_manager):
+                while audio_response := self._listen_audio_msg(publisher, task_id):
                     yield audio_response
-                else:
-                    break
-            yield response
+                if publisher and isinstance(response, ErrorStreamResponse):
+                    publisher.cancel()
+                    yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                    yield response
+                    return
+                yield response
 
-        start_listener_time = time.time()
-        # timeout
-        while (time.time() - start_listener_time) < TTS_AUTO_PLAY_TIMEOUT:
             if publisher is None:
-                break
-            audio = publisher.check_and_get_audio()
-            if audio is None:
-                # release cpu
-                # sleep 20 ms ( 40ms => 1280 byte audio file,20ms => 640 byte audio file)
-                time.sleep(TTS_AUTO_PLAY_YIELD_CPU_TIME)
-                continue
-            if audio.status == "finish":
-                break
-            else:
-                start_listener_time = time.time()
-                yield MessageAudioStreamResponse(audio=audio.audio, task_id=task_id)
-        if publisher:
-            yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                return
+
+            publisher.publish(None)
+            while True:
+                audio = publisher.check_and_get_audio(block=True)
+                assert audio is not None
+                if audio.status == "responding":
+                    yield MessageAudioStreamResponse(audio=audio.audio, audio_type=audio.audio_type, task_id=task_id)
+                    continue
+                if audio.status not in {"finish", "error"}:
+                    raise RuntimeError(f"TTS publisher returned an unknown status: {audio.status}")
+
+                yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
+                if audio.status == "error":
+                    if audio.error is None:
+                        raise RuntimeError("TTS publisher returned an error terminal without an exception")
+                    yield ErrorStreamResponse(err=audio.error, task_id=task_id)
+                return
+        finally:
+            if publisher:
+                publisher.cancel()
 
     def _process_stream_response(
         self, publisher: AppGeneratorTTSPublisher | None, trace_manager: TraceQueueManager | None = None
@@ -380,8 +389,6 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline[EasyUIAppGenerat
                     yield self.ping_stream_response()
                 case _:
                     continue
-        if publisher:
-            publisher.publish(None)
         if self._conversation_name_generate_thread:
             logger.debug("Conversation name generation running as daemon thread")
 

api/core/base/tts/app_generator_tts_publisher.py

@@ -1,10 +1,8 @@
 import base64
-import concurrent.futures
 import logging
 import queue
 import re
 import threading
-from collections.abc import Iterable
 
 from core.app.entities.queue_entities import (
     MessageQueueMessage,
@@ -14,42 +12,23 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
     WorkflowQueueMessage,
 )
-from core.model_manager import ModelInstance, ModelManager
+from core.base.tts.audio_mime import (
+    DEFAULT_TTS_AUDIO_MIME_TYPE,
+    get_model_audio_mime_type,
+    inspect_audio_stream,
+)
+from core.model_manager import ModelManager
 from graphon.model_runtime.entities.message_entities import TextPromptMessageContent
 from graphon.model_runtime.entities.model_entities import ModelType
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
 
 
 class AudioTrunk:
-    def __init__(self, status: str, audio):
+    def __init__(self, status: str, audio, audio_type: str | None = None, error: Exception | None = None):
         self.audio = audio
         self.status = status
-
-
-def _invoice_tts(text_content: str, model_instance: ModelInstance, voice: str):
-    if not text_content or text_content.isspace():
-        return
-    return model_instance.invoke_tts(content_text=text_content.strip(), voice=voice)
-
-
-def _process_future(
-    future_queue: queue.Queue[concurrent.futures.Future[Iterable[bytes] | None] | None],
-    audio_queue: queue.Queue[AudioTrunk],
-):
-    while True:
-        try:
-            future = future_queue.get()
-            if future is None:
-                break
-            invoke_result = future.result()
-            if not invoke_result:
-                continue
-            for audio in invoke_result:
-                audio_base64 = base64.b64encode(bytes(audio))
-                audio_queue.put(AudioTrunk("responding", audio=audio_base64))
-        except Exception as e:
-            logging.getLogger(__name__).warning(e)
-            break
-    audio_queue.put(AudioTrunk("finish", b""))
+        self.audio_type = audio_type
+        self.error = error
 
 
 class AppGeneratorTTSPublisher:
@@ -64,6 +43,7 @@ class AppGeneratorTTSPublisher:
         self.model_instance = self.model_manager.get_default_model_instance(
             tenant_id=self.tenant_id, model_type=ModelType.TTS
         )
+        self._declared_audio_mime_type = get_model_audio_mime_type(self.model_instance)
         self.voices = self.model_instance.get_tts_voices(language=language)
         values = [voice.get("value") for voice in self.voices]
         self.voice = voice
@@ -71,26 +51,30 @@ class AppGeneratorTTSPublisher:
             self.voice = self.voices[0].get("value")
         self.max_sentence = 2
         self._last_audio_event: AudioTrunk | None = None
-        # FIXME better way to handle this threading.start
-        threading.Thread(target=self._runtime).start()
-        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
+        self._cancelled = threading.Event()
+        threading.Thread(target=self._runtime, daemon=True).start()
 
     def publish(self, message: WorkflowQueueMessage | MessageQueueMessage | None, /):
         self._msg_queue.put(message)
 
+    def cancel(self):
+        if self._cancelled.is_set():
+            return
+        self._cancelled.set()
+        self._msg_queue.put(None)
+
     def _runtime(self):
-        future_queue: queue.Queue[concurrent.futures.Future[Iterable[bytes] | None] | None] = queue.Queue()
-        threading.Thread(target=_process_future, args=(future_queue, self._audio_queue)).start()
-        while True:
-            try:
+        audio_type = self._declared_audio_mime_type or DEFAULT_TTS_AUDIO_MIME_TYPE
+        resolved_audio_type: str | None = None
+        try:
+            while True:
                 message = self._msg_queue.get()
+                if self._cancelled.is_set():
+                    return
+                text_content = ""
+                incremental_request = False
                 if message is None:
-                    if self.msg_text and len(self.msg_text.strip()) > 0:
-                        futures_result = self.executor.submit(
-                            _invoice_tts, self.msg_text, self.model_instance, self.voice
-                        )
-                        future_queue.put(futures_result)
-                    break
+                    text_content = self.msg_text
                 else:
                     match message.event:
                         case QueueAgentMessageEvent() | QueueLLMChunkEvent():
@@ -113,34 +97,57 @@ class AppGeneratorTTSPublisher:
                             output = message.event.outputs.get("output", "")
                             if isinstance(output, str):
                                 self.msg_text += output
-                self.last_message = message
-                sentence_arr, text_tmp = self._extract_sentence(self.msg_text)
-                if len(sentence_arr) >= min(self.max_sentence, 7):
-                    self.max_sentence += 1
-                    text_content = "".join(sentence_arr)
-                    futures_result = self.executor.submit(_invoice_tts, text_content, self.model_instance, self.voice)
-                    future_queue.put(futures_result)
-                    if isinstance(text_tmp, str):
+                    sentence_arr, text_tmp = self._extract_sentence(self.msg_text)
+                    if self._declared_audio_mime_type == DEFAULT_TTS_AUDIO_MIME_TYPE and len(sentence_arr) >= min(
+                        self.max_sentence, 7
+                    ):
+                        self.max_sentence += 1
+                        text_content = "".join(sentence_arr)
                         self.msg_text = text_tmp
-                    else:
-                        self.msg_text = ""
+                        incremental_request = True
 
-            except Exception as e:
-                self.logger.warning(e)
-                break
-        future_queue.put(None)
+                if text_content and not text_content.isspace():
+                    invoke_result = self.model_instance.invoke_tts(content_text=text_content.strip(), voice=self.voice)
+                    audio_stream, next_audio_type = inspect_audio_stream(invoke_result, self._declared_audio_mime_type)
+                    if self._cancelled.is_set():
+                        return
+                    if incremental_request and next_audio_type != DEFAULT_TTS_AUDIO_MIME_TYPE:
+                        raise InvokeBadRequestError(
+                            "The TTS model declared MP3 but returned a format that cannot be played incrementally"
+                        )
+                    if resolved_audio_type and resolved_audio_type != next_audio_type:
+                        raise InvokeBadRequestError(
+                            "TTS provider changed MIME type between audio responses: "
+                            f"{resolved_audio_type} then {next_audio_type}"
+                        )
+                    resolved_audio_type = audio_type = next_audio_type
+                    for audio in audio_stream:
+                        # ponytail: reads stop at the next chunk; add transport cancellation when Graphon exposes it.
+                        if self._cancelled.is_set():
+                            return
+                        self._audio_queue.put(
+                            AudioTrunk(
+                                "responding",
+                                audio=base64.b64encode(audio),
+                                audio_type=audio_type,
+                            )
+                        )
 
-    def check_and_get_audio(self):
+                if message is None:
+                    break
+        except Exception as e:
+            self.logger.warning("TTS generation failed", exc_info=True)
+            self._audio_queue.put(AudioTrunk("error", b"", audio_type=audio_type, error=e))
+            return
+
+        self._audio_queue.put(AudioTrunk("finish", b"", audio_type=audio_type))
+
+    def check_and_get_audio(self, *, block: bool = False):
         try:
-            if self._last_audio_event and self._last_audio_event.status == "finish":
-                if self.executor:
-                    self.executor.shutdown(wait=False)
+            if self._last_audio_event and self._last_audio_event.status in {"finish", "error"}:
                 return self._last_audio_event
-            audio = self._audio_queue.get_nowait()
-            if audio and audio.status == "finish":
-                self.executor.shutdown(wait=False)
-            if audio:
-                self._last_audio_event = audio
+            audio = self._audio_queue.get(block=block)
+            self._last_audio_event = audio
             return audio
         except queue.Empty:
             return None

api/core/base/tts/audio_mime.py

@@ -0,0 +1,180 @@
+import logging
+from collections.abc import Generator, Iterable
+from itertools import chain
+from typing import TYPE_CHECKING
+
+from core.plugin.entities.plugin_daemon import TTSAudioChunk
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
+
+if TYPE_CHECKING:
+    from core.model_manager import ModelInstance
+
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TTS_AUDIO_MIME_TYPE = "audio/mpeg"
+_SIGNATURE_SIZE = 32
+_AUDIO_MIME_TYPE_ALIASES = {
+    "mp3": "audio/mpeg",
+    "audio/mp3": "audio/mpeg",
+    "audio/mpeg": "audio/mpeg",
+    "wav": "audio/wav",
+    "wave": "audio/wav",
+    "audio/wav": "audio/wav",
+    "audio/wave": "audio/wav",
+    "audio/x-wav": "audio/wav",
+    "ogg": "audio/ogg",
+    "oga": "audio/ogg",
+    "audio/ogg": "audio/ogg",
+    "flac": "audio/flac",
+    "audio/flac": "audio/flac",
+    "aac": "audio/aac",
+    "audio/aac": "audio/aac",
+    "m4a": "audio/mp4",
+    "mp4": "audio/mp4",
+    "audio/mp4": "audio/mp4",
+    "webm": "audio/webm",
+    "audio/webm": "audio/webm",
+}
+
+
+def normalize_audio_mime_type(audio_type: object | None) -> str | None:
+    """Convert provider audio-type metadata into a browser MIME type."""
+    if not isinstance(audio_type, str):
+        return None
+
+    mime_type = audio_type.split(";", maxsplit=1)[0].strip().lower()
+    return _AUDIO_MIME_TYPE_ALIASES.get(mime_type)
+
+
+def get_model_audio_mime_type(model_instance: "ModelInstance") -> str | None:
+    """Read the model's declared TTS MIME type without making it mandatory."""
+    try:
+        model_schema = model_instance.get_model_schema()
+        audio_type = model_schema.model_properties.get(ModelPropertyKey.AUDIO_TYPE)
+    except Exception:
+        logger.debug("Unable to resolve the declared audio type for the TTS model", exc_info=True)
+        return None
+
+    return normalize_audio_mime_type(audio_type)
+
+
+def sniff_audio_mime_type(audio: bytes | bytearray | memoryview) -> str | None:
+    """Identify common audio containers from their leading bytes."""
+    signature = bytes(audio[:_SIGNATURE_SIZE])
+    if len(signature) >= 12 and signature[:4] == b"RIFF" and signature[8:12] == b"WAVE":
+        return "audio/wav"
+    if signature.startswith(b"OggS"):
+        return "audio/ogg"
+    if signature.startswith(b"fLaC"):
+        return "audio/flac"
+    if signature.startswith(b"\x1aE\xdf\xa3"):
+        return "audio/webm"
+    if len(signature) >= 8 and signature[4:8] == b"ftyp":
+        return "audio/mp4"
+    if len(signature) >= 2 and signature[0] == 0xFF:
+        if signature[1] & 0xF6 == 0xF0:
+            return "audio/aac"
+        if signature[1] & 0xE0 == 0xE0:
+            return "audio/mpeg"
+    return None
+
+
+def _normalize_reported_mime_type(mime_type: object | None, source: str) -> str | None:
+    if mime_type is None:
+        return None
+
+    normalized_mime_type = normalize_audio_mime_type(mime_type)
+    if normalized_mime_type is None:
+        raise InvokeBadRequestError(f"TTS provider returned an unsupported {source} MIME type: {mime_type!r}")
+    return normalized_mime_type
+
+
+def _extract_audio_chunk(chunk: bytes | bytearray | memoryview | TTSAudioChunk) -> tuple[bytes, str | None]:
+    """Accept old byte chunks and the compatibility carrier for current Graphon."""
+    if isinstance(chunk, TTSAudioChunk):
+        return bytes(chunk), chunk.mime_type
+
+    if isinstance(chunk, (bytes, bytearray, memoryview)):
+        return bytes(chunk), None
+
+    raise InvokeBadRequestError("TTS provider returned a chunk that is not audio bytes")
+
+
+def resolve_audio_mime_type(
+    audio: bytes | bytearray | memoryview,
+    declared_mime_type: str | None = None,
+    reported_mime_type: str | None = None,
+) -> str:
+    """Validate the provider MIME against magic bytes and choose the true type.
+
+    ``reported_mime_type`` is emitted with a TTS chunk by current plugin
+    runtimes, and therefore takes precedence over schema metadata. Schema
+    metadata remains a fallback for old plugins that omit the new field.
+    """
+    normalized_declared_mime_type = _normalize_reported_mime_type(declared_mime_type, "schema")
+    normalized_reported_mime_type = _normalize_reported_mime_type(reported_mime_type, "chunk")
+    detected_mime_type = sniff_audio_mime_type(audio)
+
+    expected_mime_type = normalized_reported_mime_type or normalized_declared_mime_type
+    if expected_mime_type and detected_mime_type and expected_mime_type != detected_mime_type:
+        raise InvokeBadRequestError(
+            "TTS provider output MIME does not match its audio bytes: "
+            f"declared {expected_mime_type}, detected {detected_mime_type}"
+        )
+
+    if normalized_reported_mime_type:
+        if normalized_declared_mime_type and normalized_declared_mime_type != normalized_reported_mime_type:
+            logger.info(
+                "TTS chunk MIME %s overrides schema MIME %s",
+                normalized_reported_mime_type,
+                normalized_declared_mime_type,
+            )
+        return normalized_reported_mime_type
+
+    if detected_mime_type:
+        return detected_mime_type
+
+    return normalized_declared_mime_type or DEFAULT_TTS_AUDIO_MIME_TYPE
+
+
+def inspect_audio_stream(
+    audio_stream: Iterable[bytes | bytearray | memoryview | TTSAudioChunk], declared_mime_type: str | None = None
+) -> tuple[Generator[bytes, None, None], str]:
+    """Peek at and validate a TTS stream without dropping its leading bytes."""
+    iterator = iter(audio_stream)
+    leading_chunks: list[bytes] = []
+    signature = bytearray()
+    reported_mime_type: str | None = None
+
+    while len(signature) < _SIGNATURE_SIZE:
+        try:
+            chunk, chunk_mime_type = _extract_audio_chunk(next(iterator))
+        except StopIteration:
+            break
+        normalized_chunk_mime_type = _normalize_reported_mime_type(chunk_mime_type, "chunk")
+        if normalized_chunk_mime_type:
+            if reported_mime_type and reported_mime_type != normalized_chunk_mime_type:
+                raise InvokeBadRequestError(
+                    "TTS provider changed MIME type within one audio response: "
+                    f"{reported_mime_type} then {normalized_chunk_mime_type}"
+                )
+            reported_mime_type = normalized_chunk_mime_type
+        leading_chunks.append(chunk)
+        signature.extend(chunk[: _SIGNATURE_SIZE - len(signature)])
+
+    mime_type = resolve_audio_mime_type(signature, declared_mime_type, reported_mime_type)
+
+    def validated_stream() -> Generator[bytes, None, None]:
+        for chunk in chain(leading_chunks, iterator):
+            audio, chunk_mime_type = _extract_audio_chunk(chunk)
+            normalized_chunk_mime_type = _normalize_reported_mime_type(chunk_mime_type, "chunk")
+            if normalized_chunk_mime_type and normalized_chunk_mime_type != mime_type:
+                raise InvokeBadRequestError(
+                    "TTS provider changed MIME type within one audio response: "
+                    f"{mime_type} then {normalized_chunk_mime_type}"
+                )
+            yield audio
+
+    return validated_stream(), mime_type

api/core/plugin/backwards_invocation/model.py

@@ -3,6 +3,7 @@ from binascii import hexlify, unhexlify
 from collections.abc import Generator
 from typing import Any
 
+from core.base.tts.audio_mime import get_model_audio_mime_type, inspect_audio_stream
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
 from core.model_manager import ModelManager
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
@@ -214,12 +215,11 @@ class PluginModelBackwardsInvocation(BaseBackwardsInvocation):
             model=payload.model,
         )
 
-        # invoke model
-        response = model_instance.invoke_tts(content_text=payload.content_text, voice=payload.voice)
-
         def handle() -> Generator[dict[str, Any], None, None]:
-            for chunk in response:
-                yield {"result": hexlify(chunk).decode("utf-8")}
+            response = model_instance.invoke_tts(content_text=payload.content_text, voice=payload.voice)
+            audio_stream, mime_type = inspect_audio_stream(response, get_model_audio_mime_type(model_instance))
+            for chunk in audio_stream:
+                yield {"result": hexlify(chunk).decode("utf-8"), "mime_type": mime_type}
 
         return handle()
 

api/core/plugin/entities/plugin_daemon.py

@@ -135,6 +135,28 @@ class PluginStringResultResponse(BaseModel):
     result: str = Field(description="The result of the string.")
 
 
+class PluginTTSResultResponse(PluginStringResultResponse):
+    """One TTS data chunk returned by the plugin daemon."""
+
+    mime_type: str | None = Field(default=None, description="The MIME type of the audio chunk.")
+
+
+class TTSAudioChunk(bytes):
+    """A bytes-compatible TTS chunk carrying optional daemon MIME metadata.
+
+    The currently released Graphon runtime exposes TTS output as ``bytes``.
+    This carrier preserves that contract while retaining the daemon field until
+    the structured Graphon ``TTSChunk`` protocol is available in a release.
+    """
+
+    mime_type: str | None
+
+    def __new__(cls, data: bytes | bytearray | memoryview, mime_type: str | None = None):
+        instance = super().__new__(cls, data)
+        instance.mime_type = mime_type
+        return instance
+
+
 class PluginVoiceEntity(BaseModel):
     name: str = Field(description="The name of the voice.")
     value: str = Field(description="The value of the voice.")

api/core/plugin/impl/model.py

@@ -11,7 +11,9 @@ from core.plugin.entities.plugin_daemon import (
     PluginModelSchemaEntity,
     PluginStringResultResponse,
     PluginTextEmbeddingNumTokensResponse,
+    PluginTTSResultResponse,
     PluginVoicesResponse,
+    TTSAudioChunk,
 )
 from core.plugin.impl.base import BasePluginClient
 from core.plugin.impl.exc import PluginInvokeError, PluginLLMPollingUnsupportedError
@@ -580,14 +582,14 @@ class PluginModelClient(BasePluginClient):
         credentials: dict[str, Any],
         content_text: str,
         voice: str,
-    ) -> Generator[bytes, None, None]:
+    ) -> Generator[TTSAudioChunk, None, None]:
         """
         Invoke tts
         """
         response = self._request_with_plugin_daemon_response_stream(
             method="POST",
             path=f"plugin/{tenant_id}/dispatch/tts/invoke",
-            type_=PluginStringResultResponse,
+            type_=PluginTTSResultResponse,
             data=jsonable_encoder(
                 self._dispatch_payload(
                     user_id=user_id,
@@ -611,7 +613,7 @@ class PluginModelClient(BasePluginClient):
         try:
             for result in response:
                 hex_str = result.result
-                yield binascii.unhexlify(hex_str)
+                yield TTSAudioChunk(binascii.unhexlify(hex_str), mime_type=result.mime_type)
         except PluginDaemonInnerError as e:
             raise ValueError(e.message + str(e.code))
 

api/core/tools/builtin_tool/providers/audio/tools/tts.py

@@ -1,9 +1,9 @@
-import io
 from collections.abc import Generator
 from typing import Any, override
 
 from sqlalchemy.orm import Session
 
+from core.base.tts.audio_mime import get_model_audio_mime_type, inspect_audio_stream
 from core.model_manager import ModelManager
 from core.plugin.entities.parameters import PluginParameterOption
 from core.tools.builtin_tool.tool import BuiltinTool
@@ -45,15 +45,12 @@ class TTSTool(BuiltinTool):
             else:
                 raise ValueError("Sorry, no voice available.")
         tts = model_instance.invoke_tts(content_text=tool_parameters.get("text"), voice=voice)  # type: ignore[arg-type]
-        buffer = io.BytesIO()
-        for chunk in tts:
-            buffer.write(chunk)
-
-        wav_bytes = buffer.getvalue()
+        audio_stream, mime_type = inspect_audio_stream(tts, get_model_audio_mime_type(model_instance))
+        audio_bytes = b"".join(audio_stream)
         yield self.create_text_message("Audio generated successfully")
         yield self.create_blob_message(
-            blob=wav_bytes,
-            meta={"mime_type": "audio/x-wav"},
+            blob=audio_bytes,
+            meta={"mime_type": mime_type},
         )
 
     def get_available_models(self) -> list[tuple[str, str, list[Any]]]:

api/openapi/markdown/service-openapi.md

@@ -300,7 +300,7 @@ Convert text to speech.
 
 | Code | Description |
 | ---- | ----------- |
-| 200 | Returns the generated audio. Generator responses are streamed by the service as `audio/mpeg`; otherwise the provider output is returned directly. |
+| 200 | Returns the generated audio. The `Content-Type` header reflects the provider audio container, verified from the response bytes when recognizable. The binary response can be AAC, FLAC, MP4, MP3, Ogg, WAV, or WebM. |
 | 400 | - `app_unavailable` : App unavailable or misconfigured. - `provider_not_initialize` : No valid model provider credentials found. - `provider_quota_exceeded` : Model provider quota exhausted. - `model_currently_not_support` : Current model does not support this operation. - `completion_request_error` : Text-to-speech request failed. |
 | 401 | Unauthorized - invalid API token |
 | 403 | Forbidden - token scope, app, dataset, or workspace access denied |

api/services/audio_service.py

@@ -1,7 +1,7 @@
 import io
 import logging
 import uuid
-from collections.abc import Generator
+from collections.abc import Iterable
 from typing import cast
 
 from flask import Response, stream_with_context
@@ -11,6 +11,7 @@ from werkzeug.datastructures import FileStorage
 
 from constants import AUDIO_EXTENSIONS
 from core.app.apps.agent_app.app_feature_projection import merge_agent_app_features
+from core.base.tts.audio_mime import get_model_audio_mime_type, inspect_audio_stream, resolve_audio_mime_type
 from core.model_manager import ModelManager
 from graphon.model_runtime.entities.model_entities import ModelType
 from models.agent_config_entities import AgentSoulConfig
@@ -37,6 +38,21 @@ _ASR_MIME_TYPE_ALIASES = {
 logger = logging.getLogger(__name__)
 
 
+def _create_tts_response(
+    audio: Iterable[bytes] | bytes | bytearray | memoryview, declared_mime_type: str | None
+) -> Response:
+    """Create a response whose Content-Type matches the returned audio container."""
+    if isinstance(audio, (bytes, bytearray, memoryview)):
+        audio_bytes = bytes(audio)
+        return Response(audio_bytes, content_type=resolve_audio_mime_type(audio_bytes, declared_mime_type))
+
+    audio_stream, mime_type = inspect_audio_stream(audio, declared_mime_type)
+    return Response(
+        stream_with_context(audio_stream),  # pyrefly: ignore[no-matching-overload]
+        content_type=mime_type,
+    )
+
+
 class AudioService:
     @staticmethod
     def _get_message_by_ref(session: Session, message_ref: MessageRef) -> Message | None:
@@ -209,7 +225,10 @@ class AudioService:
                     else:
                         raise ValueError("Sorry, no voice available.")
 
-                return model_instance.invoke_tts(content_text=text_content.strip(), voice=voice)
+                return (
+                    model_instance.invoke_tts(content_text=text_content.strip(), voice=voice),
+                    get_model_audio_mime_type(model_instance),
+                )
             except Exception as e:
                 raise e
 
@@ -225,17 +244,17 @@ class AudioService:
                 return None
 
             else:
-                response = invoke_tts(text_content=message.answer, app_model=app_model, voice=voice, is_draft=is_draft)
-                if isinstance(response, Generator):
-                    return Response(stream_with_context(response), content_type="audio/mpeg")  # type: ignore
-                return response
+                response, declared_mime_type = invoke_tts(
+                    text_content=message.answer, app_model=app_model, voice=voice, is_draft=is_draft
+                )
+                return _create_tts_response(response, declared_mime_type)
         else:
             if text is None:
                 raise ValueError("Text is required")
-            response = invoke_tts(text_content=text, app_model=app_model, voice=voice, is_draft=is_draft)
-            if isinstance(response, Generator):
-                return Response(stream_with_context(response), content_type="audio/mpeg")  # type: ignore
-            return response
+            response, declared_mime_type = invoke_tts(
+                text_content=text, app_model=app_model, voice=voice, is_draft=is_draft
+            )
+            return _create_tts_response(response, declared_mime_type)
 
     @classmethod
     def transcript_tts_voices(cls, tenant_id: str, language: str):

api/tests/test_containers_integration_tests/services/test_audio_service_db.py

@@ -168,7 +168,9 @@ class TestAudioServiceTranscriptTTSMessageLookup:
                 voice="en-US-Neural",
             )
 
-        assert result == b"audio from message"
+        assert result is not None
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"audio from message"
         mock_model_instance.invoke_tts.assert_called_once_with(
             content_text="Hello from message",
             voice="en-US-Neural",

api/tests/unit_tests/controllers/test_swagger.py

@@ -370,7 +370,7 @@ def test_service_openapi_documents_non_json_response_media_types():
         "text/event-stream",
     }
     assert _response_content_types(paths["/workflow/{task_id}/events"]["get"]) == {"text/event-stream"}
-    assert _response_content_types(paths["/text-to-audio"]["post"]) == {"audio/mpeg"}
+    assert _response_content_types(paths["/text-to-audio"]["post"]) == {"application/octet-stream"}
     assert _response_content_types(paths["/files/{file_id}/preview"]["get"]) == {
         "application/octet-stream",
         "application/pdf",

api/tests/unit_tests/core/app/apps/advanced_chat/test_generate_task_pipeline_core.py

@@ -273,7 +273,7 @@ class TestAdvancedChatGenerateTaskPipeline:
 
     def test_listen_audio_msg_returns_audio_stream(self):
         pipeline = _make_pipeline()
-        publisher = SimpleNamespace(check_and_get_audio=lambda: AudioTrunk(status="stream", audio="data"))
+        publisher = SimpleNamespace(check_and_get_audio=lambda: AudioTrunk(status="responding", audio="data"))
 
         response = pipeline._listen_audio_msg(publisher=publisher, task_id="task")
 

api/tests/unit_tests/core/app/apps/workflow/test_generate_task_pipeline_core.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import json
-import logging
 from types import SimpleNamespace
 
 import pytest
@@ -196,7 +195,7 @@ class TestWorkflowGenerateTaskPipeline:
 
     def test_listen_audio_msg_returns_audio_stream(self):
         pipeline = _make_pipeline()
-        publisher = SimpleNamespace(check_and_get_audio=lambda: AudioTrunk(status="stream", audio="data"))
+        publisher = SimpleNamespace(check_and_get_audio=lambda: AudioTrunk(status="responding", audio="data"))
 
         response = pipeline._listen_audio_msg(publisher=publisher, task_id="task")
 
@@ -477,6 +476,7 @@ class TestWorkflowGenerateTaskPipeline:
 
     def test_wrapper_process_stream_response_emits_audio_end(self, monkeypatch: pytest.MonkeyPatch):
         pipeline = _make_pipeline()
+        pipeline._base_task_pipeline.stream = True
         pipeline._workflow_features_dict = {
             "text_to_speech": {"enabled": True, "autoPlay": "enabled", "voice": "v", "language": "en"}
         }
@@ -486,17 +486,23 @@ class TestWorkflowGenerateTaskPipeline:
             def __init__(self, *args, **kwargs):
                 self.calls = 0
 
-            def check_and_get_audio(self):
+            def check_and_get_audio(self, *, block=False):
                 self.calls += 1
                 if self.calls == 1:
-                    return AudioTrunk(status="stream", audio="data")
+                    assert not block
+                    return AudioTrunk(status="responding", audio="data")
                 if self.calls == 2:
+                    assert not block
                     return None
+                assert block
                 return AudioTrunk(status="finish", audio="")
 
             def publish(self, message):
                 return None
 
+            def cancel(self):
+                return None
+
         monkeypatch.setattr(
             "core.app.apps.workflow.generate_task_pipeline.AppGeneratorTTSPublisher",
             _Publisher,
@@ -626,33 +632,30 @@ class TestWorkflowGenerateTaskPipeline:
         responses = list(pipeline._wrapper_process_stream_response())
         assert responses == [PingStreamResponse(task_id="task")]
 
-    def test_wrapper_process_stream_response_final_audio_none_then_finish(self, monkeypatch: pytest.MonkeyPatch):
+    def test_wrapper_process_stream_response_uses_a_blocking_terminal_read(self, monkeypatch: pytest.MonkeyPatch):
         pipeline = _make_pipeline()
+        pipeline._base_task_pipeline.stream = True
         pipeline._workflow_features_dict = {
             "text_to_speech": {"enabled": True, "autoPlay": "enabled", "voice": "v", "language": "en"}
         }
         pipeline._process_stream_response = lambda **kwargs: iter([])
 
-        sleep_spy = []
+        blocking_reads = []
 
         class _Publisher:
             def __init__(self, *args, **kwargs):
-                self.calls = 0
+                pass
 
-            def check_and_get_audio(self):
-                self.calls += 1
-                if self.calls == 1:
-                    return None
+            def check_and_get_audio(self, *, block=False):
+                blocking_reads.append(block)
                 return AudioTrunk(status="finish", audio="")
 
             def publish(self, message):
                 _ = message
 
-        time_values = iter([0.0, 0.0, 0.2])
-        monkeypatch.setattr("core.app.apps.workflow.generate_task_pipeline.time.time", lambda: next(time_values))
-        monkeypatch.setattr(
-            "core.app.apps.workflow.generate_task_pipeline.time.sleep", lambda _: sleep_spy.append(True)
-        )
+            def cancel(self):
+                return None
+
         monkeypatch.setattr(
             "core.app.apps.workflow.generate_task_pipeline.AppGeneratorTTSPublisher",
             _Publisher,
@@ -660,13 +663,14 @@ class TestWorkflowGenerateTaskPipeline:
 
         responses = list(pipeline._wrapper_process_stream_response())
 
-        assert sleep_spy
+        assert blocking_reads == [True]
         assert any(isinstance(item, MessageAudioEndStreamResponse) for item in responses)
 
-    def test_wrapper_process_stream_response_handles_audio_exception(
-        self, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+    def test_wrapper_process_stream_response_does_not_swallow_audio_queue_exceptions(
+        self, monkeypatch: pytest.MonkeyPatch
     ):
         pipeline = _make_pipeline()
+        pipeline._base_task_pipeline.stream = True
         pipeline._workflow_features_dict = {
             "text_to_speech": {"enabled": True, "autoPlay": "enabled", "voice": "v", "language": "en"}
         }
@@ -674,28 +678,25 @@ class TestWorkflowGenerateTaskPipeline:
 
         class _Publisher:
             def __init__(self, *args, **kwargs):
-                self.called = False
+                pass
 
-            def check_and_get_audio(self):
-                if not self.called:
-                    self.called = True
-                    raise RuntimeError("tts failure")
-                return AudioTrunk(status="finish", audio="")
+            def check_and_get_audio(self, *, block=False):
+                assert block
+                raise RuntimeError("tts failure")
 
             def publish(self, message):
                 _ = message
 
-        monkeypatch.setattr("core.app.apps.workflow.generate_task_pipeline.time.time", lambda: 0.0)
+            def cancel(self):
+                return None
+
         monkeypatch.setattr(
             "core.app.apps.workflow.generate_task_pipeline.AppGeneratorTTSPublisher",
             _Publisher,
         )
 
-        with caplog.at_level(logging.ERROR, logger="core.app.apps.workflow.generate_task_pipeline"):
-            responses = list(pipeline._wrapper_process_stream_response())
-
-        assert "Fails to get audio trunk, task_id: task" in caplog.messages
-        assert any(isinstance(item, MessageAudioEndStreamResponse) for item in responses)
+        with pytest.raises(RuntimeError, match="tts failure"):
+            list(pipeline._wrapper_process_stream_response())
 
     @pytest.mark.parametrize("sqlite_session", [(WorkflowAppLog,)], indirect=True)
     def test_database_session_rolls_back_on_error(
@@ -856,7 +857,7 @@ class TestWorkflowGenerateTaskPipeline:
 
         responses = list(pipeline._process_stream_response(tts_publisher=_Publisher()))
         assert responses == ["started", "text", "dispatched", "error"]
-        assert publisher_calls == [None]
+        assert publisher_calls == []
 
     def test_process_stream_response_break_paths(self):
         pipeline = _make_pipeline()

api/tests/unit_tests/core/app/task_pipeline/test_easy_ui_based_generate_task_pipeline.py

@@ -361,10 +361,7 @@ class TestEasyUIBasedGenerateTaskPipelineProcessStreamResponse:
         list(pipeline._process_stream_response(publisher=publisher, trace_manager=None))
 
         # Assert
-        # Called once with message and once with None at the end
-        assert publisher.publish.call_count == 2
-        publisher.publish.assert_any_call(mock_queue_message)
-        publisher.publish.assert_any_call(None)
+        publisher.publish.assert_called_once_with(mock_queue_message)
 
     def test_trace_manager_passed_to_save_message(self, pipeline, committed_sessions):
         """Test that trace manager is passed to _save_message."""

api/tests/unit_tests/core/app/task_pipeline/test_easy_ui_based_generate_task_pipeline_core.py

@@ -1026,9 +1026,16 @@ class TestEasyUiBasedGenerateTaskPipeline:
         )
 
         class _Publisher:
-            def check_and_get_audio(self):
+            def check_and_get_audio(self, *, block=False):
+                assert block
                 return AudioTrunk("finish", "")
 
+            def publish(self, message):
+                assert message is None
+
+            def cancel(self):
+                return None
+
         inline_audio = MessageAudioStreamResponse(task_id="task", audio="inline")
         audio_calls = iter([inline_audio, None])
         payload = PingStreamResponse(task_id="task")
@@ -1044,8 +1051,9 @@ class TestEasyUiBasedGenerateTaskPipeline:
         assert responses[0] == inline_audio
         assert responses[1] == payload
         assert isinstance(responses[-1], MessageAudioEndStreamResponse)
+        assert "audio_type" not in responses[-1].model_dump()
 
-    def test_wrapper_process_stream_response_timeout_yields_audio_chunk(self, monkeypatch: pytest.MonkeyPatch):
+    def test_wrapper_process_stream_response_waits_for_the_audio_terminal(self, monkeypatch: pytest.MonkeyPatch):
         conversation = _make_conversation(AppMode.CHAT)
         message = _make_message()
         entity = _make_entity(ChatAppGenerateEntity, AppMode.CHAT)
@@ -1062,30 +1070,99 @@ class TestEasyUiBasedGenerateTaskPipeline:
 
         class _Publisher:
             def __init__(self):
-                self._events = iter([None, AudioTrunk("responding", "later"), AudioTrunk("finish", "")])
+                self._events = iter([AudioTrunk("responding", "later"), AudioTrunk("finish", "")])
 
-            def check_and_get_audio(self):
+            def check_and_get_audio(self, *, block=False):
+                assert block
                 return next(self._events)
 
-        clock = {"value": 0.0}
+            def publish(self, message):
+                assert message is None
 
-        def _fake_time():
-            clock["value"] += 0.1
-            return clock["value"]
+            def cancel(self):
+                return None
 
         _set_method(pipeline, "_process_stream_response", lambda publisher, trace_manager: iter([]))
         monkeypatch.setattr(
             "core.app.task_pipeline.easy_ui_based_generate_task_pipeline.AppGeneratorTTSPublisher",
             lambda tenant_id, voice, language: _Publisher(),
         )
-        monkeypatch.setattr("core.app.task_pipeline.easy_ui_based_generate_task_pipeline.time.time", _fake_time)
-        monkeypatch.setattr("core.app.task_pipeline.easy_ui_based_generate_task_pipeline.time.sleep", lambda _: None)
-
         responses = list(pipeline._wrapper_process_stream_response())
 
         assert any(isinstance(item, MessageAudioStreamResponse) for item in responses)
         assert isinstance(responses[-1], MessageAudioEndStreamResponse)
 
+    def test_wrapper_process_stream_response_ends_audio_before_reporting_tts_error(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        conversation = _make_conversation(AppMode.CHAT)
+        message = _make_message()
+        entity = _make_entity(ChatAppGenerateEntity, AppMode.CHAT)
+        entity.app_config.app_model_config_dict = {
+            "text_to_speech": {"autoPlay": "enabled", "enabled": True, "voice": "v", "language": "en"}
+        }
+        pipeline = EasyUIBasedGenerateTaskPipeline(
+            application_generate_entity=entity,
+            queue_manager=_FakeQueueManager(),
+            conversation=conversation,
+            message=message,
+            stream=True,
+        )
+        error = RuntimeError("tts failed")
+
+        class _Publisher:
+            def check_and_get_audio(self, *, block=False):
+                assert block
+                return AudioTrunk("error", b"", error=error)
+
+            def publish(self, message):
+                assert message is None
+
+            def cancel(self):
+                return None
+
+        _set_method(pipeline, "_process_stream_response", lambda publisher, trace_manager: iter([]))
+        monkeypatch.setattr(
+            "core.app.task_pipeline.easy_ui_based_generate_task_pipeline.AppGeneratorTTSPublisher",
+            lambda tenant_id, voice, language: _Publisher(),
+        )
+
+        responses = list(pipeline._wrapper_process_stream_response())
+
+        assert isinstance(responses[-2], MessageAudioEndStreamResponse)
+        assert isinstance(responses[-1], ErrorStreamResponse)
+        assert responses[-1].err is error
+
+    def test_wrapper_process_stream_response_ends_audio_before_a_main_error(self, monkeypatch: pytest.MonkeyPatch):
+        conversation = _make_conversation(AppMode.CHAT)
+        message = _make_message()
+        entity = _make_entity(ChatAppGenerateEntity, AppMode.CHAT)
+        entity.app_config.app_model_config_dict = {
+            "text_to_speech": {"autoPlay": "enabled", "enabled": True, "voice": "v", "language": "en"}
+        }
+        pipeline = EasyUIBasedGenerateTaskPipeline(
+            application_generate_entity=entity,
+            queue_manager=_FakeQueueManager(),
+            conversation=conversation,
+            message=message,
+            stream=True,
+        )
+        publisher = Mock()
+        publisher.check_and_get_audio.return_value = None
+        error = ErrorStreamResponse(task_id="task", err=RuntimeError("generation failed"))
+        _set_method(pipeline, "_process_stream_response", lambda publisher, trace_manager: iter([error]))
+        monkeypatch.setattr(
+            "core.app.task_pipeline.easy_ui_based_generate_task_pipeline.AppGeneratorTTSPublisher",
+            lambda tenant_id, voice, language: publisher,
+        )
+
+        responses = list(pipeline._wrapper_process_stream_response())
+
+        assert isinstance(responses[-2], MessageAudioEndStreamResponse)
+        assert responses[-1] is error
+        assert publisher.cancel.called
+        publisher.publish.assert_not_called()
+
     def test_process_stream_response_handles_stop_event_and_output_replacement(self, monkeypatch: pytest.MonkeyPatch):
         conversation = _make_conversation(AppMode.CHAT)
         message = _make_message()

api/tests/unit_tests/core/base/test_app_generator_tts_publisher.py

@@ -1,20 +1,15 @@
 import base64
-import queue
+from types import SimpleNamespace
 from unittest.mock import MagicMock
 
 import pytest
 from pytest_mock import MockerFixture
 
-from core.base.tts.app_generator_tts_publisher import (
-    AppGeneratorTTSPublisher,
-    AudioTrunk,
-    _invoice_tts,
-    _process_future,
-)
-
-# =========================
-# Fixtures
-# =========================
+from core.app.entities.queue_entities import QueueNodeSucceededEvent, QueueTextChunkEvent
+from core.base.tts.app_generator_tts_publisher import AppGeneratorTTSPublisher, AudioTrunk
+from core.plugin.entities.plugin_daemon import TTSAudioChunk
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
 
 
 @pytest.fixture
@@ -22,11 +17,12 @@ def mock_model_instance(mocker: MockerFixture):
     model = mocker.MagicMock()
     model.invoke_tts.return_value = [b"audio1", b"audio2"]
     model.get_tts_voices.return_value = [{"value": "voice1"}, {"value": "voice2"}]
+    model.get_model_schema.return_value = SimpleNamespace(model_properties={ModelPropertyKey.AUDIO_TYPE: "mp3"})
     return model
 
 
 @pytest.fixture
-def mock_model_manager(mocker, mock_model_instance):
+def mock_model_manager(mocker: MockerFixture, mock_model_instance: MagicMock):
     manager = mocker.MagicMock()
     manager.get_default_model_instance.return_value = mock_model_instance
     mocker.patch("core.base.tts.app_generator_tts_publisher.ModelManager.for_tenant", return_value=manager)
@@ -35,158 +31,88 @@ def mock_model_manager(mocker, mock_model_instance):
 
 @pytest.fixture(autouse=True)
 def patch_threads(mocker: MockerFixture):
-    """Prevent real threads from starting during tests"""
+    """Run the worker explicitly in tests."""
     mocker.patch("threading.Thread.start", return_value=None)
 
 
-# =========================
-# AudioTrunk Tests
-# =========================
+def _text_event(text: str) -> MagicMock:
+    event = MagicMock()
+    event.event = MagicMock(spec=QueueTextChunkEvent)
+    event.event.text = text
+    return event
+
+
+def _run(publisher: AppGeneratorTTSPublisher, *messages: MagicMock) -> None:
+    for message in messages:
+        publisher._msg_queue.put(message)
+    publisher._msg_queue.put(None)
+    publisher._runtime()
 
 
 class TestAudioTrunk:
-    def test_audio_trunk_initialization(self):
-        trunk = AudioTrunk("responding", b"data")
-        assert trunk.status == "responding"
-        assert trunk.audio == b"data"
+    def test_initialization(self):
+        error = RuntimeError("failed")
+        trunk = AudioTrunk("error", b"", error=error)
 
-
-# =========================
-# _invoice_tts Tests
-# =========================
-
-
-class TestInvoiceTTS:
-    @pytest.mark.parametrize(
-        "text",
-        [None, "", "   "],
-    )
-    def test_invoice_tts_empty_or_none_returns_none(self, text, mock_model_instance):
-        result = _invoice_tts(text, mock_model_instance, "voice1")
-        assert result is None
-        mock_model_instance.invoke_tts.assert_not_called()
-
-    def test_invoice_tts_valid_text(self, mock_model_instance):
-        result = _invoice_tts(" hello ", mock_model_instance, "voice1")
-        mock_model_instance.invoke_tts.assert_called_once_with(
-            content_text="hello",
-            voice="voice1",
-        )
-        assert result == [b"audio1", b"audio2"]
-
-
-# =========================
-# _process_future Tests
-# =========================
-
-
-class TestProcessFuture:
-    def test_process_future_normal_flow(self):
-        future_queue = queue.Queue()
-        audio_queue = queue.Queue()
-
-        future = MagicMock()
-        future.result.return_value = [b"abc"]
-
-        future_queue.put(future)
-        future_queue.put(None)
-
-        _process_future(future_queue, audio_queue)
-
-        first = audio_queue.get()
-        assert first.status == "responding"
-        assert first.audio == base64.b64encode(b"abc")
-
-        finish = audio_queue.get()
-        assert finish.status == "finish"
-
-    def test_process_future_empty_result(self):
-        future_queue = queue.Queue()
-        audio_queue = queue.Queue()
-
-        future = MagicMock()
-        future.result.return_value = None
-
-        future_queue.put(future)
-        future_queue.put(None)
-
-        _process_future(future_queue, audio_queue)
-
-        finish = audio_queue.get()
-        assert finish.status == "finish"
-
-    def test_process_future_exception(self, mocker: MockerFixture):
-        future_queue = queue.Queue()
-        audio_queue = queue.Queue()
-
-        future = MagicMock()
-        future.result.side_effect = Exception("error")
-
-        future_queue.put(future)
-
-        _process_future(future_queue, audio_queue)
-
-        finish = audio_queue.get()
-        assert finish.status == "finish"
-
-
-# =========================
-# AppGeneratorTTSPublisher Tests
-# =========================
+        assert trunk.status == "error"
+        assert trunk.audio == b""
+        assert trunk.error is error
 
 
 class TestAppGeneratorTTSPublisher:
     def test_initialization_valid_voice(self, mock_model_manager):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+
         assert publisher.voice == "voice1"
         assert publisher.max_sentence == 2
         assert publisher.msg_text == ""
 
     def test_initialization_invalid_voice_fallback(self, mock_model_manager):
         publisher = AppGeneratorTTSPublisher("tenant", "invalid_voice")
+
         assert publisher.voice == "voice1"
 
     def test_publish_puts_message_in_queue(self, mock_model_manager):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
         message = MagicMock()
+
         publisher.publish(message)
+
         assert publisher._msg_queue.get() == message
 
-    def test_check_and_get_audio_no_audio(self, mock_model_manager):
+    def test_cancel_discards_queued_text(self, mock_model_manager, mock_model_instance: MagicMock):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        result = publisher.check_and_get_audio()
-        assert result is None
+        publisher.publish(_text_event("First. Second."))
 
-    def test_check_and_get_audio_non_finish_event(self, mock_model_manager):
+        publisher.cancel()
+        publisher._runtime()
+
+        mock_model_instance.invoke_tts.assert_not_called()
+        assert publisher._audio_queue.empty()
+
+    def test_check_and_get_audio_returns_none_without_audio(self, mock_model_manager):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        trunk = AudioTrunk("responding", b"abc")
+
+        assert publisher.check_and_get_audio() is None
+
+    def test_check_and_get_audio_returns_the_resolved_mime_type(self, mock_model_manager):
+        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+        trunk = AudioTrunk("responding", b"abc", audio_type="audio/wav")
         publisher._audio_queue.put(trunk)
 
         result = publisher.check_and_get_audio()
 
-        assert result.status == "responding"
-        assert publisher._last_audio_event == trunk
+        assert result is trunk
+        assert result.audio_type == "audio/wav"
 
-    def test_check_and_get_audio_finish_event(self, mock_model_manager):
+    @pytest.mark.parametrize("status", ["finish", "error"])
+    def test_check_and_get_audio_caches_terminal_events(self, mock_model_manager, status: str):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-        finish_trunk = AudioTrunk("finish", b"")
-        publisher._audio_queue.put(finish_trunk)
+        terminal = AudioTrunk(status, b"", error=RuntimeError("failed") if status == "error" else None)
+        publisher._audio_queue.put(terminal)
 
-        result = publisher.check_and_get_audio()
-
-        assert result.status == "finish"
-        publisher.executor.shutdown.assert_called_once()
-
-    def test_check_and_get_audio_cached_finish(self, mock_model_manager):
-        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-        publisher._last_audio_event = AudioTrunk("finish", b"")
-
-        result = publisher.check_and_get_audio()
-
-        assert result.status == "finish"
-        publisher.executor.shutdown.assert_called_once()
+        assert publisher.check_and_get_audio(block=True) is terminal
+        assert publisher.check_and_get_audio() is terminal
 
     @pytest.mark.parametrize(
         ("text", "expected_sentences", "expected_remaining"),
@@ -199,188 +125,160 @@ class TestAppGeneratorTTSPublisher:
     )
     def test_extract_sentence(self, mock_model_manager, text, expected_sentences, expected_remaining):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+
         sentences, remaining = publisher._extract_sentence(text)
+
         assert sentences == expected_sentences
         assert remaining == expected_remaining
 
-    def test_runtime_handles_none_message_with_buffer(self, mock_model_manager):
+    def test_runtime_generates_the_final_buffer(self, mock_model_manager, mock_model_instance: MagicMock):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-        publisher.msg_text = "Hello."
+        publisher.msg_text = " Hello. "
 
-        publisher._msg_queue.put(None)
-        publisher._runtime()
+        _run(publisher)
 
-        publisher.executor.submit.assert_called_once()
+        mock_model_instance.invoke_tts.assert_called_once_with(content_text="Hello.", voice="voice1")
+        assert publisher._audio_queue.get().audio == base64.b64encode(b"audio1")
+        assert publisher._audio_queue.get().audio == base64.b64encode(b"audio2")
+        assert publisher._audio_queue.get().status == "finish"
 
-    def test_runtime_handles_none_message_without_buffer(self, mock_model_manager):
+    def test_runtime_skips_an_empty_final_buffer(self, mock_model_manager, mock_model_instance: MagicMock):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
         publisher.msg_text = "   "
 
-        publisher._msg_queue.put(None)
-        publisher._runtime()
+        _run(publisher)
 
-        publisher.executor.submit.assert_not_called()
+        mock_model_instance.invoke_tts.assert_not_called()
+        assert publisher._audio_queue.get().status == "finish"
 
-    def test_runtime_sentence_threshold_triggers_submit(self, mock_model_manager, mocker: MockerFixture):
+    def test_runtime_generates_incremental_mp3_after_the_sentence_threshold(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
 
-        # Force sentence extraction to hit threshold condition
-        mocker.patch.object(
-            publisher,
-            "_extract_sentence",
-            return_value=(["Hello world.", " Second sentence."], ""),
+        _run(publisher, _text_event("Hello world. Second sentence."))
+
+        mock_model_instance.invoke_tts.assert_called_once_with(
+            content_text="Hello world. Second sentence.", voice="voice1"
         )
 
-        from core.app.entities.queue_entities import QueueTextChunkEvent
+    def test_runtime_waits_for_terminal_when_the_schema_has_no_audio_type(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
+        mock_model_instance.get_model_schema.return_value = SimpleNamespace(model_properties={})
+        wav = b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data"
+        mock_model_instance.invoke_tts.return_value = [TTSAudioChunk(wav, "audio/wav")]
+        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+        event = _text_event("Hello. World.")
+        messages = iter([event, None])
 
-        event = MagicMock()
-        event.event = MagicMock(spec=QueueTextChunkEvent)
-        event.event.text = "Hello world. Second sentence."
+        def get_message():
+            message = next(messages)
+            if message is None:
+                mock_model_instance.invoke_tts.assert_not_called()
+            return message
 
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
+        publisher._msg_queue = MagicMock()
+        publisher._msg_queue.get.side_effect = get_message
 
         publisher._runtime()
 
-        assert publisher.executor.submit.called
+        mock_model_instance.invoke_tts.assert_called_once_with(content_text="Hello. World.", voice="voice1")
+        assert publisher._audio_queue.get().audio_type == "audio/wav"
+        assert publisher._audio_queue.get().status == "finish"
 
-    def test_runtime_handles_text_chunk_event(self, mock_model_manager):
+    def test_runtime_waits_for_terminal_when_the_model_declares_wav(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
+        mock_model_instance.get_model_schema.return_value = SimpleNamespace(
+            model_properties={ModelPropertyKey.AUDIO_TYPE: "wav"}
+        )
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
 
-        from core.app.entities.queue_entities import QueueTextChunkEvent
+        _run(publisher, _text_event("Hello. World."))
 
-        event = MagicMock()
-        event.event = MagicMock(spec=QueueTextChunkEvent)
-        event.event.text = "Hello world."
+        mock_model_instance.invoke_tts.assert_called_once_with(content_text="Hello. World.", voice="voice1")
 
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
-
-        publisher._runtime()
-
-        assert publisher.executor.submit.called
-
-    def test_runtime_handles_node_succeeded_event_with_output(self, mock_model_manager):
+    def test_runtime_rejects_a_non_mp3_incremental_response_before_emitting_audio(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
+        wav = b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data"
+        mock_model_instance.invoke_tts.return_value = [TTSAudioChunk(wav, "audio/wav")]
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
 
-        from core.app.entities.queue_entities import QueueNodeSucceededEvent
+        _run(publisher, _text_event("First. Second. tail"))
 
+        terminal = publisher._audio_queue.get()
+        assert terminal.status == "error"
+        assert isinstance(terminal.error, InvokeBadRequestError)
+        assert publisher._audio_queue.empty()
+
+    def test_runtime_rejects_mime_changes_between_audio_responses(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
+        mp3 = b"\xff\xfb" + b"\x00" * 30
+        wav = b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data"
+        mock_model_instance.invoke_tts.side_effect = [
+            [TTSAudioChunk(mp3, "audio/mpeg")],
+            [TTSAudioChunk(wav, "audio/wav")],
+        ]
+        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+
+        _run(publisher, _text_event("First. Second. tail"))
+
+        responding = publisher._audio_queue.get()
+        terminal = publisher._audio_queue.get()
+        assert responding.status == "responding"
+        assert terminal.status == "error"
+        assert isinstance(terminal.error, InvokeBadRequestError)
+        assert "between audio responses" in str(terminal.error)
+        assert publisher._audio_queue.empty()
+
+    def test_runtime_turns_a_lazy_provider_failure_into_an_error_terminal(
+        self, mock_model_manager, mock_model_instance: MagicMock
+    ):
+        def failing_stream():
+            raise RuntimeError("provider failed")
+            yield b""  # pragma: no cover
+
+        mock_model_instance.invoke_tts.return_value = failing_stream()
+        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
+        publisher.msg_text = "Hello"
+
+        _run(publisher)
+
+        terminal = publisher._audio_queue.get()
+        assert terminal.status == "error"
+        assert isinstance(terminal.error, RuntimeError)
+        assert publisher._audio_queue.empty()
+
+    def test_runtime_handles_node_succeeded_output(self, mock_model_manager, mock_model_instance: MagicMock):
+        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
         event = MagicMock()
         event.event = MagicMock(spec=QueueNodeSucceededEvent)
         event.event.outputs = {"output": "Hello world."}
 
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
+        _run(publisher, event)
 
-        publisher._runtime()
+        mock_model_instance.invoke_tts.assert_called_once()
 
-        assert publisher.executor.submit.called
-
-    def test_runtime_handles_node_succeeded_event_without_output(self, mock_model_manager):
+    def test_runtime_ignores_node_succeeded_without_output(self, mock_model_manager, mock_model_instance: MagicMock):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-
-        from core.app.entities.queue_entities import QueueNodeSucceededEvent
-
         event = MagicMock()
         event.event = MagicMock(spec=QueueNodeSucceededEvent)
         event.event.outputs = None
 
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
+        _run(publisher, event)
 
-        publisher._runtime()
+        mock_model_instance.invoke_tts.assert_not_called()
 
-        publisher.executor.submit.assert_not_called()
-
-    def test_runtime_handles_agent_message_event_list_content(self, mock_model_manager, mocker: MockerFixture):
-        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-
-        from core.app.entities.queue_entities import QueueAgentMessageEvent
-        from graphon.model_runtime.entities.llm_entities import LLMResultChunk, LLMResultChunkDelta
-        from graphon.model_runtime.entities.message_entities import (
-            AssistantPromptMessage,
-            ImagePromptMessageContent,
-            TextPromptMessageContent,
-        )
-
-        chunk = LLMResultChunk(
-            model="model",
-            delta=LLMResultChunkDelta(
-                index=0,
-                message=AssistantPromptMessage(
-                    content=[
-                        TextPromptMessageContent(data="Hello "),
-                        ImagePromptMessageContent(format="png", mime_type="image/png", base64_data="a"),
-                    ]
-                ),
-            ),
-        )
-        event = MagicMock(event=QueueAgentMessageEvent(chunk=chunk))
-
-        mocker.patch.object(publisher, "_extract_sentence", return_value=([], ""))
-
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
-
-        publisher._runtime()
-
-        assert publisher.msg_text == "Hello "
-
-    def test_runtime_handles_agent_message_event_empty_content(self, mock_model_manager, mocker: MockerFixture):
-        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-
-        from core.app.entities.queue_entities import QueueAgentMessageEvent
-        from graphon.model_runtime.entities.llm_entities import LLMResultChunk, LLMResultChunkDelta
-        from graphon.model_runtime.entities.message_entities import AssistantPromptMessage
-
-        chunk = LLMResultChunk(
-            model="model",
-            delta=LLMResultChunkDelta(
-                index=0,
-                message=AssistantPromptMessage(content=""),
-            ),
-        )
-        event = MagicMock(event=QueueAgentMessageEvent(chunk=chunk))
-
-        mocker.patch.object(publisher, "_extract_sentence", return_value=([], ""))
-
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
-
-        publisher._runtime()
-
-        assert publisher.msg_text == ""
-
-    def test_runtime_resets_msg_text_when_text_tmp_not_str(self, mock_model_manager, mocker: MockerFixture):
-        publisher = AppGeneratorTTSPublisher("tenant", "voice1")
-        publisher.executor = MagicMock()
-
-        from core.app.entities.queue_entities import QueueTextChunkEvent
-
-        event = MagicMock()
-        event.event = MagicMock(spec=QueueTextChunkEvent)
-        event.event.text = "Hello world. Another sentence."
-
-        mocker.patch.object(publisher, "_extract_sentence", return_value=(["A.", "B."], None))
-
-        publisher._msg_queue.put(event)
-        publisher._msg_queue.put(None)
-
-        publisher._runtime()
-
-        assert publisher.msg_text == ""
-
-    def test_runtime_exception_path(self, mock_model_manager):
+    def test_runtime_turns_message_processing_failure_into_an_error_terminal(self, mock_model_manager):
         publisher = AppGeneratorTTSPublisher("tenant", "voice1")
         publisher._msg_queue = MagicMock()
-        publisher._msg_queue.get.side_effect = Exception("error")
+        publisher._msg_queue.get.side_effect = RuntimeError("failed")
 
         publisher._runtime()
+
+        terminal = publisher._audio_queue.get()
+        assert terminal.status == "error"
+        assert isinstance(terminal.error, RuntimeError)

api/tests/unit_tests/core/base/tts/test_audio_mime.py

@@ -0,0 +1,66 @@
+from types import SimpleNamespace
+from typing import cast
+
+import pytest
+
+from core.base.tts.audio_mime import (
+    get_model_audio_mime_type,
+    inspect_audio_stream,
+    resolve_audio_mime_type,
+    sniff_audio_mime_type,
+)
+from core.model_manager import ModelInstance
+from core.plugin.entities.plugin_daemon import TTSAudioChunk
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
+
+
+def test_inspect_audio_stream_preserves_the_prefix_with_matching_chunk_mime_type() -> None:
+    chunks = [b"RIFF\x24\x00\x00\x00", b"WAVEfmt ", b"audio-data"]
+
+    stream, mime_type = inspect_audio_stream(
+        iter([TTSAudioChunk(chunk, "audio/wav") for chunk in chunks]), "audio/mpeg"
+    )
+
+    assert mime_type == "audio/wav"
+    assert list(stream) == chunks
+
+
+def test_inspect_audio_stream_rejects_mime_magic_mismatch() -> None:
+    with pytest.raises(InvokeBadRequestError, match="declared audio/mpeg, detected audio/wav"):
+        inspect_audio_stream([TTSAudioChunk(b"RIFF\x24\x00\x00\x00WAVEfmt ", "audio/mpeg")])
+
+
+def test_inspect_audio_stream_rejects_unsupported_reported_mime_type() -> None:
+    with pytest.raises(InvokeBadRequestError, match="unsupported chunk MIME type"):
+        inspect_audio_stream([TTSAudioChunk(b"audio", "application/octet-stream")])
+
+
+def test_resolve_audio_mime_type_falls_back_to_the_declared_model_type() -> None:
+    assert resolve_audio_mime_type(b"unrecognised", "audio/ogg") == "audio/ogg"
+
+
+def test_model_audio_mime_type_normalizes_plugin_metadata() -> None:
+    model_instance = SimpleNamespace(
+        get_model_schema=lambda: SimpleNamespace(model_properties={ModelPropertyKey.AUDIO_TYPE: "audio/x-wav"})
+    )
+
+    assert get_model_audio_mime_type(cast(ModelInstance, model_instance)) == "audio/wav"
+
+
+@pytest.mark.parametrize(
+    ("signature", "mime_type"),
+    [
+        (b"\xff\xfb" + b"\x00" * 30, "audio/mpeg"),
+        (b"\xff\xf1" + b"\x00" * 30, "audio/aac"),
+    ],
+)
+def test_sniff_audio_mime_type_distinguishes_mp3_frames_from_adts(signature: bytes, mime_type: str) -> None:
+    assert sniff_audio_mime_type(signature) == mime_type
+
+
+def test_id3_metadata_is_not_treated_as_proof_of_an_mp3_codec() -> None:
+    signature = b"ID3" + b"\x00" * 29
+
+    assert sniff_audio_mime_type(signature) is None
+    assert resolve_audio_mime_type(signature, reported_mime_type="audio/aac") == "audio/aac"

api/tests/unit_tests/core/plugin/impl/test_model_client.py

@@ -511,7 +511,12 @@ class TestPluginModelClient:
         mocker.patch.object(
             client,
             "_request_with_plugin_daemon_response_stream",
-            return_value=iter([SimpleNamespace(result="68656c6c6f"), SimpleNamespace(result="21")]),
+            return_value=iter(
+                [
+                    SimpleNamespace(result="68656c6c6f", mime_type="audio/wav"),
+                    SimpleNamespace(result="21", mime_type="audio/wav"),
+                ]
+            ),
         )
 
         result = list(
@@ -528,6 +533,7 @@ class TestPluginModelClient:
         )
 
         assert result == [b"hello", b"!"]
+        assert [chunk.mime_type for chunk in result] == ["audio/wav", "audio/wav"]
 
     def test_invoke_tts_wraps_plugin_daemon_inner_error(self, mocker: MockerFixture):
         client = PluginModelClient()

api/tests/unit_tests/core/plugin/test_backwards_invocation_model.py

@@ -1,8 +1,14 @@
-from unittest.mock import patch
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from core.plugin.backwards_invocation.model import PluginModelBackwardsInvocation
-from core.plugin.entities.request import RequestInvokeSummary
+from core.plugin.entities.plugin_daemon import TTSAudioChunk
+from core.plugin.entities.request import RequestInvokeSummary, RequestInvokeTTS
 from graphon.model_runtime.entities.message_entities import UserPromptMessage
+from graphon.model_runtime.entities.model_entities import ModelPropertyKey
 from models.account import Tenant
 
 
@@ -60,3 +66,73 @@ def test_invoke_summary_uses_same_user_scope_for_token_helpers():
         prompt_messages=[UserPromptMessage(content="short")],
         user_id="user-1",
     )
+
+
+def test_invoke_tts_emits_the_verified_mime_type_for_backwards_invocation():
+    tenant = Tenant(name="Test Workspace")
+    tenant.id = "tenant-1"
+    model_instance = MagicMock()
+    model_instance.invoke_tts.return_value = [b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data"]
+    model_instance.get_model_schema.return_value = SimpleNamespace(
+        model_properties={ModelPropertyKey.AUDIO_TYPE: "wav"}
+    )
+    payload = RequestInvokeTTS(
+        provider="provider-a",
+        model="qwen3-tts-flash",
+        content_text="hello",
+        voice="Cherry",
+    )
+
+    with patch.object(PluginModelBackwardsInvocation, "_get_bound_model_instance", return_value=model_instance):
+        result = list(PluginModelBackwardsInvocation.invoke_tts("user-1", tenant, payload))
+
+    assert result == [
+        {
+            "result": "524946462400000057415645666d742010000000617564696f2d64617461",
+            "mime_type": "audio/wav",
+        }
+    ]
+
+
+def test_invoke_tts_defers_provider_errors_until_the_response_generator_is_consumed():
+    tenant = Tenant(name="Test Workspace")
+    tenant.id = "tenant-1"
+    model_instance = MagicMock()
+    model_instance.invoke_tts.side_effect = RuntimeError("provider failed")
+    payload = RequestInvokeTTS(
+        provider="provider-a",
+        model="qwen3-tts-flash",
+        content_text="hello",
+        voice="Cherry",
+    )
+
+    with patch.object(PluginModelBackwardsInvocation, "_get_bound_model_instance", return_value=model_instance):
+        response = PluginModelBackwardsInvocation.invoke_tts("user-1", tenant, payload)
+
+    model_instance.invoke_tts.assert_not_called()
+    with pytest.raises(RuntimeError, match="provider failed"):
+        next(response)
+
+
+def test_backwards_event_stream_serializes_a_deferred_mime_error():
+    tenant = Tenant(name="Test Workspace")
+    tenant.id = "tenant-1"
+    model_instance = MagicMock()
+    model_instance.invoke_tts.return_value = [
+        TTSAudioChunk(b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data", "audio/mpeg")
+    ]
+    model_instance.get_model_schema.return_value = SimpleNamespace(
+        model_properties={ModelPropertyKey.AUDIO_TYPE: "mp3"}
+    )
+    payload = RequestInvokeTTS(
+        provider="provider-a",
+        model="qwen3-tts-flash",
+        content_text="hello",
+        voice="Cherry",
+    )
+
+    with patch.object(PluginModelBackwardsInvocation, "_get_bound_model_instance", return_value=model_instance):
+        response = PluginModelBackwardsInvocation.invoke_tts("user-1", tenant, payload)
+        chunks = list(PluginModelBackwardsInvocation.convert_to_event_stream(response))
+
+    assert json.loads(chunks[0])["error"].startswith("TTS provider output MIME does not match")

api/tests/unit_tests/core/tools/test_builtin_tools_extra.py

@@ -287,6 +287,7 @@ def test_tts_invoke_returns_messages(monkeypatch: pytest.MonkeyPatch, sqlite_ses
     )
     messages = list(tts.invoke(session=sqlite_session, user_id="u", tool_parameters={"model": "p#m", "text": "hello"}))
     assert [m.type for m in messages] == [ToolInvokeMessage.MessageType.TEXT, ToolInvokeMessage.MessageType.BLOB]
+    assert messages[1].meta == {"mime_type": "audio/mpeg"}
     assert captured_manager_kwargs == {"tenant_id": "tenant-1", "user_id": "u"}
 
 

api/tests/unit_tests/services/test_audio_service.py

@@ -60,9 +60,12 @@ from unittest.mock import MagicMock, Mock, patch
 from uuid import uuid4
 
 import pytest
+from flask import Flask
 from sqlalchemy.orm import Session
 from werkzeug.datastructures import FileStorage
 
+from core.plugin.entities.plugin_daemon import TTSAudioChunk
+from graphon.model_runtime.errors.invoke import InvokeBadRequestError
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import ConversationFromSource, MessageStatus
 from models.model import App, AppMode, AppModelConfig, Message
@@ -581,7 +584,8 @@ class TestAudioServiceTTS:
         )
 
         # Assert
-        assert result == b"audio data"
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"audio data"
         mock_model_manager_class.assert_called_once_with(tenant_id=app.tenant_id, user_id="user-123")
         mock_model_instance.invoke_tts.assert_called_once_with(
             content_text="Hello world",
@@ -619,7 +623,8 @@ class TestAudioServiceTTS:
         )
 
         # Assert
-        assert result == b"audio data"
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"audio data"
         # Verify default voice was used
         call_args = mock_model_instance.invoke_tts.call_args
         assert call_args.kwargs["voice"] == "default-voice"
@@ -656,7 +661,8 @@ class TestAudioServiceTTS:
         )
 
         # Assert
-        assert result == b"audio data"
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"audio data"
         call_args = mock_model_instance.invoke_tts.call_args
         assert call_args.kwargs["voice"] == "auto-voice"
 
@@ -696,7 +702,8 @@ class TestAudioServiceTTS:
         )
 
         # Assert
-        assert result == b"draft audio"
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"draft audio"
         mock_workflow_service.get_draft_workflow.assert_called_once_with(app_model=app, session=sqlite_session)
 
     @patch("services.audio_service.ModelManager.for_tenant", autospec=True)
@@ -762,12 +769,69 @@ class TestAudioServiceTTS:
         )
 
         # Assert
-        assert result == b"message audio"
+        assert result.content_type == "audio/mpeg"
+        assert result.get_data() == b"message audio"
         mock_model_instance.invoke_tts.assert_called_once_with(
             content_text="Message answer",
             voice="message-voice",
         )
 
+    @patch("services.audio_service.ModelManager.for_tenant", autospec=True)
+    def test_transcript_tts_uses_detected_wav_mime_type_for_streams(
+        self,
+        mock_model_manager_class,
+        factory: AudioServiceTestDataFactory,
+        sqlite_session: Session,
+        app: Flask,
+    ):
+        app_model_config = factory.create_app_model_config_mock(
+            text_to_speech_dict={"enabled": True, "voice": "en-US-Neural"}
+        )
+        app_model = factory.create_app_mock(mode=AppMode.CHAT, app_model_config=app_model_config)
+        mock_model_instance = MagicMock()
+        mock_model_instance.invoke_tts.return_value = iter(
+            [b"RIFF\x24\x00\x00\x00WAVEfmt ", b"\x10\x00\x00\x00audio-data"]
+        )
+        mock_model_manager_class.return_value.get_default_model_instance.return_value = mock_model_instance
+
+        with app.test_request_context("/text-to-audio", method="POST"):
+            result = AudioService.transcript_tts(
+                app_model=app_model,
+                session=sqlite_session,
+                text="Hello world",
+                voice="en-US-Neural",
+            )
+
+        assert result.content_type == "audio/wav"
+        assert result.get_data() == b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data"
+
+    @patch("services.audio_service.ModelManager.for_tenant", autospec=True)
+    def test_transcript_tts_returns_provider_output_error_for_mime_magic_mismatch(
+        self,
+        mock_model_manager_class,
+        factory: AudioServiceTestDataFactory,
+        sqlite_session: Session,
+        app: Flask,
+    ):
+        app_model_config = factory.create_app_model_config_mock(
+            text_to_speech_dict={"enabled": True, "voice": "en-US-Neural"}
+        )
+        app_model = factory.create_app_mock(mode=AppMode.CHAT, app_model_config=app_model_config)
+        mock_model_instance = MagicMock()
+        mock_model_instance.invoke_tts.return_value = iter(
+            [TTSAudioChunk(b"RIFF\x24\x00\x00\x00WAVEfmt \x10\x00\x00\x00audio-data", "audio/mpeg")]
+        )
+        mock_model_manager_class.return_value.get_default_model_instance.return_value = mock_model_instance
+
+        with app.test_request_context("/text-to-audio", method="POST"):
+            with pytest.raises(InvokeBadRequestError, match="output MIME does not match"):
+                AudioService.transcript_tts(
+                    app_model=app_model,
+                    session=sqlite_session,
+                    text="Hello world",
+                    voice="en-US-Neural",
+                )
+
     def test_transcript_tts_raises_error_when_text_missing(
         self,
         factory: AudioServiceTestDataFactory,

packages/contracts/generated/api/service/zod.gen.ts

@@ -3280,7 +3280,7 @@ export const zGetSiteResponse = zSite
 export const zPostTextToAudioBody = zTextToAudioPayloadWithUser
 
 /**
- * Returns the generated audio. Generator responses are streamed by the service as `audio/mpeg`; otherwise the provider output is returned directly.
+ * Returns the generated audio. The `Content-Type` header reflects the provider audio container, verified from the response bytes when recognizable. The binary response can be AAC, FLAC, MP4, MP3, Ogg, WAV, or WebM.
  */
 export const zPostTextToAudioResponse = z.custom<Blob | File>(
   (value) => value instanceof Blob || value instanceof File,

web/app/components/base/audio-btn/__tests__/audio.spec.ts

@@ -37,6 +37,9 @@ type Reader = {
 
 type AudioResponse = {
   status: number
+  headers: {
+    get: (name: string) => string | null
+  }
   body: {
     getReader: () => Reader
   }
@@ -152,7 +155,7 @@ const testState = {
 }
 
 class MockMediaSourceCtor extends MockMediaSource {
-  static isTypeSupported = vi.fn(() => true)
+  static isTypeSupported = vi.fn((_mimeType: string) => true)
 
   constructor() {
     super()
@@ -194,7 +197,11 @@ const setMediaSourceSupport = (options: { mediaSource: boolean; managedMediaSour
   })
 }
 
-const makeAudioResponse = (status: number, reads: ReaderResult[]): AudioResponse => {
+const makeAudioResponse = (
+  status: number,
+  reads: ReaderResult[],
+  contentType = 'audio/mpeg',
+): AudioResponse => {
   const read = vi.fn<() => Promise<ReaderResult>>()
   reads.forEach((result) => {
     read.mockResolvedValueOnce(result)
@@ -202,6 +209,9 @@ const makeAudioResponse = (status: number, reads: ReaderResult[]): AudioResponse
 
   return {
     status,
+    headers: {
+      get: (name: string) => (name.toLowerCase() === 'content-type' ? contentType : null),
+    },
     body: {
       getReader: () => ({ read }),
     },
@@ -274,14 +284,15 @@ describe('AudioPlayer', () => {
   })
 
   describe('constructor behavior', () => {
-    it('should initialize media source, audio, and media element source when MediaSource exists', () => {
+    it('should initialize default MP3 streaming before the first play request', () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
-      const mediaSource = testState.mediaSources[0]
 
-      expect(player.mediaSource).toBe(mediaSource as unknown as MediaSource)
+      expect(player.mediaSource).toBe(testState.mediaSources[0])
+      expect(testState.mediaSources).toHaveLength(1)
+      expect(MockMediaSourceCtor.isTypeSupported).toHaveBeenCalledWith('audio/mpeg')
       expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(1)
       expect(audio!.src).toBe('blob:mock-url')
       expect(audio!.autoplay).toBe(true)
@@ -289,36 +300,9 @@ describe('AudioPlayer', () => {
       expect(audioContext!.connect).toHaveBeenCalledTimes(1)
     })
 
-    it('should use complete-audio fallback when no MediaSource implementation exists', () => {
-      setMediaSourceSupport({ mediaSource: false, managedMediaSource: false })
-
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const audio = testState.audios[0]
-
-      expect(player.mediaSource).toBeNull()
-      expect(audio!.src).toBe('')
-      expect(audio!.autoplay).toBe(false)
-      expect(globalThis.URL.createObjectURL).not.toHaveBeenCalled()
-    })
-
-    it('should use complete-audio fallback when MP3 MediaSource is unsupported', () => {
-      MockMediaSourceCtor.isTypeSupported.mockReturnValue(false)
-
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const audio = testState.audios[0]
-
-      expect(MockMediaSourceCtor.isTypeSupported).toHaveBeenCalledWith('audio/mpeg')
-      expect(player.mediaSource).toBeNull()
-      expect(testState.mediaSources).toHaveLength(0)
-      expect(audio!.src).toBe('')
-      expect(audio!.autoplay).toBe(false)
-      expect(globalThis.URL.createObjectURL).not.toHaveBeenCalled()
-    })
-
     it('should configure fallback audio controls when ManagedMediaSource is used', () => {
       setMediaSourceSupport({ mediaSource: false, managedMediaSource: true })
 
-      // Create with callback to ensure constructor path completes with fallback source.
       const player = new AudioPlayer('/text-to-audio', false, 'msg-1', 'hello', undefined, vi.fn())
       const audio = testState.audios[0]
 
@@ -363,8 +347,9 @@ describe('AudioPlayer', () => {
       expect(callback).toHaveBeenCalledWith('canplay')
     })
 
-    it('should initialize source buffer only once when sourceopen fires multiple times', () => {
+    it('should initialize source buffer only once when sourceopen fires multiple times', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', vi.fn())
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
 
       mediaSource!.emit('sourceopen')
@@ -391,20 +376,78 @@ describe('AudioPlayer', () => {
         expect(mockTextToAudioStream).toHaveBeenCalledTimes(1)
       })
 
-      expect(mockTextToAudioStream).toHaveBeenCalledWith(
-        '/text-to-audio',
-        AppSourceType.webApp,
-        { content_type: 'audio/mpeg' },
-        {
-          message_id: 'msg-1',
-          streaming: true,
-          voice: 'en-US',
-          text: 'hello',
-        },
-      )
+      expect(mockTextToAudioStream).toHaveBeenCalledWith('/text-to-audio', AppSourceType.webApp, {
+        message_id: 'msg-1',
+        streaming: true,
+        voice: 'en-US',
+        text: 'hello',
+      })
       expect(player.isLoadData).toBe(true)
     })
 
+    it('should replace pending MP3 playback with WAV without reporting its abort', async () => {
+      let rejectInitialPlay: ((reason: DOMException) => void) | undefined
+      let resolveWavPlay: (() => void) | undefined
+      mockTextToAudioStream.mockResolvedValue(
+        makeAudioResponse(
+          200,
+          [
+            { value: new Uint8Array([1, 2]), done: false },
+            { value: new Uint8Array([3, 4]), done: true },
+          ],
+          'audio/wav; codecs=1',
+        ),
+      )
+      vi.mocked(globalThis.URL.createObjectURL)
+        .mockReturnValueOnce('blob:media-source')
+        .mockReturnValueOnce('blob:wav')
+      const callback = vi.fn()
+      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      const audio = testState.audios[0]
+      audio!.play
+        .mockImplementationOnce(
+          () =>
+            new Promise<void>((_resolve, reject) => {
+              rejectInitialPlay = reject
+            }),
+        )
+        .mockImplementationOnce(
+          () =>
+            new Promise<void>((resolve) => {
+              resolveWavPlay = resolve
+            }),
+        )
+
+      player.playAudio()
+
+      expect(audio!.play).toHaveBeenCalledTimes(1)
+      expect(mockTextToAudioStream).toHaveBeenCalledTimes(1)
+
+      await waitFor(() => expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(2))
+      expect(player.mediaSource).toBeNull()
+      expect(testState.mediaSources).toHaveLength(1)
+      expect(globalThis.URL.revokeObjectURL).toHaveBeenCalledWith('blob:media-source')
+      const audioBlob = vi.mocked(globalThis.URL.createObjectURL).mock.calls[1]![0] as Blob
+      expect(audioBlob).toMatchObject({ type: 'audio/wav', size: 4 })
+      expect(new Uint8Array(await audioBlob.arrayBuffer())).toEqual(new Uint8Array([1, 2, 3, 4]))
+      expect(audio!.play).toHaveBeenCalledTimes(2)
+
+      rejectInitialPlay?.(new DOMException('Playback aborted', 'AbortError'))
+      await Promise.resolve()
+      await Promise.resolve()
+
+      expect(callback).not.toHaveBeenCalledWith('error')
+      player.playAudio()
+      expect(audio!.play).toHaveBeenCalledTimes(2)
+
+      resolveWavPlay?.()
+      await waitFor(() => expect(callback).toHaveBeenCalledWith('play'))
+      audio!.emit('ended')
+
+      expect(callback.mock.calls.map(([event]) => event)).toEqual(['play', 'ended'])
+      expect(mockTextToAudioStream).toHaveBeenCalledTimes(1)
+    })
+
     it('should emit error callback and reset load flag when stream response status is not 200', async () => {
       MockMediaSourceCtor.isTypeSupported.mockReturnValue(false)
       const callback = vi.fn()
@@ -484,6 +527,7 @@ describe('AudioPlayer', () => {
           undefined,
           callback,
         )
+        await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
         const audio = testState.audios[0]
         const audioContext = testState.audioContexts[0]
 
@@ -502,6 +546,7 @@ describe('AudioPlayer', () => {
     it('should request media playback before a suspended audio context finishes resuming', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', false, 'msg-1', 'hello', undefined, callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
       let resolveResume: (() => void) | undefined
@@ -540,6 +585,7 @@ describe('AudioPlayer', () => {
           undefined,
           callback,
         )
+        await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
         const audio = testState.audios[0]
         const audioContext = testState.audioContexts[0]
 
@@ -559,6 +605,7 @@ describe('AudioPlayer', () => {
     it('should report an error when the audio context remains interrupted and allow retry', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', false, 'msg-1', 'hello', undefined, callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
 
@@ -586,6 +633,7 @@ describe('AudioPlayer', () => {
     it('should play ended audio when data is already loaded', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', false, 'msg-1', 'hello', undefined, callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
 
@@ -600,9 +648,10 @@ describe('AudioPlayer', () => {
       })
     })
 
-    it('should report loaded audio that is already playing without replaying it', () => {
+    it('should report loaded audio that is already playing without replaying it', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', false, 'msg-1', 'hello', undefined, callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
 
@@ -629,6 +678,75 @@ describe('AudioPlayer', () => {
       expect(player.isLoadData).toBe(false)
     })
 
+    it('should discard partial HTTP audio before a successful retry', async () => {
+      MockMediaSourceCtor.isTypeSupported.mockReturnValueOnce(true).mockReturnValue(false)
+      const callback = vi.fn()
+      const failedReader = {
+        read: vi
+          .fn()
+          .mockResolvedValueOnce({ value: new Uint8Array([1, 2]), done: false })
+          .mockRejectedValueOnce(new Error('stream lost')),
+      }
+      mockTextToAudioStream
+        .mockResolvedValueOnce({
+          status: 200,
+          headers: { get: () => 'audio/mpeg' },
+          body: { getReader: () => failedReader },
+        } as AudioResponse)
+        .mockResolvedValueOnce(
+          makeAudioResponse(200, [{ value: new Uint8Array([3, 4]), done: true }]),
+        )
+      const player = new AudioPlayer('/text-to-audio', false, 'msg-2', 'world', undefined, callback)
+      const audio = testState.audios[0]
+
+      player.playAudio()
+
+      await waitFor(() => expect(callback).toHaveBeenCalledWith('error'))
+      expect(player.cacheBuffers).toHaveLength(0)
+      expect(player.mediaSource).toBeNull()
+      expect(audio!.src).toBe('')
+      expect(globalThis.URL.revokeObjectURL).toHaveBeenCalledWith('blob:mock-url')
+
+      audio!.paused = true
+      player.playAudio()
+
+      await waitFor(() => expect(audio!.play).toHaveBeenCalledTimes(2))
+      expect(mockTextToAudioStream).toHaveBeenCalledTimes(2)
+      expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(2)
+      const audioBlob = vi.mocked(globalThis.URL.createObjectURL).mock.calls[1]![0] as Blob
+      expect(audioBlob).toMatchObject({ type: 'audio/mpeg', size: 2 })
+      expect(new Uint8Array(await audioBlob.arrayBuffer())).toEqual(new Uint8Array([3, 4]))
+    })
+
+    it('should ignore a pending HTTP response after destroy', async () => {
+      let resolveResponse: ((response: AudioResponse) => void) | undefined
+      const getReader = vi.fn()
+      const callback = vi.fn()
+      mockTextToAudioStream.mockImplementationOnce(
+        () =>
+          new Promise<AudioResponse>((resolve) => {
+            resolveResponse = resolve
+          }),
+      )
+      const player = new AudioPlayer('/text-to-audio', false, 'msg-2', 'world', undefined, callback)
+
+      player.playAudio()
+      player.destroy()
+      callback.mockClear()
+      resolveResponse?.({
+        status: 200,
+        headers: { get: () => 'audio/wav' },
+        body: { getReader },
+      } as unknown as AudioResponse)
+      await Promise.resolve()
+      await Promise.resolve()
+
+      expect(getReader).not.toHaveBeenCalled()
+      expect(testState.mediaSources).toHaveLength(1)
+      expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(1)
+      expect(callback).not.toHaveBeenCalled()
+    })
+
     it('should call pause flow and notify paused event when pauseAudio is invoked', () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
@@ -655,15 +773,18 @@ describe('AudioPlayer', () => {
     it('should end stream without playback when playAudioWithAudio receives empty content', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
 
       await player.playAudioWithAudio('', true)
 
       expect(player.isLoadData).toBe(false)
-      expect(player.cacheBuffers).toHaveLength(0)
+      expect(player.cacheBuffers).toHaveLength(1)
       expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
 
       mediaSource!.emit('sourceopen')
+      expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
+      mediaSource!.sourceBuffer.emit('updateend')
 
       expect(mediaSource!.endOfStream).toHaveBeenCalledTimes(1)
       expect(callback).not.toHaveBeenCalledWith('play')
@@ -674,13 +795,13 @@ describe('AudioPlayer', () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
-      const mediaSource = testState.mediaSources[0]
       const audioBase64 = Buffer.from('hello').toString('base64')
 
-      mediaSource!.emit('sourceopen')
       audio!.paused = true
       audioContext!.state = 'suspended'
-      await player.playAudioWithAudio(audioBase64, true)
+      await player.playAudioWithAudio(audioBase64, true, 'audio/mpeg')
+      const mediaSource = testState.mediaSources[0]
+      mediaSource!.emit('sourceopen')
 
       expect(player.isLoadData).toBe(true)
       expect(player.cacheBuffers).toHaveLength(0)
@@ -701,7 +822,7 @@ describe('AudioPlayer', () => {
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
 
-      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), false)
+      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), false, 'audio/mpeg')
 
       expect(player.isLoadData).toBe(false)
       expect(audioContext!.resume).not.toHaveBeenCalled()
@@ -733,6 +854,44 @@ describe('AudioPlayer', () => {
       expect(callback).toHaveBeenCalledWith('play')
     })
 
+    it('should combine automatic WAV TTS chunks into one playable blob', async () => {
+      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', vi.fn())
+      const audio = testState.audios[0]
+
+      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true, 'audio/wav')
+      await player.playAudioWithAudio(Buffer.from([3, 4]).toString('base64'), true, 'audio/wav')
+
+      expect(audio!.play).not.toHaveBeenCalled()
+      await player.playAudioWithAudio('', false)
+
+      await waitFor(() => expect(audio!.play).toHaveBeenCalledTimes(1))
+      expect(player.mediaSource).toBeNull()
+      expect(testState.mediaSources).toHaveLength(1)
+      expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(2)
+      const audioBlob = vi.mocked(globalThis.URL.createObjectURL).mock.calls[1]![0] as Blob
+      expect(audioBlob).toMatchObject({ type: 'audio/wav', size: 4 })
+      expect(new Uint8Array(await audioBlob.arrayBuffer())).toEqual(new Uint8Array([1, 2, 3, 4]))
+    })
+
+    it('should resume the audio context when WAV playback is requested while buffering', async () => {
+      const callback = vi.fn()
+      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      const audio = testState.audios[0]
+      const audioContext = testState.audioContexts[0]
+
+      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true, 'audio/wav')
+      audioContext!.state = 'suspended'
+      player.playAudio()
+
+      expect(audioContext!.resume).toHaveBeenCalledTimes(1)
+      expect(audio!.play).not.toHaveBeenCalled()
+
+      audio!.play.mockRejectedValueOnce(new DOMException('Not allowed', 'NotAllowedError'))
+      await player.playAudioWithAudio('', false)
+
+      await waitFor(() => expect(callback).toHaveBeenCalledWith('error'))
+    })
+
     it('should not start fallback playback after it is paused while buffering', async () => {
       MockMediaSourceCtor.isTypeSupported.mockReturnValue(false)
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', vi.fn())
@@ -750,6 +909,7 @@ describe('AudioPlayer', () => {
     it('should fall back to a complete MP3 when addSourceBuffer throws', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
       const audio = testState.audios[0]
       mediaSource!.addSourceBuffer.mockImplementationOnce(() => {
@@ -757,7 +917,6 @@ describe('AudioPlayer', () => {
       })
 
       mediaSource!.emit('sourceopen')
-      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true)
       await player.playAudioWithAudio('', false)
 
       await waitFor(() => expect(audio!.play).toHaveBeenCalledTimes(1))
@@ -769,13 +928,12 @@ describe('AudioPlayer', () => {
 
     it('should complete buffered fallback when addSourceBuffer throws after stream end', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', vi.fn())
+      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
       const audio = testState.audios[0]
       mediaSource!.addSourceBuffer.mockImplementationOnce(() => {
         throw new DOMException('Unsupported type', 'NotSupportedError')
       })
-
-      await player.playAudioWithAudio(Buffer.from([1, 2]).toString('base64'), true)
       await waitFor(() => expect(audio!.play).toHaveBeenCalledTimes(1))
       await player.playAudioWithAudio('', false)
       audio!.paused = true
@@ -798,7 +956,7 @@ describe('AudioPlayer', () => {
 
       audio!.paused = false
       audio!.ended = true
-      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), true)
+      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), true, 'audio/mpeg')
 
       expect(audio!.play).toHaveBeenCalledTimes(1)
       await waitFor(() => expect(callback).toHaveBeenCalledWith('play'))
@@ -812,7 +970,7 @@ describe('AudioPlayer', () => {
       audio!.paused = false
       audio!.ended = false
       audio!.played = {}
-      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), true)
+      await player.playAudioWithAudio(Buffer.from('hello').toString('base64'), true, 'audio/mpeg')
 
       expect(audio!.play).not.toHaveBeenCalled()
       expect(callback).not.toHaveBeenCalledWith('play')
@@ -821,6 +979,7 @@ describe('AudioPlayer', () => {
     it('should report a play failure and retry without requesting audio again', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const mediaSource = testState.mediaSources[0]
       mockTextToAudioStream.mockResolvedValue(
@@ -849,6 +1008,7 @@ describe('AudioPlayer', () => {
     it('should report a resume failure and allow playback to be retried', async () => {
       const callback = vi.fn()
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', callback)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
       const mediaSource = testState.mediaSources[0]
@@ -878,50 +1038,16 @@ describe('AudioPlayer', () => {
     })
   })
 
-  describe('buffering internals', () => {
-    it('should finish stream when receiveAudioData gets an undefined chunk', () => {
+  describe('MSE buffering', () => {
+    it('should preserve a chunk received before sourceopen', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const finishStream = vi
-        .spyOn(player as unknown as { finishStream: () => void }, 'finishStream')
-        .mockImplementation(() => {})
-      ;(
-        player as unknown as { receiveAudioData: (data: Uint8Array | undefined) => void }
-      ).receiveAudioData(undefined)
 
-      expect(finishStream).toHaveBeenCalledTimes(1)
-    })
-
-    it('should finish stream when receiveAudioData gets empty bytes', () => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const finishStream = vi
-        .spyOn(player as unknown as { finishStream: () => void }, 'finishStream')
-        .mockImplementation(() => {})
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        new Uint8Array(0),
+      await player.playAudioWithAudio(
+        Buffer.from([1, 2, 3]).toString('base64'),
+        false,
+        'audio/mpeg',
       )
-
-      expect(finishStream).toHaveBeenCalledTimes(1)
-    })
-
-    it('should queue incoming buffer when source buffer is updating', () => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
       const mediaSource = testState.mediaSources[0]
-      mediaSource!.emit('sourceopen')
-      mediaSource!.sourceBuffer.updating = true
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        new Uint8Array([1, 2, 3]),
-      )
-
-      expect(player.cacheBuffers.length).toBe(1)
-    })
-
-    it('should preserve audio received before sourceopen and append it once ready', () => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const mediaSource = testState.mediaSources[0]
-
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        new Uint8Array([1, 2, 3]),
-      )
 
       expect(player.cacheBuffers).toHaveLength(1)
       expect(mediaSource!.sourceBuffer.appendBuffer).not.toHaveBeenCalled()
@@ -929,114 +1055,68 @@ describe('AudioPlayer', () => {
       mediaSource!.emit('sourceopen')
 
       expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledTimes(1)
+      expect(new Uint8Array(mediaSource!.sourceBuffer.appendBuffer.mock.calls[0]![0])).toEqual(
+        new Uint8Array([1, 2, 3]),
+      )
       expect(player.cacheBuffers).toHaveLength(0)
     })
 
-    it('should append queued buffers in order after updateend', () => {
+    it('should append queued chunks in order after updateend', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
+
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
       mediaSource!.emit('sourceopen')
       mediaSource!.sourceBuffer.updating = true
-
-      const first = new Uint8Array([1])
-      const second = new Uint8Array([2])
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        first,
-      )
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        second,
-      )
+      await player.playAudioWithAudio(Buffer.from([2]).toString('base64'), false, 'audio/mpeg')
+      await player.playAudioWithAudio(Buffer.from([3]).toString('base64'), false, 'audio/mpeg')
 
       mediaSource!.sourceBuffer.updating = false
       mediaSource!.sourceBuffer.emit('updateend')
-      expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledTimes(1)
-      expect(new Uint8Array(mediaSource!.sourceBuffer.appendBuffer.mock.calls[0]![0])).toEqual(
-        first,
-      )
-
       mediaSource!.sourceBuffer.emit('updateend')
-      expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledTimes(2)
-      expect(new Uint8Array(mediaSource!.sourceBuffer.appendBuffer.mock.calls[1]![0])).toEqual(
-        second,
-      )
+
+      expect(
+        mediaSource!.sourceBuffer.appendBuffer.mock.calls.map(
+          ([buffer]) => new Uint8Array(buffer)[0],
+        ),
+      ).toEqual([1, 2, 3])
     })
 
-    it('should append previously queued buffer before new one when source buffer is idle', () => {
+    it('should end MediaSource only after every queued chunk is appended', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
+
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
+      await player.playAudioWithAudio(Buffer.from([2]).toString('base64'), false, 'audio/mpeg')
+      await player.playAudioWithAudio('', false)
       const mediaSource = testState.mediaSources[0]
+
       mediaSource!.emit('sourceopen')
-
-      const existingBuffer = new ArrayBuffer(2)
-      player.cacheBuffers = [existingBuffer]
-      mediaSource!.sourceBuffer.updating = false
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        new Uint8Array([9]),
-      )
-
-      expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledTimes(1)
-      expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledWith(existingBuffer)
-      expect(player.cacheBuffers.length).toBe(1)
-    })
-
-    it('should end the stream only after the final queued buffer is appended', () => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const mediaSource = testState.mediaSources[0]
-      mediaSource!.emit('sourceopen')
-      mediaSource!.sourceBuffer.updating = true
-      player.cacheBuffers = [new ArrayBuffer(3)]
-
-      ;(player as unknown as { finishStream: () => void }).finishStream()
-
       expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
 
-      mediaSource!.sourceBuffer.updating = false
       mediaSource!.sourceBuffer.emit('updateend')
-      expect(mediaSource!.sourceBuffer.appendBuffer).toHaveBeenCalledTimes(1)
       expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
 
       mediaSource!.sourceBuffer.emit('updateend')
       expect(mediaSource!.endOfStream).toHaveBeenCalledTimes(1)
     })
 
-    it('should end an open stream at most once', () => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const mediaSource = testState.mediaSources[0]
-      mediaSource!.emit('sourceopen')
-
-      ;(player as unknown as { finishStream: () => void }).finishStream()
-      ;(player as unknown as { finishStream: () => void }).finishStream()
-
-      expect(mediaSource!.endOfStream).toHaveBeenCalledTimes(1)
-    })
-
-    it.each(['closed', 'ended'] as const)('should not end a %s media source', (readyState) => {
-      const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
-      const mediaSource = testState.mediaSources[0]
-      mediaSource!.emit('sourceopen')
-      mediaSource!.readyState = readyState
-
-      ;(player as unknown as { finishStream: () => void }).finishStream()
-
-      expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
-    })
-
-    it('should stop buffering and release browser resources after destroy', async () => {
+    it('should release resources and stop pending buffering after destroy', async () => {
       const player = new AudioPlayer('/text-to-audio', true, 'msg-1', 'hello', 'en-US', null)
+      await player.playAudioWithAudio(Buffer.from([1]).toString('base64'), false, 'audio/mpeg')
       const mediaSource = testState.mediaSources[0]
       const audio = testState.audios[0]
       const audioContext = testState.audioContexts[0]
-      mediaSource!.emit('sourceopen')
 
       player.destroy()
-      ;(player as unknown as { receiveAudioData: (data: Uint8Array) => void }).receiveAudioData(
-        new Uint8Array([1]),
-      )
-      ;(player as unknown as { finishStream: () => void }).finishStream()
+      await player.playAudioWithAudio(Buffer.from([2]).toString('base64'), false, 'audio/mpeg')
+      mediaSource!.emit('sourceopen')
       mediaSource!.sourceBuffer.emit('updateend')
       await Promise.resolve()
 
+      expect(testState.mediaSources).toHaveLength(1)
+      expect(globalThis.URL.createObjectURL).toHaveBeenCalledTimes(1)
+      expect(mediaSource!.addSourceBuffer).not.toHaveBeenCalled()
       expect(mediaSource!.sourceBuffer.appendBuffer).not.toHaveBeenCalled()
-      expect(mediaSource!.endOfStream).not.toHaveBeenCalled()
       expect(audio!.pause).toHaveBeenCalledTimes(1)
       expect(audioContext!.close).toHaveBeenCalledTimes(1)
       expect(globalThis.URL.revokeObjectURL).toHaveBeenCalledWith('blob:mock-url')

web/app/components/base/audio-btn/audio.ts

@@ -1,6 +1,6 @@
 import { AppSourceType, textToAudioStream } from '@/service/share'
 
-const AUDIO_CONTENT_TYPE = 'audio/mpeg'
+const DEFAULT_AUDIO_CONTENT_TYPE = 'audio/mpeg'
 
 declare global {
   // oxlint-disable-next-line typescript/consistent-type-definitions
@@ -25,9 +25,12 @@ export class AudioPlayer {
   private streamEnded = false
   private endOfStreamCalled = false
   private destroyed = false
-  private playbackPending = false
+  private pendingPlaybackSource: string | null = null
   private playWhenReady = false
+  private mediaSourceDisabled = false
   private sourceOpenListener?: () => void
+  private audioMimeType = DEFAULT_AUDIO_CONTENT_TYPE
+  private reader?: ReadableStreamDefaultReader<Uint8Array>
   constructor(
     streamUrl: string,
     isPublic: boolean,
@@ -43,25 +46,10 @@ export class AudioPlayer {
     this.isPublic = isPublic
     this.voice = voice
     this.callback = callback
-    // Compatible with iphone ios17 ManagedMediaSource
-    const MediaSourceConstructor = window.ManagedMediaSource || window.MediaSource
-    const isManagedMediaSource = Boolean(
-      window.ManagedMediaSource && MediaSourceConstructor === window.ManagedMediaSource,
-    )
-    const supportsStreaming = Boolean(MediaSourceConstructor?.isTypeSupported?.(AUDIO_CONTENT_TYPE))
-    this.mediaSource =
-      supportsStreaming && MediaSourceConstructor ? new MediaSourceConstructor() : null
     this.audio = new Audio()
+    this.mediaSource = null
+    this.initializeMediaSource()
     this.setCallback(callback)
-    if (this.mediaSource && isManagedMediaSource) {
-      // if use  ManagedMediaSource
-      this.audio.disableRemotePlayback = true
-      this.audio.controls = true
-    }
-    this.listenMediaSource(AUDIO_CONTENT_TYPE)
-    this.objectUrl = this.mediaSource ? URL.createObjectURL(this.mediaSource) : ''
-    this.audio.src = this.objectUrl
-    this.audio.autoplay = Boolean(this.mediaSource)
     const source = this.audioContext.createMediaElementSource(this.audio)
     source.connect(this.audioContext.destination)
   }
@@ -78,15 +66,76 @@ export class AudioPlayer {
         this.sourceBuffer?.addEventListener('updateend', this.flushBuffers)
         this.flushBuffers()
       } catch {
-        this.mediaSource = null
-        this.audio.autoplay = false
-        this.releaseObjectUrl()
+        this.mediaSourceDisabled = true
+        this.releaseMediaSource()
         if (this.streamEnded) this.finishBlobAudio()
       }
     }
     this.mediaSource?.addEventListener('sourceopen', this.sourceOpenListener)
   }
 
+  private initializeMediaSource() {
+    if (this.destroyed || this.mediaSourceDisabled) return
+
+    const MediaSourceConstructor = window.ManagedMediaSource || window.MediaSource
+    const isManagedMediaSource = Boolean(
+      window.ManagedMediaSource && MediaSourceConstructor === window.ManagedMediaSource,
+    )
+    const supportsStreaming =
+      this.audioMimeType === DEFAULT_AUDIO_CONTENT_TYPE &&
+      Boolean(MediaSourceConstructor?.isTypeSupported?.(this.audioMimeType))
+    if (!supportsStreaming || !MediaSourceConstructor) return
+
+    this.mediaSource = new MediaSourceConstructor()
+    if (isManagedMediaSource) {
+      this.audio.disableRemotePlayback = true
+      this.audio.controls = true
+    }
+    this.listenMediaSource(this.audioMimeType)
+    this.objectUrl = URL.createObjectURL(this.mediaSource)
+    this.audio.src = this.objectUrl
+    this.audio.autoplay = true
+  }
+
+  private setAudioMimeType(audioType: string | null | undefined) {
+    const mimeType = audioType?.split(';', 1)[0]?.trim() || DEFAULT_AUDIO_CONTENT_TYPE
+    if (mimeType !== this.audioMimeType) {
+      this.audioMimeType = mimeType
+      this.releaseMediaSource()
+    }
+
+    if (!this.streamEnded && !this.mediaSource && !this.objectUrl) this.initializeMediaSource()
+    if (this.mediaSource && this.playWhenReady) this.requestPlayback()
+  }
+
+  private releaseMediaSource() {
+    if (this.sourceOpenListener)
+      this.mediaSource?.removeEventListener('sourceopen', this.sourceOpenListener)
+
+    if (this.sourceBuffer) {
+      this.sourceBuffer.removeEventListener('updateend', this.flushBuffers)
+      if (this.mediaSource?.readyState === 'open') {
+        try {
+          this.sourceBuffer.abort()
+        } catch {}
+      }
+    }
+
+    this.sourceBuffer = undefined
+    this.mediaSource = null
+    this.endOfStreamCalled = false
+    this.audio.autoplay = false
+    this.releaseObjectUrl()
+  }
+
+  private failLoad() {
+    this.cacheBuffers = []
+    this.streamEnded = false
+    this.releaseMediaSource()
+    this.isLoadData = false
+    this.callback?.('error')
+  }
+
   private flushBuffers = () => {
     if (
       this.destroyed ||
@@ -109,38 +158,40 @@ export class AudioPlayer {
   }
 
   private requestPlayback(reportIfPlaying = false) {
-    if (this.destroyed || this.playbackPending) return
+    const playbackSource = this.objectUrl
+    if (this.destroyed || !playbackSource || this.pendingPlaybackSource === playbackSource) return
     if (!this.isAudioContextPaused() && !this.audio.paused && !this.audio.ended) {
       if (reportIfPlaying) this.callback?.('play')
       return
     }
 
-    this.playbackPending = true
-    void this.resumeAndPlay()
+    this.pendingPlaybackSource = playbackSource
+    void this.resumeAndPlay(playbackSource)
   }
 
   private isAudioContextPaused() {
     return this.audioContext.state === 'suspended' || this.audioContext.state === 'interrupted'
   }
 
-  private async resumeAndPlay() {
+  private async resumeAndPlay(playbackSource: string) {
     try {
       const pendingOperations: Promise<unknown>[] = []
       if (this.isAudioContextPaused()) pendingOperations.push(this.audioContext.resume())
       if (this.audio.paused || this.audio.ended) pendingOperations.push(this.audio.play())
 
       await Promise.all(pendingOperations)
-      if (this.destroyed) return
+      if (this.destroyed || !this.playWhenReady || playbackSource !== this.objectUrl) return
       if (this.isAudioContextPaused()) {
         this.callback?.('error')
         return
       }
 
-      if (!this.destroyed) this.callback?.('play')
+      this.callback?.('play')
     } catch {
-      if (!this.destroyed) this.callback?.('error')
+      if (!this.destroyed && this.playWhenReady && playbackSource === this.objectUrl)
+        this.callback?.('error')
     } finally {
-      this.playbackPending = false
+      if (this.pendingPlaybackSource === playbackSource) this.pendingPlaybackSource = null
     }
   }
 
@@ -204,7 +255,6 @@ export class AudioPlayer {
       const audioResponse = (await textToAudioStream(
         this.url,
         this.isPublic ? AppSourceType.webApp : AppSourceType.installedApp,
-        { content_type: 'audio/mpeg' },
         {
           message_id: this.msgId,
           streaming: true,
@@ -212,15 +262,18 @@ export class AudioPlayer {
           text: this.msgContent,
         },
       )) as Response
+      if (this.destroyed) return
       if (audioResponse.status !== 200) {
-        this.isLoadData = false
-        this.callback?.('error')
+        this.failLoad()
         return
       }
+      this.setAudioMimeType(audioResponse.headers.get('content-type'))
       if (!audioResponse.body) throw new Error('Audio response body is missing')
       const reader = audioResponse.body.getReader()
+      this.reader = reader
       while (true) {
         const { value, done } = await reader.read()
+        if (this.destroyed) return
         if (value?.byteLength) this.receiveAudioData(value)
         if (done) {
           this.finishStream()
@@ -228,22 +281,23 @@ export class AudioPlayer {
         }
       }
     } catch {
-      this.isLoadData = false
-      this.callback?.('error')
+      if (!this.destroyed) this.failLoad()
+    } finally {
+      this.reader = undefined
     }
   }
 
   // play audio
   public playAudio() {
+    this.playWhenReady = true
     if (this.isLoadData) {
       if (!this.mediaSource && !this.objectUrl) {
-        this.playWhenReady = true
+        if (this.isAudioContextPaused()) void this.audioContext.resume().catch(() => {})
         return
       }
       this.requestPlayback(true)
     } else {
       this.isLoadData = true
-      this.playWhenReady = true
       if (this.mediaSource) this.requestPlayback(true)
       else if (this.isAudioContextPaused()) void this.audioContext.resume().catch(() => {})
       this.loadAudio()
@@ -261,11 +315,12 @@ export class AudioPlayer {
     this.finishBlobAudio()
   }
 
-  public async playAudioWithAudio(audio: string, play = true) {
+  public async playAudioWithAudio(audio: string, play = true, audioType?: string) {
     if (!audio || !audio.length) {
       this.finishStream()
       return
     }
+    this.setAudioMimeType(audioType)
     const audioContent = Uint8Array.from(atob(audio), (char) => char.charCodeAt(0))
     this.receiveAudioData(audioContent)
     if (play) {
@@ -287,23 +342,13 @@ export class AudioPlayer {
 
     this.destroyed = true
     this.cacheBuffers = []
+    void this.reader?.cancel().catch(() => {})
+    this.reader = undefined
     this.callback?.('paused')
     this.audio.pause()
 
-    if (this.sourceOpenListener)
-      this.mediaSource?.removeEventListener('sourceopen', this.sourceOpenListener)
-
-    if (this.sourceBuffer) {
-      this.sourceBuffer.removeEventListener('updateend', this.flushBuffers)
-      if (this.mediaSource?.readyState === 'open') {
-        try {
-          this.sourceBuffer.abort()
-        } catch {}
-      }
-    }
-
+    this.releaseMediaSource()
     void this.audioContext.close().catch(() => {})
-    this.releaseObjectUrl()
   }
 
   private receiveAudioData(unit8Array: Uint8Array | undefined) {
@@ -327,7 +372,7 @@ export class AudioPlayer {
       return
     }
 
-    const audioBlob = new Blob(this.cacheBuffers, { type: AUDIO_CONTENT_TYPE })
+    const audioBlob = new Blob(this.cacheBuffers, { type: this.audioMimeType })
     this.cacheBuffers = []
     this.releaseObjectUrl()
     this.objectUrl = URL.createObjectURL(audioBlob)

web/app/components/base/chat/chat/__tests__/hooks.spec.tsx

@@ -70,7 +70,7 @@ type HookCallbacks = {
   onHumanInputFormFilled: (filled: Record<string, unknown>) => void
   onHumanInputFormTimeout: (timeout: Record<string, unknown>) => void
   onWorkflowPaused: (workflowPaused: Record<string, unknown>) => void
-  onTTSChunk: (messageId: string, audio: string) => void
+  onTTSChunk: (messageId: string, audio: string, audioType?: string) => void
   onTTSEnd: (messageId: string, audio: string) => void
   onReasoning: (chunk: {
     data: { message_id?: string; reasoning: string; node_id?: string; is_final?: boolean }

web/app/components/base/chat/chat/hooks.ts

@@ -899,11 +899,11 @@ export const useChat = (
               responseItem.workflowProcess.tracing[currentIndex] = nodeFinishedData as any
           })
         },
-        onTTSChunk: (messageId: string, audio: string) => {
+        onTTSChunk: (messageId: string, audio: string, audioType?: string) => {
           if (!audio || audio === '') return
           const audioPlayer = getOrCreatePlayer()
           if (audioPlayer) {
-            audioPlayer.playAudioWithAudio(audio, true)
+            audioPlayer.playAudioWithAudio(audio, true, audioType)
             AudioPlayerManager.getInstance().resetMsgId(messageId)
           }
         },
@@ -1636,11 +1636,11 @@ export const useChat = (
             parentId: data.parent_message_id,
           })
         },
-        onTTSChunk: (messageId: string, audio: string) => {
+        onTTSChunk: (messageId: string, audio: string, audioType?: string) => {
           if (!audio || audio === '') return
           const audioPlayer = getOrCreatePlayer()
           if (audioPlayer) {
-            audioPlayer.playAudioWithAudio(audio, true)
+            audioPlayer.playAudioWithAudio(audio, true, audioType)
             AudioPlayerManager.getInstance().resetMsgId(messageId)
           }
         },

web/app/components/workflow-app/hooks/__tests__/use-workflow-run-callbacks.spec.ts

@@ -136,9 +136,9 @@ describe('useWorkflowRun callbacks helpers', () => {
       workflowData,
     )
 
-    callbacks.onTTSChunk?.('message-1', 'audio-chunk')
+    callbacks.onTTSChunk?.('message-1', 'audio-chunk', 'audio/wav')
     expect(getOrCreatePlayer).toHaveBeenCalled()
-    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true)
+    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true, 'audio/wav')
     expect(mockResetMsgId).toHaveBeenCalledWith('message-1')
 
     callbacks.onWorkflowPaused?.({ workflow_run_id: 'run-2' } as never)
@@ -210,7 +210,7 @@ describe('useWorkflowRun callbacks helpers', () => {
     expect(handlers.handleWorkflowNodeStarted).not.toHaveBeenCalled()
 
     finalCallbacks.onTTSChunk?.('message-2', 'audio-chunk')
-    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true)
+    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true, undefined)
     expect(mockResetMsgId).toHaveBeenCalledWith('message-2')
 
     finalCallbacks.onTTSChunk?.('message-3', '')
@@ -322,7 +322,7 @@ describe('useWorkflowRun callbacks helpers', () => {
     expect(fetchInspectVars).toHaveBeenCalledWith({})
     expect(invalidAllLastRun).toHaveBeenCalled()
     expect(userCallbacks.onCompleted).toHaveBeenCalledWith(false, '')
-    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true)
+    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true, undefined)
     expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-finished', false)
     expect(mockResetMsgId).toHaveBeenCalledWith('message-1')
     expect(handlers.handleWorkflowPaused).toHaveBeenCalled()
@@ -501,7 +501,7 @@ describe('useWorkflowRun callbacks helpers', () => {
       finalCallbacks,
     )
     expect(mockSseGet).toHaveBeenCalledTimes(1)
-    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true)
+    expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-chunk', true, undefined)
     expect(player.playAudioWithAudio).toHaveBeenCalledWith('audio-finished', false)
     expect(clearAbortController).toHaveBeenCalled()
     expect(handlers.handleWorkflowFailed).toHaveBeenCalled()

web/app/components/workflow-app/hooks/use-workflow-run-callbacks.ts

@@ -245,11 +245,11 @@ export const createBaseWorkflowRunCallbacks = ({
     onReasoning: (params) => {
       handleWorkflowReasoning(params)
     },
-    onTTSChunk: (messageId: string, audio: string) => {
+    onTTSChunk: (messageId: string, audio: string, audioType?: string) => {
       if (!audio || audio === '') return
       const audioPlayer = getOrCreatePlayer()
       if (audioPlayer) {
-        audioPlayer.playAudioWithAudio(audio, true)
+        audioPlayer.playAudioWithAudio(audio, true, audioType)
         AudioPlayerManager.getInstance().resetMsgId(messageId)
       }
     },
@@ -430,9 +430,9 @@ export const createFinalWorkflowRunCallbacks = ({
     onReasoning: (params) => {
       handleWorkflowReasoning(params)
     },
-    onTTSChunk: (messageId: string, audio: string) => {
+    onTTSChunk: (messageId: string, audio: string, audioType?: string) => {
       if (!audio || audio === '') return
-      player?.playAudioWithAudio(audio, true)
+      player?.playAudioWithAudio(audio, true, audioType)
       AudioPlayerManager.getInstance().resetMsgId(messageId)
     },
     onTTSEnd: (_messageId: string, audio: string) => {

web/service/base.spec.ts

@@ -328,6 +328,44 @@ describe('handleStream', () => {
       expect(onData).not.toHaveBeenCalled()
     })
 
+    it('should pass the TTS MIME type through the stream boundary', async () => {
+      const onData = vi.fn()
+      const onCompleted = vi.fn()
+      const onTTSChunk = vi.fn()
+      const ttsEvent = {
+        event: 'tts_message',
+        message_id: 'message-1',
+        audio: 'audio-chunk',
+        audio_type: 'audio/wav',
+      }
+      const mockReader = {
+        read: vi
+          .fn()
+          .mockResolvedValueOnce({
+            done: false,
+            value: new TextEncoder().encode(`data: ${JSON.stringify(ttsEvent)}\n`),
+          })
+          .mockResolvedValueOnce({ done: true, value: undefined }),
+      }
+      const mockResponse = {
+        ok: true,
+        body: { getReader: () => mockReader },
+      } as unknown as Response
+      const interveningNoops = Array.from({ length: 18 }, () => undefined)
+
+      ;(handleStream as (...args: unknown[]) => void)(
+        mockResponse,
+        onData,
+        onCompleted,
+        ...interveningNoops,
+        onTTSChunk,
+      )
+
+      await waitFor(() => {
+        expect(onTTSChunk).toHaveBeenCalledWith('message-1', 'audio-chunk', 'audio/wav')
+      })
+    })
+
     it('should complete with error when the stream reader rejects', async () => {
       const onData = vi.fn()
       const onCompleted = vi.fn()

web/service/base.ts

@@ -116,7 +116,7 @@ type IOnParallelBranchFinished = (parallelBranchFinished: ParallelBranchFinished
 type IOnTextChunk = (textChunk: TextChunkResponse) => void
 type IOnReasoning = (reasoningChunk: ReasoningChunkResponse) => void
 type IOnTTSChunk = (messageId: string, audioStr: string, audioType?: string) => void
-type IOnTTSEnd = (messageId: string, audioStr: string, audioType?: string) => void
+type IOnTTSEnd = (messageId: string, audioStr: string) => void
 type IOnTextReplace = (textReplace: TextReplaceResponse) => void
 type IOnLoopStarted = (workflowStarted: LoopStartedResponse) => void
 type IOnLoopNext = (workflowStarted: LoopNextResponse) => void

web/service/share.ts

@@ -401,7 +401,6 @@ export const audioToText = (
 export const textToAudioStream = (
   url: string,
   appSourceType: AppSourceType,
-  header: { content_type: string },
   body: {
     streaming: boolean
     voice?: string
@@ -409,7 +408,7 @@ export const textToAudioStream = (
     text?: string | null | undefined
   },
 ) => {
-  return getAction('post', appSourceType)(url, { body, header }, { needAllResponseContent: true })
+  return getAction('post', appSourceType)(url, { body }, { needAllResponseContent: true })
 }
 
 export const fetchAccessToken = async ({