[唤醒应答加速]：完成ACK音频预热缓存，包含唤醒后复用播放和回归测试

tests/test_live_runtime.py

@@ -113,6 +113,19 @@ class QueueLlmProvider:
         yield ReplyDelta("", finish_reason="stop")
 
 
+class CountingTtsProvider:
+    def __init__(self) -> None:
+        self.delegate = SineTtsProvider()
+        self.synthesized_texts: list[str] = []
+
+    def load(self) -> None:
+        self.delegate.load()
+
+    def synthesize(self, text: str) -> AudioSegment:
+        self.synthesized_texts.append(text)
+        return self.delegate.synthesize(text)
+
+
 class RecordingReporter:
     def __init__(self) -> None:
         self.statuses: list[str] = []
@@ -282,6 +295,37 @@ class LiveRuntimeTests(unittest.TestCase):
         positions = [event_types.index(item) for item in expected_order]
         self.assertEqual(positions, sorted(positions))
 
+    def test_wake_ack_audio_is_prepared_once_and_reused(self) -> None:
+        frames = []
+        for idx, _text in enumerate(["第一问", "第二问"]):
+            base_id = idx * 5
+            base_ms = idx * 120
+            frames.append(wake_frame(base_id, base_ms))
+            frames.extend(segment_frames(base_id + 1, base_ms + 20))
+        transport = MemoryAudioTransport(frames, flush_clears_input=False)
+        ack_tts = CountingTtsProvider()
+        runtime = VoiceAssistantPipeline(
+            config=AppConfig(llm_api_key="secret", speech_provider="cloud", wake_ack_text="我在"),
+            transport=transport,
+            wakeword=KeywordWakeWordProvider(),
+            vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40),
+            stt=QueueSttProvider(["第一问", "第二问"]),
+            realtime_stt=None,
+            llm=MockLlmProvider(["这是答复。"]),
+            tts=SineTtsProvider(),
+            ack_tts=ack_tts,
+            context=ConversationContext(),
+            reporter=RecordingReporter(),
+            event_bus=PipelineEventBus(),
+        )
+
+        summary = runtime.run(max_turns=2)
+
+        self.assertEqual(summary.completed_turns, 2)
+        self.assertEqual(ack_tts.synthesized_texts, ["我在"])
+        self.assertEqual(transport.played_segments[0].metadata["text"], "我在")
+        self.assertEqual(transport.played_segments[2].metadata["text"], "我在")
+
     def test_zero_post_playback_drain_flushes_without_dropping_prefilled_question(self) -> None:
         runtime, stt, _, transport, reporter = make_runtime(["第一问"])
         self.assertEqual(runtime.config.post_playback_drain_ms, 0)