[播报文本净化]：完成TTS表情包过滤，包含emoji清理、上下文净化和回归测试

tests/test_live_runtime.py

@@ -26,7 +26,7 @@ from owner_voice_pet.events import (
     PipelineEventBus,
 )
 from owner_voice_pet.llm import MockLlmProvider
-from owner_voice_pet.models import AudioFrame, AudioSegment, PlaybackResult, ReplyDelta, Transcript, TransportHealth
+from owner_voice_pet.models import AudioFrame, AudioSegment, ErrorCode, PlaybackResult, ReplyDelta, Transcript, TransportHealth
 from owner_voice_pet.assistant_pipeline import VoiceAssistantPipeline
 from owner_voice_pet.runtime import build_live_runtime
 from owner_voice_pet.stt import MetadataSttProvider, SherpaOnnxSttProvider
@@ -409,6 +409,64 @@ class LiveRuntimeTests(unittest.TestCase):
         self.assertEqual(llm.calls[0][-1].content, "第一问")
         self.assertNotIn("小杰小杰", llm.calls[0][-1].content)
 
+    def test_assistant_reply_sanitizes_tts_text_and_context(self) -> None:
+        frames = [wake_frame(0, 0)]
+        frames.extend(segment_frames(1, 20, partials=["第一问", "第一问"]))
+        transport = MemoryAudioTransport(frames, flush_clears_input=False)
+        stt = QueueSttProvider(["第一问"])
+        llm = QueueLlmProvider([["你好 😊。没问题[捂脸]，我来帮你。"]])
+        context = ConversationContext()
+        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=stt,
+            realtime_stt=MetadataSttProvider(),
+            llm=llm,
+            tts=SineTtsProvider(),
+            context=context,
+            reporter=RecordingReporter(),
+            event_bus=PipelineEventBus(),
+        )
+
+        summary = runtime.run(max_turns=1)
+        context_texts = [message.content for message in context.messages()]
+
+        self.assertEqual(summary.completed_turns, 1)
+        self.assertEqual(transport.played_segments[0].metadata["text"], "你好。没问题，我来帮你。")
+        self.assertIn("你好。没问题，我来帮你。", context_texts)
+        self.assertFalse(any("😊" in text or "[捂脸]" in text for text in context_texts))
+
+    def test_emoji_only_reply_recovers_without_tts_playback(self) -> None:
+        frames = [wake_frame(0, 0)]
+        frames.extend(segment_frames(1, 20, partials=["第一问", "第一问"]))
+        transport = MemoryAudioTransport(frames, flush_clears_input=False)
+        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=MetadataSttProvider(),
+            llm=QueueLlmProvider([["😂😂"]]),
+            tts=SineTtsProvider(),
+            context=ConversationContext(),
+            reporter=RecordingReporter(),
+            event_bus=PipelineEventBus(),
+        )
+
+        summary = runtime.run(once=True)
+        event_types = [event.type for event in runtime.event_bus.events]
+
+        self.assertEqual(summary.completed_turns, 0)
+        self.assertEqual(summary.failed_turns, 1)
+        self.assertIsNotNone(summary.last_error)
+        self.assertEqual(summary.last_error.code, ErrorCode.TTS_EMPTY_AUDIO)
+        self.assertEqual(transport.played_segments, [])
+        self.assertNotIn(TTS_STARTED, event_types)
+        self.assertEqual(event_types[-1], STANDBY_RESUMED)
+
     def test_live_runtime_uses_primary_speaker_endpoint_by_default(self) -> None:
         runtime = build_live_runtime(AppConfig(llm_api_key="secret"))
         self.assertIsInstance(runtime.vad_recorder, PrimarySpeakerVadRecorder)

tests/test_pipeline_llm_tts.py

@@ -12,7 +12,7 @@ from owner_voice_pet.models import AudioFrame, ErrorCode, Message, PipelineState
 from owner_voice_pet.pipeline import VoicePipeline
 from owner_voice_pet.stt import MetadataSttProvider
 from owner_voice_pet.transport import MemoryAudioTransport
-from owner_voice_pet.tts import CloudTtsProvider, SentenceBuffer, SineTtsProvider
+from owner_voice_pet.tts import CloudTtsProvider, SentenceBuffer, SineTtsProvider, sanitize_tts_text
 from owner_voice_pet.vad import EnergyVadProvider, VadRecorder
 from owner_voice_pet.wakeword import KeywordWakeWordProvider
 
@@ -60,6 +60,12 @@ class PipelineLlmTtsTests(unittest.TestCase):
         self.assertEqual(buffer.feed("剩余").copy(), [])
         self.assertEqual(buffer.flush(), ["剩余"])
 
+    def test_sanitize_tts_text_removes_unspeakable_expression_tokens(self) -> None:
+        self.assertEqual(sanitize_tts_text("你好 😊"), "你好")
+        self.assertEqual(sanitize_tts_text("好的![开心](x.png)"), "好的")
+        self.assertEqual(sanitize_tts_text("没问题[捂脸]，我来帮你。"), "没问题，我来帮你。")
+        self.assertEqual(sanitize_tts_text("😂😂"), "")
+
     def test_sine_tts_generates_non_empty_audio(self) -> None:
         provider = SineTtsProvider()
         provider.load()
@@ -126,6 +132,22 @@ class PipelineLlmTtsTests(unittest.TestCase):
         self.assertEqual(len(transport.played_segments), 1)
         self.assertEqual(pipeline.context.messages()[-1].role, "assistant")
 
+    def test_pipeline_sanitizes_tts_text_and_assistant_context(self) -> None:
+        frames = [
+            frame(0, 0, {"wake_word": "小杰小杰", "wake_confidence": 0.95}),
+            frame(1, 20, {"speech": True, "transcript": "你好"}),
+            frame(2, 40, {"speech": True}),
+            frame(3, 60, {"speech": False}),
+            frame(4, 80, {"speech": False}),
+        ]
+        pipeline, transport = make_pipeline(frames, MockLlmProvider(["你好 😊。没问题[捂脸]，我来帮你。"]))
+        result = pipeline.run_once()
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.assistant_text, "你好。没问题，我来帮你。")
+        self.assertEqual(transport.played_segments[0].metadata["text"], "你好。没问题，我来帮你。")
+        self.assertEqual(pipeline.context.messages()[-1].content, "你好。没问题，我来帮你。")
+
     def test_pipeline_skips_llm_on_empty_transcript(self) -> None:
         llm = MockLlmProvider(["不应调用"])
         frames = [