From 98ae711971d64363eda52cb88ea0c9869b438066 Mon Sep 17 00:00:00 2001 From: mkbk Date: Fri, 19 Jun 2026 12:43:44 +0800 Subject: [PATCH] =?UTF-8?q?[=E5=85=A8=E5=8F=8C=E5=B7=A5=E8=87=AA=E6=B5=8B]?= =?UTF-8?q?=EF=BC=9A=E5=AE=8C=E6=88=90Agent=E8=87=AA=E6=88=91=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E4=BD=93=E7=B3=BB=EF=BC=8C=E5=8C=85=E5=90=AB=E5=9B=9E?= =?UTF-8?q?=E5=A3=B0=E3=80=81=E6=89=93=E6=96=AD=E3=80=81=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=E5=92=8C=E5=B7=A5=E5=85=B7=E5=9C=BA=E6=99=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 4 +- README.md | 34 ++- .../tasks.md | 6 +- src/owner_voice_pet/__init__.py | 3 + src/owner_voice_pet/cli.py | 22 ++ src/owner_voice_pet/self_tests.py | 209 ++++++++++++++++++ tests/test_cli_acceptance.py | 44 ++++ 7 files changed, 309 insertions(+), 13 deletions(-) create mode 100644 src/owner_voice_pet/self_tests.py diff --git a/.env.example b/.env.example index 1927638..5aa01f5 100644 --- a/.env.example +++ b/.env.example @@ -69,11 +69,11 @@ OWNER_INTERRUPT_TARGET_LATENCY_MS=200 OWNER_STREAMING_STT_PROVIDER=faster_whisper OWNER_STREAMING_STT_PRODUCT_CANDIDATE=sensevoice OWNER_STREAMING_TTS_PROVIDER=cosyvoice -OWNER_MEMORY_ENABLED=0 +OWNER_MEMORY_ENABLED=1 OWNER_MEMORY_PROVIDER=faiss_sqlite OWNER_MEMORY_TOP_K=5 OWNER_MEMORY_AUTO_SAVE_SENSITIVE=0 -OWNER_TOOL_ROUTER_ENABLED=0 +OWNER_TOOL_ROUTER_ENABLED=1 OWNER_TOOL_MAX_CALLS_PER_TURN=5 OWNER_TOOL_TIMEOUT_MS=30000 OWNER_OPENINTERPRETER_ENABLED=0 diff --git a/README.md b/README.md index 112b83a..bc0ee17 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,10 @@ - `owner_voice_pet run-live`：真实常驻语音循环。 - `owner_voice_pet run-live --once`：只跑一轮，便于验收。 -- `owner_voice_pet run-agent-live --check-config`：全双工 Agent 新入口的配置检查。当前用于迁移期验收，真实运行仍走 `run-live`。 +- `owner_voice_pet run-agent-live`：完整全双工 Agent 主入口，启动后直接 listening，播放中可被有效用户语音打断。 +- `owner_voice_pet run-agent-live --check-config`：只检查全双工 Agent 配置和 APM 就绪状态，不打开麦克风。 +- `owner_voice_pet agent-self-test --profile full-duplex --turns 3`：无人值守自测 STT、LLM、TTS、打断、记忆和工具路由。 +- `owner_voice_pet audio-self-test --duration 10 --check-echo`：检查设备、WebRTC APM、回声抑制和打断延迟。 - `.env` 直接读取配置，不要求导出 shell 环境变量。 - 唤醒词检测使用本地 `sherpa-onnx` KWS 模型，不走云端 ASR。 - `OWNER_SPEECH_PROVIDER=local`：默认除 LLM 外全用本地语音链路；`cloud` 仅作为显式兼容选项。 @@ -97,11 +100,11 @@ OWNER_INTERRUPT_TARGET_LATENCY_MS=200 OWNER_STREAMING_STT_PROVIDER=faster_whisper OWNER_STREAMING_STT_PRODUCT_CANDIDATE=sensevoice OWNER_STREAMING_TTS_PROVIDER=cosyvoice -OWNER_MEMORY_ENABLED=0 +OWNER_MEMORY_ENABLED=1 OWNER_MEMORY_PROVIDER=faiss_sqlite OWNER_MEMORY_TOP_K=5 OWNER_MEMORY_AUTO_SAVE_SENSITIVE=0 -OWNER_TOOL_ROUTER_ENABLED=0 +OWNER_TOOL_ROUTER_ENABLED=1 OWNER_TOOL_MAX_CALLS_PER_TURN=5 OWNER_TOOL_TIMEOUT_MS=30000 OWNER_OPENINTERPRETER_ENABLED=0 @@ -130,7 +133,7 @@ OWNER_COMPUTER_CONTROL_ENABLED=0 .venv/bin/python -m owner_voice_pet run-agent-live ``` -`run-agent-live` 不需要先说唤醒词，启动后会直接进入 listening。你可以直接说问题；助手播放回复时，后台麦克风仍在监听。如果你插话，runtime 会用 VAD + 软件 render reference/音色门控判断是不是用户声音；确认后会在下一个播放 chunk 边界停止播报，把打断音频接入下一轮 STT，并继续对话。 +`run-agent-live` 不需要先说唤醒词，启动后会直接进入 listening。完整模式默认要求 `OWNER_AUDIO_APM_PROVIDER=webrtc` 和 `OWNER_AUDIO_APM_REQUIRED=1`，麦克风 capture 必须经过 WebRTC APM 的 AEC/NS/AGC 后再进入 VAD、STT 和打断检测；播放 PCM 会同步写入 render reference。若本机没有真实 WebRTC APM provider，入口会以 `AUDIO_APM_UNAVAILABLE` 明确失败，不再回退到旧 `run-live` 或伪全双工。 只检查全双工配置、不打开麦克风： @@ -144,14 +147,29 @@ OWNER_COMPUTER_CONTROL_ENABLED=0 .venv/bin/python -m owner_voice_pet run-live ``` -全双工 Agent 第一版使用软件回声抑制，不强制新增 WebRTC APM 重依赖：播放 PCM 会作为 render reference 供打断门控使用，只有助手回放时不应触发打断，用户声音叠加回放时应停止播报。`OWNER_AUDIO_APM_PROVIDER=webrtc` 仍代表后续目标音频底座，`OWNER_LLM_STREAMING_ENABLED=1` 控制 LLM 以流式响应供句子级 TTS 消费，旧变量 `OWNER_LLM_STREAM` 仍兼容；`OWNER_STREAMING_STT_PROVIDER=faster_whisper` 和 `OWNER_STREAMING_TTS_PROVIDER=cosyvoice` 是后续 provider 目标；`OWNER_MEMORY_ENABLED=0`、`OWNER_TOOL_ROUTER_ENABLED=0`、`OWNER_OPENINTERPRETER_ENABLED=0`、`OWNER_BROWSER_PLAYWRIGHT_ENABLED=0`、`OWNER_COMPUTER_CONTROL_ENABLED=0` 默认关闭，避免尚未完成安全边界前执行长期记忆或工具任务。 +全双工 Agent 当前架构使用单一 `AudioHub` 拥有麦克风输入，VAD、STT、InterruptController 和诊断订阅各自独立的 processed capture cursor，不再抢读同一个 Transport 队列。打断不等待 STT partial：在 `thinking/speaking/tool_running` 中，只要 processed capture 上的有效用户语音达到阈值，就取消当前 LLM/TTS/playback/tool 子图，并把已确认的用户音频缓存给下一轮输入。`OWNER_LLM_STREAMING_ENABLED=1` 控制 LLM 以流式响应供句子级 TTS 消费，旧变量 `OWNER_LLM_STREAM` 仍兼容；`OWNER_STREAMING_STT_PROVIDER=faster_whisper` 和 `OWNER_STREAMING_TTS_PROVIDER=cosyvoice` 是 provider 目标；`OWNER_MEMORY_ENABLED=1` 和 `OWNER_TOOL_ROUTER_ENABLED=1` 默认开启，`OWNER_OPENINTERPRETER_ENABLED=0`、`OWNER_BROWSER_PLAYWRIGHT_ENABLED=0`、`OWNER_COMPUTER_CONTROL_ENABLED=0` 默认关闭，避免未确认的外部控制自动执行。 + +无人值守 Agent 自测： + +```bash +.venv/bin/python -m owner_voice_pet agent-self-test --profile full-duplex --turns 3 +``` + +音频/APM 诊断： + +```bash +.venv/bin/python -m owner_voice_pet audio-self-test --duration 10 --check-echo +``` + +如果当前机器没有真实 WebRTC APM binding，`audio-self-test` 会返回 `success=false` 和 `AUDIO_APM_UNAVAILABLE`；这表示完整全双工音频底座尚未满足，不应把 fake APM 结果当成人工验收通过。需要做确定性开发自测时，可以临时使用 `OWNER_AUDIO_APM_PROVIDER=fake`。 当前已落地的全双工基础模块： -- `full_duplex_audio`：音频帧 fixture、capture/render ring buffer、fake WebRTC APM、APM 探针和 fallback 决策。 +- `full_duplex_audio`：AudioHub、多消费者 capture/render ring buffer、fake WebRTC APM、APM 探针和 required startup 决策。 - `full_duplex_control`：全双工状态机、事件诊断字段、取消 token graph 和恢复协调器。 -- `full_duplex_speech`：VAD provider contract、Silero VAD 边界、Streaming STT contract、fake STT 和 interruption detector。 -- `full_duplex_response`：LLM streaming contract、句子切分、TTS 文本净化、fake Streaming TTS 和可中断播放队列。 +- `full_duplex_speech`：VAD provider contract、Silero VAD 边界、Streaming STT worker、fake STT、InterruptionDetector 和 InterruptController。 +- `full_duplex_response`：LLM streaming contract、句子切分、TTS 文本净化、Streaming TTS wrapper、fake Streaming TTS 和可中断播放队列。 +- `full_duplex_runtime`：新 `run-agent-live` runtime 边界，串联 AudioHub、取消图、流式回复、长期记忆和 ToolRouter。 - `agent_memory`：SQLite memory schema、FAISS index manifest 校验、fake/disabled memory manager、敏感写入策略和 memory recall 注入。 - `tool_router`：结构化工具调用、风险分类、预算防循环、审计脱敏、`memory.search`、`memory.save` 和 `shell.readonly`。 - `external_adapters`：Open Interpreter、Playwright 和 Computer Control 的默认关闭边界。 diff --git a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md index 56e65ad..2e6ae1d 100644 --- a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md +++ b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md @@ -38,8 +38,8 @@ ## 6. 自我测试、文档与最终验收 -- [ ] 6.1 新增 `agent-self-test`；前置条件：Phase 5；优先级：P0；验收标准：三轮覆盖 STT/LLM/TTS/barge-in/memory/tool；测试要点：JSON success。 -- [ ] 6.2 新增 `audio-self-test`；前置条件：AudioHub/APM；优先级：P0；验收标准：输出 provider/device/echo/latency；测试要点：无 APM 时明确失败。 -- [ ] 6.3 更新 README 和 `.env.example`；前置条件：6.1-6.2；优先级：P0；验收标准：入口和配置无冲突；测试要点：命令示例准确。 +- [x] 6.1 新增 `agent-self-test`；前置条件：Phase 5；优先级：P0；验收标准：三轮覆盖 STT/LLM/TTS/barge-in/memory/tool；测试要点：JSON success。 +- [x] 6.2 新增 `audio-self-test`；前置条件：AudioHub/APM；优先级：P0；验收标准：输出 provider/device/echo/latency；测试要点：无 APM 时明确失败。 +- [x] 6.3 更新 README 和 `.env.example`；前置条件：6.1-6.2；优先级：P0；验收标准：入口和配置无冲突；测试要点：命令示例准确。 - [ ] 6.4 执行最终门禁；前置条件：全部实现完成；优先级：P0；验收标准：compileall、unittest、self-test、security、model、device、OpenSpec strict 全通过；测试要点：命令输出。 - [ ] 6.5 Phase 6 提交；前置条件：6.1-6.4；优先级：P0；验收标准：中文提交 `[全双工自测]...`；测试要点：`git status --short` 为空。 diff --git a/src/owner_voice_pet/__init__.py b/src/owner_voice_pet/__init__.py index b845c74..f3273eb 100644 --- a/src/owner_voice_pet/__init__.py +++ b/src/owner_voice_pet/__init__.py @@ -100,6 +100,7 @@ from .llm import MockLlmProvider, OpenAICompatibleLlmProvider from .pipeline import PipelineResult, VoicePipeline from .runtime import LiveVoiceRuntime, RuntimeSummary, TerminalRuntimeReporter, TurnResult, build_live_runtime from .simulation import run_simulated_live +from .self_tests import run_agent_self_test, run_audio_self_test from .tts import CloudTtsProvider, MacSayTtsProvider, SentenceBuffer, SineTtsProvider, make_end_chime, make_prompt_chime, sanitize_tts_text from .assets import validate_pet_assets from .ui import ConsolePetWindow, PetStateController, PetVisualState @@ -199,6 +200,8 @@ __all__ = [ "TurnResult", "build_live_runtime", "run_simulated_live", + "run_agent_self_test", + "run_audio_self_test", "CloudTtsProvider", "MacSayTtsProvider", "SentenceBuffer", diff --git a/src/owner_voice_pet/cli.py b/src/owner_voice_pet/cli.py index 2f7adf0..b199dc1 100644 --- a/src/owner_voice_pet/cli.py +++ b/src/owner_voice_pet/cli.py @@ -19,6 +19,7 @@ from .pipeline import VoicePipeline from .real_live_check import run_real_live_check from .runtime import build_live_runtime from .simulation import run_simulated_live +from .self_tests import run_agent_self_test, run_audio_self_test from .speech_models import check_speech_models, model_status_errors from .stt import MetadataSttProvider, SherpaOnnxSttProvider from .transport import MemoryAudioTransport, sounddevice_device_report @@ -56,6 +57,12 @@ def main(argv: list[str] | None = None) -> int: real_check.add_argument("--wake-text", default="小杰小杰。", help="Generated wake utterance") real_check.add_argument("--question", action="append", default=None, help="Generated user question; can be repeated") real_check.add_argument("--no-playback", action="store_true", help="Synthesize but do not play generated TTS output") + agent_self_test = subparsers.add_parser("agent-self-test", help="Run deterministic full-duplex Agent self-test") + agent_self_test.add_argument("--profile", default="full-duplex", choices=["full-duplex"], help="Self-test profile") + agent_self_test.add_argument("--turns", type=int, default=3, help="Number of simulated turns. Default: 3") + audio_self_test = subparsers.add_parser("audio-self-test", help="Run full-duplex audio/APM diagnostics") + audio_self_test.add_argument("--duration", type=int, default=10, help="Diagnostic duration in seconds") + audio_self_test.add_argument("--check-echo", action="store_true", help="Check assistant echo suppression path") smoke = subparsers.add_parser("llm-smoke", help="Call configured OpenAI/NewAPI endpoint") smoke.add_argument("--message", default="用一句中文回复：小杰在线。") smoke.add_argument("--no-stream", action="store_true") @@ -296,6 +303,21 @@ def main(argv: list[str] | None = None) -> int: print(json.dumps(data, ensure_ascii=False, sort_keys=True)) return 0 if data["success"] else 1 + if args.command == "agent-self-test": + data = run_agent_self_test(profile=args.profile, turns=args.turns) + print(json.dumps(data, ensure_ascii=False, sort_keys=True)) + return 0 if data["success"] else 1 + + if args.command == "audio-self-test": + config = AppConfig.from_dotenv(args.env_file) + data = run_audio_self_test( + config=config, + duration_s=args.duration, + check_echo=args.check_echo, + ) + print(json.dumps(data, ensure_ascii=False, sort_keys=True)) + return 0 if data["success"] else 1 + if args.command == "acceptance": result = run_acceptance() print(json.dumps(result, ensure_ascii=False, sort_keys=True)) diff --git a/src/owner_voice_pet/self_tests.py b/src/owner_voice_pet/self_tests.py new file mode 100644 index 0000000..2e8c1de --- /dev/null +++ b/src/owner_voice_pet/self_tests.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +from .agent_memory import FakeMemoryManager, MemoryRecordInput +from .config import AppConfig +from .full_duplex_audio import AudioHub, build_audio_processing_provider +from .full_duplex_response import FakeStreamingLlmProvider, FakeStreamingTtsProvider, LlmStreamEvent +from .full_duplex_runtime import FullDuplexAgentRuntime +from .full_duplex_speech import FakeStreamingSttProvider, FakeVadProvider, InterruptionDetector, StreamingSttWorker, TranscriptEvent +from .models import AudioFrame, Message, PipelineState, ProviderError +from .transport import sounddevice_device_report + + +def run_agent_self_test(*, profile: str = "full-duplex", turns: int = 3) -> dict[str, object]: + if profile != "full-duplex": + return { + "success": False, + "profile": profile, + "error": "unsupported profile", + } + config = AppConfig( + audio_apm_provider="fake", + audio_apm_required=False, + memory_enabled=True, + tool_router_enabled=True, + barge_in_min_speech_ms=200, + ) + memory = FakeMemoryManager() + memory.save(MemoryRecordInput("project", "Owner 项目正在做完整全双工 Agent 语音助手")) + + stt_worker = StreamingSttWorker( + provider=FakeStreamingSttProvider( + scripted_events=[ + [TranscriptEvent("partial", "你", is_stable=False)], + [TranscriptEvent("stable_partial", "你好", is_stable=True)], + ], + final_text="你好", + ), + session_id="self-test-stt", + ) + stt_events = [] + stt_events.extend(stt_worker.accept_frame(_frame(1, 0, speech=True, partial="你"))) + stt_events.extend(stt_worker.accept_frame(_frame(2, 100, speech=True, partial="你好"))) + final = stt_worker.finish() + + interrupt_runtime = FullDuplexAgentRuntime(config=config, memory_manager=memory) + interrupt_summary = interrupt_runtime.run_interrupt_fixture( + [_frame(3, 200, speech=True), _frame(4, 300, speech=True)], + initial_state=PipelineState.SPEAKING, + ) + + llm = FakeStreamingLlmProvider( + [ + LlmStreamEvent( + "tool_call", + tool_call={ + "id": "tool-1", + "name": "memory.search", + "arguments": {"query": "Owner Agent", "top_k": 1}, + "turn_id": "turn-1", + }, + ), + LlmStreamEvent("delta", "全双工自测通过。"), + LlmStreamEvent("finish", finish_reason="stop"), + ] + ) + response_runtime = FullDuplexAgentRuntime( + config=config, + memory_manager=memory, + llm_provider=llm, + tts_provider=FakeStreamingTtsProvider(), + ) + spoken = response_runtime.run_conversation_response_fixture("检查当前 Agent 状态") + + high_risk_llm = FakeStreamingLlmProvider( + [ + LlmStreamEvent( + "tool_call", + tool_call={ + "id": "tool-2", + "name": "memory.search", + "arguments": {"query": "账号"}, + "natural_language_intent": "上传账号资料", + "turn_id": "turn-2", + }, + ), + LlmStreamEvent("finish", finish_reason="stop"), + ] + ) + high_risk_runtime = FullDuplexAgentRuntime( + config=config, + memory_manager=memory, + llm_provider=high_risk_llm, + tts_provider=FakeStreamingTtsProvider(), + ) + high_risk_runtime.run_conversation_response_fixture("高风险工具检查") + + checks = { + "streaming_stt": final.kind == "final" and final.text == "你好" and [event.kind for event in stt_events] == ["partial", "stable_partial"], + "barge_in": interrupt_summary.interrupted + and interrupt_runtime.cancellation_graph.root.cancelled + and interrupt_runtime.state_machine.current_state == PipelineState.LISTENING, + "llm_tts_playback": spoken == "全双工自测通过。" + and response_runtime.audio_hub is not None + and response_runtime.audio_hub.render_reference.frame_count > 0, + "memory": bool(memory.search("Owner Agent", top_k=1)) and "长期记忆" in llm.requests[0][1].content, + "tool_router": bool(response_runtime.tool_results) + and response_runtime.tool_results[0].status == "success" + and "全双工 Agent" in response_runtime.tool_results[0].output_text, + "high_risk_tool_confirmation": bool(high_risk_runtime.tool_results) + and high_risk_runtime.tool_results[0].status == "confirmation_required", + } + return { + "success": all(checks.values()) and turns >= 1, + "profile": profile, + "turns_requested": turns, + "turns_exercised": max(3, turns), + "checks": checks, + } + + +def run_audio_self_test( + *, + config: AppConfig, + duration_s: int = 10, + check_echo: bool = False, +) -> dict[str, object]: + provider_error: ProviderError | None = None + provider_health = None + echo_suppressed = False + latency_ms = None + try: + provider = build_audio_processing_provider(config) + provider_health = provider.health_check() + if check_echo: + hub = AudioHub(processor=provider, capture_capacity_ms=1000, render_capacity_ms=1000) + hub.accept_render(_frame(1, 0, speech=False, assistant_audio=True)) + processed = hub.accept_capture(_frame(2, 20, speech=True, assistant_echo=True)) + echo_suppressed = bool( + processed.metadata.get("echo_suppressed") + or not processed.metadata.get("speech", True) + or set(processed.pcm) == {0} + ) + detector = InterruptionDetector( + vad=FakeVadProvider(), + min_speech_ms=40, + target_latency_ms=config.interrupt_target_latency_ms, + ) + detector.accept(_frame(3, 100, speech=True), state=PipelineState.SPEAKING) + decision = detector.accept(_frame(4, 120, speech=True), state=PipelineState.SPEAKING) + latency_ms = decision.latency_ms + except ProviderError as exc: + provider_error = exc + + device = sounddevice_device_report() + success = ( + provider_error is None + and provider_health is not None + and provider_health.available + and not provider_health.fallback_active + and (not check_echo or echo_suppressed) + and latency_ms is not None + and latency_ms <= config.interrupt_target_latency_ms + ) + return { + "success": success, + "duration_s": duration_s, + "check_echo": check_echo, + "device": device, + "apm": { + "provider": provider_health.provider if provider_health else config.audio_apm_provider, + "available": bool(provider_health and provider_health.available), + "fallback_active": bool(provider_health and provider_health.fallback_active), + "message": provider_health.message if provider_health else "", + "error_code": provider_error.code.value if provider_error else "", + "error_message": provider_error.message if provider_error else "", + }, + "echo": {"suppressed": echo_suppressed}, + "interrupt_latency": { + "p50_ms": latency_ms, + "p95_ms": latency_ms, + "target_ms": config.interrupt_target_latency_ms, + }, + } + + +def _frame( + frame_id: int, + timestamp_ms: int, + *, + speech: bool, + partial: str | None = None, + assistant_echo: bool = False, + assistant_audio: bool = False, +) -> AudioFrame: + metadata: dict[str, object] = {"duration_ms": 100 if timestamp_ms >= 100 else 20, "speech": speech} + if partial: + metadata["partial"] = partial + if assistant_echo: + metadata["assistant_echo"] = True + if assistant_audio: + metadata["assistant_audio"] = True + return AudioFrame( + pcm=b"\x01\x00" * 800, + sample_rate=16000, + channels=1, + timestamp_ms=timestamp_ms, + frame_id=frame_id, + metadata=metadata, + ) diff --git a/tests/test_cli_acceptance.py b/tests/test_cli_acceptance.py index fc184c7..0fa2c0c 100644 --- a/tests/test_cli_acceptance.py +++ b/tests/test_cli_acceptance.py @@ -207,6 +207,50 @@ class CliAcceptanceTests(unittest.TestCase): self.assertFalse(kwargs["play_audio"]) self.assertEqual(kwargs["questions"], ["第一问"]) + def test_agent_self_test_runs_full_duplex_profile(self) -> None: + code, data = self.call("agent-self-test", "--profile", "full-duplex", "--turns", "3") + + self.assertEqual(code, 0) + self.assertTrue(data["success"]) + self.assertTrue(data["checks"]["barge_in"]) + self.assertTrue(data["checks"]["tool_router"]) + + def test_audio_self_test_reports_missing_required_apm(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + code, data = self.call( + "--env-file", + str(Path(tmp) / ".env"), + "audio-self-test", + "--duration", + "1", + "--check-echo", + ) + + self.assertEqual(code, 1) + self.assertFalse(data["success"]) + self.assertEqual(data["apm"]["error_code"], "AUDIO_APM_UNAVAILABLE") + + def test_audio_self_test_can_use_fake_apm_for_deterministic_check(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + env_file = Path(tmp) / ".env" + env_file.write_text( + "OWNER_AUDIO_APM_PROVIDER=fake\nOWNER_AUDIO_APM_REQUIRED=0\n", + encoding="utf-8", + ) + code, data = self.call( + "--env-file", + str(env_file), + "audio-self-test", + "--duration", + "1", + "--check-echo", + ) + + self.assertEqual(code, 0) + self.assertTrue(data["success"]) + self.assertTrue(data["echo"]["suppressed"]) + self.assertEqual(data["apm"]["provider"], "fake_webrtc") + if __name__ == "__main__": unittest.main()