[全双工运行入口]：完成run-agent-live常驻监听闭环，包含麦克风采集、语音识别和终端状态输出

tests/test_full_duplex_integration.py

@@ -23,11 +23,52 @@ from owner_voice_pet.full_duplex_testing import (
     diagnostics_contain_sensitive_data,
     sanitize_diagnostics,
 )
-from owner_voice_pet.models import Message, PipelineState
+from owner_voice_pet.llm import MockLlmProvider
+from owner_voice_pet.models import AudioFrame, Message, PipelineState
+from owner_voice_pet.stt import MetadataSttProvider
 from owner_voice_pet.tool_router import MemorySearchTool, ToolCallRequest, ToolContext, ToolRouter
+from owner_voice_pet.transport import MemoryAudioTransport
+from owner_voice_pet.tts import SineTtsProvider
+from owner_voice_pet.vad import EnergyVadProvider, VadRecorder
 
 
 class FullDuplexIntegrationTests(unittest.TestCase):
+    def test_run_agent_live_runtime_once_consumes_audio_and_replies(self) -> None:
+        frames = [
+            AudioFrame(b"\xff\x7f", 16000, 1, 0, 1, {"duration_ms": 20, "speech": True, "transcript": "直接提问"}),
+            AudioFrame(b"\xff\x7f", 16000, 1, 20, 2, {"duration_ms": 20, "speech": True, "transcript": "直接提问"}),
+            AudioFrame(b"\x00\x00", 16000, 1, 40, 3, {"duration_ms": 20, "speech": False, "transcript": "直接提问"}),
+            AudioFrame(b"\x00\x00", 16000, 1, 60, 4, {"duration_ms": 20, "speech": False, "transcript": "直接提问"}),
+        ]
+        transport = MemoryAudioTransport(frames, flush_clears_input=False)
+        runtime = FullDuplexAgentRuntime(
+            config=AppConfig(
+                audio_apm_provider="fake",
+                audio_apm_required=False,
+                memory_enabled=False,
+                tool_router_enabled=False,
+                vad_min_duration_ms=40,
+                vad_end_silence_ms=40,
+                end_chime_enabled=False,
+            ),
+            transport=transport,
+            vad_recorder=VadRecorder(EnergyVadProvider(), min_duration_ms=40, end_silence_ms=40),
+            stt=MetadataSttProvider(),
+            llm=MockLlmProvider(["这是全双工回答。"]),
+            tts=SineTtsProvider(),
+        )
+
+        summary = runtime.run(once=True)
+
+        self.assertEqual(summary.completed_turns, 1)
+        self.assertEqual(summary.failed_turns, 0)
+        self.assertEqual([message.role for message in runtime.context.messages()], ["user", "assistant"])
+        self.assertEqual(runtime.context.messages()[0].content, "直接提问")
+        self.assertEqual(runtime.context.messages()[1].content, "这是全双工回答。")
+        self.assertGreaterEqual(len(transport.played_segments), 1)
+        self.assertIsNotNone(runtime.audio_hub)
+        self.assertGreater(runtime.audio_hub.processed_capture.frame_count, 0)
+
     def test_fake_apm_echo_does_not_trigger_interruption(self) -> None:
         fixture = build_fake_full_duplex_audio_fixture()
         apm = FakeWebRtcAudioProcessingProvider()