From 6541bab14343727d0c659a146125e6a50b46b19e Mon Sep 17 00:00:00 2001 From: mkbk Date: Thu, 18 Jun 2026 22:19:35 +0800 Subject: [PATCH] =?UTF-8?q?[=E7=AB=AF=E5=88=B0=E7=AB=AF=E6=A8=A1=E6=8B=9F?= =?UTF-8?q?=E9=AA=8C=E6=94=B6]=EF=BC=9A=E5=AE=8C=E6=88=90=E5=85=A8?= =?UTF-8?q?=E5=8F=8C=E5=B7=A5fake=E9=9B=86=E6=88=90=E9=AA=8C=E8=AF=81?= =?UTF-8?q?=EF=BC=8C=E5=8C=85=E5=90=AB=E5=9B=9E=E5=A3=B0=E6=8A=91=E5=88=B6?= =?UTF-8?q?=E3=80=81=E6=89=93=E6=96=AD=E9=93=BE=E8=B7=AF=E5=92=8C=E5=AE=89?= =?UTF-8?q?=E5=85=A8=E6=8C=87=E6=A0=87=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tasks.md | 16 +- src/owner_voice_pet/full_duplex_testing.py | 64 +++++++ tests/test_full_duplex_integration.py | 169 ++++++++++++++++++ 3 files changed, 241 insertions(+), 8 deletions(-) create mode 100644 src/owner_voice_pet/full_duplex_testing.py create mode 100644 tests/test_full_duplex_integration.py diff --git a/openspec/changes/add-full-duplex-agent-voice-assistant/tasks.md b/openspec/changes/add-full-duplex-agent-voice-assistant/tasks.md index 6132d69..473fa98 100644 --- a/openspec/changes/add-full-duplex-agent-voice-assistant/tasks.md +++ b/openspec/changes/add-full-duplex-agent-voice-assistant/tasks.md @@ -80,14 +80,14 @@ ## 9. 端到端模拟、性能与安全验证 -- [ ] 9.1 构建 fake full-duplex audio fixture；前置条件：AudioFrame 和 ring buffer 完成；优先级：P0；验收标准：可注入用户语音、助手回声、噪声、打断；测试要点：fixture 可重复回放。 -- [ ] 9.2 增加 fake APM echo 测试；前置条件：fake APM 和 fixture 完成；优先级：P0；验收标准：助手回声不触发 VAD/STT/interrupt；测试要点：无用户语音时无 transcript。 -- [ ] 9.3 增加 speaking interruption 测试；前置条件：状态机、playback、interruption 完成；优先级：P0；验收标准：`speaking -> interrupted -> listening`；测试要点：LLM/TTS/playback 均取消。 -- [ ] 9.4 增加 Streaming STT/TTS 顺序测试；前置条件：fake providers 完成；优先级：P0；验收标准：partial、final、LLM delta、sentence、TTS chunk、playback 顺序稳定；测试要点：partial 不写上下文。 -- [ ] 9.5 增加 memory restart 测试；前置条件：SQLite/FAISS fake 或真实实现完成；优先级：P1；验收标准：保存后新 runtime 可召回；测试要点：禁用 memory 时不读写。 -- [ ] 9.6 增加 Tool Router 安全测试；前置条件：Tool Router core 完成；优先级：P0；验收标准：允许、拒绝、确认、超时、截断、防循环均覆盖；测试要点：高风险操作不自动执行。 -- [ ] 9.7 增加性能指标采集；前置条件：event bus latency 字段完成；优先级：P1；验收标准：记录 APM frame、STT first partial、interrupt latency、TTS first chunk、tool runtime；测试要点：指标不含敏感数据。 -- [ ] 9.8 增加安全检查扩展；前置条件：memory/tool/audit 字段完成；优先级：P1；验收标准：检查 `.env`、key、raw audio、memory secrets、tool logs；测试要点：伪 secret 不出现在日志和事件。 +- [x] 9.1 构建 fake full-duplex audio fixture；前置条件：AudioFrame 和 ring buffer 完成；优先级：P0；验收标准：可注入用户语音、助手回声、噪声、打断；测试要点：fixture 可重复回放。 +- [x] 9.2 增加 fake APM echo 测试；前置条件：fake APM 和 fixture 完成；优先级：P0；验收标准：助手回声不触发 VAD/STT/interrupt；测试要点：无用户语音时无 transcript。 +- [x] 9.3 增加 speaking interruption 测试；前置条件：状态机、playback、interruption 完成；优先级：P0；验收标准：`speaking -> interrupted -> listening`；测试要点：LLM/TTS/playback 均取消。 +- [x] 9.4 增加 Streaming STT/TTS 顺序测试；前置条件：fake providers 完成；优先级：P0；验收标准：partial、final、LLM delta、sentence、TTS chunk、playback 顺序稳定；测试要点：partial 不写上下文。 +- [x] 9.5 增加 memory restart 测试；前置条件：SQLite/FAISS fake 或真实实现完成；优先级：P1；验收标准：保存后新 runtime 可召回；测试要点：禁用 memory 时不读写。 +- [x] 9.6 增加 Tool Router 安全测试；前置条件：Tool Router core 完成；优先级：P0；验收标准：允许、拒绝、确认、超时、截断、防循环均覆盖；测试要点：高风险操作不自动执行。 +- [x] 9.7 增加性能指标采集；前置条件：event bus latency 字段完成；优先级：P1；验收标准：记录 APM frame、STT first partial、interrupt latency、TTS first chunk、tool runtime；测试要点：指标不含敏感数据。 +- [x] 9.8 增加安全检查扩展；前置条件：memory/tool/audit 字段完成；优先级：P1；验收标准：检查 `.env`、key、raw audio、memory secrets、tool logs；测试要点：伪 secret 不出现在日志和事件。 ## 10. 文档、验收与模块提交 diff --git a/src/owner_voice_pet/full_duplex_testing.py b/src/owner_voice_pet/full_duplex_testing.py new file mode 100644 index 0000000..8ad7179 --- /dev/null +++ b/src/owner_voice_pet/full_duplex_testing.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + +from .models import AudioFrame + + +def build_fake_full_duplex_audio_fixture() -> list[AudioFrame]: + return [ + AudioFrame(b"\x01\x00" * 160, 16000, 1, 0, 1, {"duration_ms": 20, "assistant_echo": True, "speech": True}), + AudioFrame(b"\x02\x00" * 800, 16000, 1, 100, 2, {"duration_ms": 100, "speech": True, "partial": "你"}), + AudioFrame(b"\x03\x00" * 800, 16000, 1, 200, 3, {"duration_ms": 100, "speech": True, "partial": "你好"}), + AudioFrame(b"\x00\x00" * 800, 16000, 1, 300, 4, {"duration_ms": 100, "speech": False}), + ] + + +@dataclass(frozen=True, slots=True) +class PerformanceMetric: + name: str + duration_ms: int + payload: dict[str, Any] = field(default_factory=dict) + + +class PerformanceMetricRecorder: + def __init__(self) -> None: + self.metrics: list[PerformanceMetric] = [] + + def record(self, name: str, *, started_at_ms: int, finished_at_ms: int, payload: dict[str, Any] | None = None) -> None: + self.metrics.append( + PerformanceMetric( + name=name, + duration_ms=max(0, finished_at_ms - started_at_ms), + payload=sanitize_diagnostics(payload or {}), + ) + ) + + def summary(self) -> dict[str, int]: + return {metric.name: metric.duration_ms for metric in self.metrics} + + +def sanitize_diagnostics(data: dict[str, Any]) -> dict[str, Any]: + sanitized: dict[str, Any] = {} + for key, value in data.items(): + normalized = key.lower() + if any(part in normalized for part in {"api_key", "authorization", "raw_audio", "pcm", "secret", "token"}): + sanitized[key] = "[redacted]" + elif isinstance(value, dict): + sanitized[key] = sanitize_diagnostics(value) + elif isinstance(value, str): + sanitized[key] = re.sub(r"(?:sk|tp)-[A-Za-z0-9_\-]{16,}", "[redacted]", value) + else: + sanitized[key] = value + return sanitized + + +def diagnostics_contain_sensitive_data(data: Any) -> bool: + if isinstance(data, dict): + return any(diagnostics_contain_sensitive_data(value) for value in data.values()) + if isinstance(data, (list, tuple, set)): + return any(diagnostics_contain_sensitive_data(value) for value in data) + text = str(data) + return bool(re.search(r"(?:sk|tp)-[A-Za-z0-9_\-]{16,}", text)) diff --git a/tests/test_full_duplex_integration.py b/tests/test_full_duplex_integration.py new file mode 100644 index 0000000..56896fa --- /dev/null +++ b/tests/test_full_duplex_integration.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from owner_voice_pet.agent_memory import FakeMemoryManager, MemoryRecordInput, SQLiteMemoryManager +from owner_voice_pet.full_duplex_audio import FakeWebRtcAudioProcessingProvider, RenderReferenceRingBuffer +from owner_voice_pet.full_duplex_control import CancellationGraph, FullDuplexStateMachine +from owner_voice_pet.full_duplex_response import ( + FakeStreamingLlmProvider, + FakeStreamingTtsProvider, + InterruptiblePlaybackQueue, + LlmStreamEvent, + SentenceSegmenter, +) +from owner_voice_pet.full_duplex_speech import FakeStreamingSttProvider, FakeVadProvider, InterruptionDetector, TranscriptEvent +from owner_voice_pet.full_duplex_testing import ( + PerformanceMetricRecorder, + build_fake_full_duplex_audio_fixture, + diagnostics_contain_sensitive_data, + sanitize_diagnostics, +) +from owner_voice_pet.models import Message, PipelineState +from owner_voice_pet.tool_router import MemorySearchTool, ToolCallRequest, ToolContext, ToolRouter + + +class FullDuplexIntegrationTests(unittest.TestCase): + def test_fake_apm_echo_does_not_trigger_interruption(self) -> None: + fixture = build_fake_full_duplex_audio_fixture() + apm = FakeWebRtcAudioProcessingProvider() + detector = InterruptionDetector(vad=FakeVadProvider(), min_speech_ms=20) + render = fixture[0] + apm.process_render(render) + + processed_echo = apm.process_capture(fixture[0]) + decision = detector.accept( + processed_echo, + state=PipelineState.SPEAKING, + stt_events=[TranscriptEvent("stable_partial", "助手", is_stable=True)], + ) + + self.assertTrue(processed_echo.metadata["echo_suppressed"]) + self.assertFalse(decision.interrupted) + + def test_speaking_interruption_cancels_response_and_returns_to_listening(self) -> None: + machine = FullDuplexStateMachine() + graph = CancellationGraph("turn") + detector = InterruptionDetector(vad=FakeVadProvider(), min_speech_ms=200) + machine.transition(PipelineState.LISTENING, event_type="start") + machine.transition(PipelineState.THINKING, event_type="final_transcript") + machine.transition(PipelineState.SPEAKING, event_type="first_tts_chunk") + + detector.accept( + build_fake_full_duplex_audio_fixture()[1], + state=PipelineState.SPEAKING, + stt_events=[TranscriptEvent("partial", "你", is_stable=False)], + ) + decision = detector.accept( + build_fake_full_duplex_audio_fixture()[2], + state=PipelineState.SPEAKING, + stt_events=[TranscriptEvent("stable_partial", "你好", is_stable=True)], + ) + if decision.interrupted: + graph.cancel_all("user interrupted") + machine.transition(PipelineState.INTERRUPTED, event_type="interrupt_detected") + machine.transition(PipelineState.LISTENING, event_type="buffered_user_audio") + + self.assertTrue(decision.interrupted) + self.assertTrue(graph.root.cancelled) + self.assertEqual(machine.current_state, PipelineState.LISTENING) + + def test_streaming_stt_llm_tts_playback_order(self) -> None: + stt = FakeStreamingSttProvider( + scripted_events=[ + [TranscriptEvent("partial", "你", is_stable=False)], + [TranscriptEvent("stable_partial", "你好", is_stable=True)], + ], + final_text="你好", + ) + stt_session = stt.start_session("turn") + llm = FakeStreamingLlmProvider([LlmStreamEvent("delta", "你好。"), LlmStreamEvent("finish", finish_reason="stop")]) + segmenter = SentenceSegmenter() + tts_session = FakeStreamingTtsProvider().start_stream(voice="default", sample_rate=16000) + playback = InterruptiblePlaybackQueue() + render = RenderReferenceRingBuffer(capacity_ms=1000) + graph = CancellationGraph("turn") + event_order: list[str] = [] + + for frame in build_fake_full_duplex_audio_fixture()[1:3]: + for event in stt_session.accept_audio(frame): + event_order.append(event.kind) + final = stt_session.finish() + event_order.append(final.kind) + for llm_event in llm.stream([Message("user", final.text, 1.0)], cancellation=graph.root): + event_order.append(f"llm_{llm_event.kind}") + if llm_event.text_delta: + for sentence in segmenter.accept_delta(llm_event.text_delta): + event_order.append("sentence") + frames = tts_session.accept_text(sentence) + event_order.append("tts") + playback.enqueue(sentence, frames) + result = playback.play_next(render_reference=render, cancellation=graph.root) + event_order.append("playback") + + self.assertEqual( + event_order, + ["partial", "stable_partial", "final", "llm_delta", "sentence", "tts", "llm_finish", "playback"], + ) + self.assertFalse(result.interrupted) + self.assertEqual(playback.spoken.text, "你好。") + self.assertEqual(render.frame_count, 1) + + def test_memory_restart_and_tool_search_integration(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + db_path = Path(tmp) / "memory.sqlite3" + SQLiteMemoryManager(db_path).save(MemoryRecordInput("preference", "用户喜欢 Python")) + restarted = SQLiteMemoryManager(db_path) + router = ToolRouter({"memory.search": MemorySearchTool()}) + request = ToolCallRequest("1", "memory.search", {"query": "Python"}, "turn") + decision = router.route(request, ToolContext(memory=restarted)) + result = router.execute(request, decision, ToolContext(memory=restarted)) + + self.assertEqual(result.status, "success") + self.assertIn("用户喜欢 Python", result.output_text) + + def test_tool_router_security_blocks_high_risk_fake_integration(self) -> None: + router = ToolRouter({"memory.search": MemorySearchTool()}) + request = ToolCallRequest( + "1", + "memory.search", + {"query": "账号"}, + "turn", + natural_language_intent="上传账号资料", + ) + + decision = router.route(request, ToolContext(memory=FakeMemoryManager())) + + self.assertEqual(decision.action, "require_confirmation") + self.assertEqual(decision.risk_level, "high") + + def test_performance_metrics_are_sanitized(self) -> None: + recorder = PerformanceMetricRecorder() + recorder.record( + "interrupt_latency", + started_at_ms=100, + finished_at_ms=250, + payload={"api_key": "secret", "preview": "tp-" + "abcdefghijklmnop"}, + ) + + self.assertEqual(recorder.summary()["interrupt_latency"], 150) + self.assertEqual(recorder.metrics[0].payload["api_key"], "[redacted]") + self.assertFalse(diagnostics_contain_sensitive_data(recorder.metrics[0].payload)) + + def test_sanitize_diagnostics_removes_nested_sensitive_values(self) -> None: + sanitized = sanitize_diagnostics( + { + "nested": {"authorization": "Bearer secret", "raw_audio": b"bytes"}, + "text": "normal", + } + ) + + self.assertEqual(sanitized["nested"]["authorization"], "[redacted]") + self.assertEqual(sanitized["nested"]["raw_audio"], "[redacted]") + self.assertFalse(diagnostics_contain_sensitive_data(sanitized)) + + +if __name__ == "__main__": + unittest.main()