diff --git a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md index 4fc9f46..82c55ff 100644 --- a/openspec/changes/complete-full-duplex-agent-runtime/tasks.md +++ b/openspec/changes/complete-full-duplex-agent-runtime/tasks.md @@ -6,11 +6,11 @@ ## 2. AudioHub 与 WebRTC APM -- [ ] 2.1 实现 `AudioHub` 和独立订阅；前置条件：Phase 1；优先级：P0；验收标准：多消费者不抢帧；测试要点：两个订阅读取相同 frame 序列。 -- [ ] 2.2 实现 capture/render ring buffer 溢出诊断；前置条件：2.1；优先级：P0；验收标准：容量上限、丢旧帧、事件记录；测试要点：overflow fixture。 -- [ ] 2.3 接入 APM provider 到 AudioHub；前置条件：2.1；优先级：P0；验收标准：raw capture 经过 APM 后进入 processed ring；测试要点：fake APM echo suppression。 -- [ ] 2.4 `run-agent-live` 启动时检查真实 APM；前置条件：2.3；优先级：P0；验收标准：required 且不可用时结构化失败；测试要点：provider unavailable test。 -- [ ] 2.5 Phase 2 提交；前置条件：2.1-2.4；优先级：P0；验收标准：中文提交 `[音频底座重构]...`；测试要点：compileall、相关单测。 +- [x] 2.1 实现 `AudioHub` 和独立订阅；前置条件：Phase 1；优先级：P0；验收标准：多消费者不抢帧；测试要点：两个订阅读取相同 frame 序列。 +- [x] 2.2 实现 capture/render ring buffer 溢出诊断；前置条件：2.1；优先级：P0；验收标准：容量上限、丢旧帧、事件记录；测试要点：overflow fixture。 +- [x] 2.3 接入 APM provider 到 AudioHub；前置条件：2.1；优先级：P0；验收标准：raw capture 经过 APM 后进入 processed ring；测试要点：fake APM echo suppression。 +- [x] 2.4 `run-agent-live` 启动时检查真实 APM；前置条件：2.3；优先级：P0；验收标准：required 且不可用时结构化失败；测试要点：provider unavailable test。 +- [x] 2.5 Phase 2 提交；前置条件：2.1-2.4；优先级：P0；验收标准：中文提交 `[音频底座重构]...`；测试要点：compileall、相关单测。 ## 3. Continuous VAD/STT 与打断 diff --git a/src/owner_voice_pet/__init__.py b/src/owner_voice_pet/__init__.py index bc8fd90..d184377 100644 --- a/src/owner_voice_pet/__init__.py +++ b/src/owner_voice_pet/__init__.py @@ -29,6 +29,16 @@ from .full_duplex_control import ( RecoveryCoordinator, StateTransition, ) +from .full_duplex_audio import ( + AudioHub, + AudioHubDiagnostic, + AudioProcessingHealth, + FakeWebRtcAudioProcessingProvider, + NoopAudioProcessingProvider, + build_audio_processing_provider, + webrtc_apm_probe, +) +from .full_duplex_runtime import FullDuplexAgentRuntime, FullDuplexRuntimeHealth from .full_duplex_speech import ( FakeStreamingSttProvider, FakeVadProvider, @@ -116,6 +126,15 @@ __all__ = [ "InvalidStateTransition", "RecoveryCoordinator", "StateTransition", + "AudioHub", + "AudioHubDiagnostic", + "AudioProcessingHealth", + "FakeWebRtcAudioProcessingProvider", + "NoopAudioProcessingProvider", + "build_audio_processing_provider", + "webrtc_apm_probe", + "FullDuplexAgentRuntime", + "FullDuplexRuntimeHealth", "FakeStreamingSttProvider", "FakeVadProvider", "InterruptionDecision", diff --git a/src/owner_voice_pet/cli.py b/src/owner_voice_pet/cli.py index 59720f2..2f7adf0 100644 --- a/src/owner_voice_pet/cli.py +++ b/src/owner_voice_pet/cli.py @@ -11,6 +11,8 @@ from .audio_preprocess import SherpaOnnxDenoiserPreprocessor from .assets import validate_pet_assets from .config import AppConfig from .conversation import ConversationContext +from .full_duplex_audio import build_audio_processing_provider +from .full_duplex_runtime import FullDuplexAgentRuntime from .llm import MockLlmProvider, OpenAICompatibleLlmProvider from .models import AudioFrame, ProviderError from .pipeline import VoicePipeline @@ -217,11 +219,31 @@ def main(argv: list[str] | None = None) -> int: ) ) return 1 + apm_error: ProviderError | None = None + apm_health = None + try: + apm_health = build_audio_processing_provider(config).health_check() + except ProviderError as exc: + apm_error = exc + full_duplex_runtime_ready = ( + apm_error is None + and apm_health is not None + and apm_health.available + and ( + config.audio_apm_provider == "fake" + or (config.audio_apm_provider == "webrtc" and not apm_health.fallback_active) + ) + ) data = { "success": bool(args.check_config), "command": "run-agent-live", "assistant_mode": config.assistant_mode, "audio_apm_provider": config.audio_apm_provider, + "audio_apm_available": bool(apm_health and apm_health.available), + "audio_apm_fallback_active": bool(apm_health and apm_health.fallback_active), + "audio_apm_message": apm_health.message if apm_health else "", + "audio_apm_error_code": apm_error.code.value if apm_error else "", + "audio_apm_error_message": apm_error.message if apm_error else "", "streaming_stt_provider": config.streaming_stt_provider, "streaming_tts_provider": config.streaming_tts_provider, "llm_streaming_enabled": config.llm_stream, @@ -232,13 +254,13 @@ def main(argv: list[str] | None = None) -> int: "browser_playwright_enabled": config.browser_playwright_enabled, "computer_control_enabled": config.computer_control_enabled, "turn_based_entry": "run-live", - "full_duplex_runtime_ready": True, + "full_duplex_runtime_ready": full_duplex_runtime_ready, } if args.check_config: print(json.dumps(data, ensure_ascii=False, sort_keys=True)) return 0 try: - summary = build_live_runtime(config).run_agent(once=args.once) + summary = FullDuplexAgentRuntime(config=config).run(once=args.once) except ProviderError as exc: print(json.dumps({"ok": False, "code": exc.code.value, "message": exc.message}, ensure_ascii=False, sort_keys=True)) return 1 diff --git a/src/owner_voice_pet/full_duplex_audio.py b/src/owner_voice_pet/full_duplex_audio.py index 7a33ccf..6b02cbe 100644 --- a/src/owner_voice_pet/full_duplex_audio.py +++ b/src/owner_voice_pet/full_duplex_audio.py @@ -3,6 +3,7 @@ from __future__ import annotations import importlib.util from collections import deque from dataclasses import dataclass +from threading import RLock from typing import Protocol from .config import AppConfig @@ -16,6 +17,15 @@ class AudioBufferWriteResult: overrun: bool = False +@dataclass(frozen=True, slots=True) +class AudioHubDiagnostic: + code: ErrorCode + ring: str + message: str + dropped_frame_ids: tuple[int, ...] = () + subscriber: str | None = None + + class AudioRingBuffer: def __init__(self, *, capacity_ms: int, name: str) -> None: if capacity_ms <= 0: @@ -75,6 +85,146 @@ class RenderReferenceRingBuffer(AudioRingBuffer): super().__init__(capacity_ms=capacity_ms, name="render_reference") +class AudioSubscription: + def __init__( + self, + hub: "AudioHub", + *, + kind: str, + name: str, + last_seen_frame_id: int, + ) -> None: + self._hub = hub + self.kind = kind + self.name = name + self._last_seen_frame_id = last_seen_frame_id + self.missed_frames = 0 + + def read_available(self, *, max_frames: int | None = None) -> tuple[AudioFrame, ...]: + frames, missed = self._hub._read_for_subscription( + kind=self.kind, + subscriber=self.name, + last_seen_frame_id=self._last_seen_frame_id, + max_frames=max_frames, + ) + if missed: + self.missed_frames += missed + if frames: + self._last_seen_frame_id = frames[-1].frame_id + return frames + + +class AudioHub: + VALID_KINDS = {"raw_capture", "processed_capture", "render_reference"} + + def __init__( + self, + *, + processor: AudioProcessingProvider, + capture_capacity_ms: int = 3000, + render_capacity_ms: int = 3000, + ) -> None: + self.processor = processor + self.raw_capture = CaptureRingBuffer(capacity_ms=capture_capacity_ms) + self.processed_capture = CaptureRingBuffer(capacity_ms=capture_capacity_ms) + self.render_reference = RenderReferenceRingBuffer(capacity_ms=render_capacity_ms) + self._diagnostics: list[AudioHubDiagnostic] = [] + self._lock = RLock() + + @property + def diagnostics(self) -> tuple[AudioHubDiagnostic, ...]: + with self._lock: + return tuple(self._diagnostics) + + def subscribe(self, kind: str, *, name: str | None = None, replay_existing: bool = False) -> AudioSubscription: + if kind not in self.VALID_KINDS: + raise ValueError(f"unsupported audio hub subscription kind: {kind}") + with self._lock: + frames = self._buffer_for(kind).frames() + if replay_existing or not frames: + last_seen_frame_id = -1 + else: + last_seen_frame_id = frames[-1].frame_id + return AudioSubscription( + self, + kind=kind, + name=name or kind, + last_seen_frame_id=last_seen_frame_id, + ) + + def accept_capture(self, frame: AudioFrame) -> AudioFrame: + with self._lock: + self._write(self.raw_capture, frame) + processed = self.processor.process_capture(frame) + self._write(self.processed_capture, processed) + return processed + + def accept_render(self, frame: AudioFrame) -> None: + with self._lock: + self.processor.process_render(frame) + self._write(self.render_reference, frame) + + def clear(self) -> None: + with self._lock: + self.raw_capture.clear() + self.processed_capture.clear() + self.render_reference.clear() + self._diagnostics.clear() + self.processor.reset_stream() + + def _write(self, buffer: AudioRingBuffer, frame: AudioFrame) -> None: + result = buffer.write(frame) + if result.overrun: + dropped_ids = tuple(item.frame_id for item in result.dropped) + self._diagnostics.append( + AudioHubDiagnostic( + code=ErrorCode.AUDIO_BUFFER_OVERRUN, + ring=buffer.name, + message=f"{buffer.name} ring buffer dropped {len(dropped_ids)} old frame(s)", + dropped_frame_ids=dropped_ids, + ) + ) + + def _buffer_for(self, kind: str) -> AudioRingBuffer: + if kind == "raw_capture": + return self.raw_capture + if kind == "processed_capture": + return self.processed_capture + if kind == "render_reference": + return self.render_reference + raise ValueError(f"unsupported audio hub subscription kind: {kind}") + + def _read_for_subscription( + self, + *, + kind: str, + subscriber: str, + last_seen_frame_id: int, + max_frames: int | None, + ) -> tuple[tuple[AudioFrame, ...], int]: + with self._lock: + frames = self._buffer_for(kind).frames() + if not frames: + return (), 0 + missed = 0 + oldest_id = frames[0].frame_id + if last_seen_frame_id >= 0 and last_seen_frame_id + 1 < oldest_id: + missed = oldest_id - last_seen_frame_id - 1 + self._diagnostics.append( + AudioHubDiagnostic( + code=ErrorCode.AUDIO_BUFFER_OVERRUN, + ring=kind, + message=f"{subscriber} missed {missed} frame(s) from {kind}", + dropped_frame_ids=tuple(range(last_seen_frame_id + 1, oldest_id)), + subscriber=subscriber, + ) + ) + unread = tuple(frame for frame in frames if frame.frame_id > last_seen_frame_id) + if max_frames is not None: + unread = unread[: max(0, max_frames)] + return unread, missed + + @dataclass(frozen=True, slots=True) class AudioProcessingHealth: provider: str @@ -237,10 +387,16 @@ def build_audio_processing_provider(config: AppConfig) -> AudioProcessingProvide health = webrtc_apm_probe() if health.available: - return NoopAudioProcessingProvider( - fallback_active=True, - message="WebRTC APM binding is detected but native processing is not wired yet", - ) + message = "WebRTC APM binding is detected but native processing is not wired yet" + if config.audio_apm_required: + raise ProviderError( + ErrorCode.AUDIO_APM_UNAVAILABLE, + message, + False, + "webrtc", + "audio_apm", + ) + return NoopAudioProcessingProvider(fallback_active=True, message=message) if config.audio_apm_required: raise ProviderError( ErrorCode.AUDIO_APM_UNAVAILABLE, diff --git a/src/owner_voice_pet/full_duplex_runtime.py b/src/owner_voice_pet/full_duplex_runtime.py new file mode 100644 index 0000000..3d44a40 --- /dev/null +++ b/src/owner_voice_pet/full_duplex_runtime.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from .config import AppConfig +from .full_duplex_audio import AudioHub, AudioProcessingProvider, build_audio_processing_provider +from .models import AudioFrame +from .runtime import RuntimeSummary + + +@dataclass(slots=True) +class FullDuplexRuntimeHealth: + audio_apm_provider: str + audio_apm_available: bool + audio_apm_fallback_active: bool + message: str + + +class FullDuplexAgentRuntime: + """Full-duplex Agent runtime boundary. + + Phase 2 wires the audio foundation only: required APM startup, AudioHub fanout, + and render/capture processing. Later phases attach continuous VAD/STT, + cancellation, streaming response, memory, and tools to this runtime. + """ + + def __init__( + self, + *, + config: AppConfig, + processor: AudioProcessingProvider | None = None, + audio_hub: AudioHub | None = None, + ) -> None: + self.config = config + self.processor = processor + self.audio_hub = audio_hub + self.health: FullDuplexRuntimeHealth | None = None + + def load_audio(self) -> FullDuplexRuntimeHealth: + if self.audio_hub is None: + processor = self.processor or build_audio_processing_provider(self.config) + self.processor = processor + self.audio_hub = AudioHub( + processor=processor, + capture_capacity_ms=self.config.audio_ring_buffer_ms, + render_capacity_ms=self.config.audio_ring_buffer_ms, + ) + provider_health = self.audio_hub.processor.health_check() + self.health = FullDuplexRuntimeHealth( + audio_apm_provider=provider_health.provider, + audio_apm_available=provider_health.available, + audio_apm_fallback_active=provider_health.fallback_active, + message=provider_health.message, + ) + return self.health + + def run(self, *, once: bool = False) -> RuntimeSummary: + self.load_audio() + if once: + self._run_audio_smoke_once() + return RuntimeSummary(completed_turns=1, failed_turns=0) + self._run_audio_smoke_once() + return RuntimeSummary(completed_turns=1, failed_turns=0) + + def _run_audio_smoke_once(self) -> None: + if self.audio_hub is None: + raise RuntimeError("audio hub is not loaded") + render = AudioFrame( + pcm=b"\x01\x00" * 160, + sample_rate=self.config.sample_rate, + channels=self.config.channels, + timestamp_ms=0, + frame_id=1, + metadata={"duration_ms": self.config.audio_frame_ms, "assistant_audio": True}, + ) + capture = AudioFrame( + pcm=b"\x01\x00" * 160, + sample_rate=self.config.sample_rate, + channels=self.config.channels, + timestamp_ms=self.config.audio_frame_ms, + frame_id=2, + metadata={"duration_ms": self.config.audio_frame_ms, "assistant_echo": True, "speech": True}, + ) + self.audio_hub.accept_render(render) + self.audio_hub.accept_capture(capture) diff --git a/tests/test_cli_acceptance.py b/tests/test_cli_acceptance.py index a5619a9..e0ff7d4 100644 --- a/tests/test_cli_acceptance.py +++ b/tests/test_cli_acceptance.py @@ -78,25 +78,57 @@ class CliAcceptanceTests(unittest.TestCase): self.assertIn("run-agent-live", buffer.getvalue()) def test_run_agent_live_check_config_reports_reserved_entrypoint(self) -> None: - code, data = self.call("run-agent-live", "--check-config") + with tempfile.TemporaryDirectory() as tmp: + env_file = str(Path(tmp) / ".env") + code, data = self.call("--env-file", env_file, "run-agent-live", "--check-config") self.assertEqual(code, 0) self.assertTrue(data["success"]) self.assertEqual(data["command"], "run-agent-live") self.assertEqual(data["assistant_mode"], "full_duplex_agent") - self.assertTrue(data["full_duplex_runtime_ready"]) + self.assertFalse(data["full_duplex_runtime_ready"]) + self.assertEqual(data["audio_apm_provider"], "webrtc") + self.assertEqual(data["audio_apm_error_code"], "AUDIO_APM_UNAVAILABLE") self.assertEqual(data["turn_based_entry"], "run-live") def test_run_agent_live_once_invokes_agent_runtime(self) -> None: class FakeRuntime: - def run_agent(self, *, once: bool = False): + def __init__(self, *, config): + self.config = config + self.once = None + + def run(self, *, once: bool = False): self.once = once return type("Summary", (), {"completed_turns": 1, "interrupted": False})() - fake_runtime = FakeRuntime() - with patch("owner_voice_pet.cli.build_live_runtime", return_value=fake_runtime): - code = main(["run-agent-live", "--once"]) + created: list[FakeRuntime] = [] + + def make_runtime(*, config): + runtime = FakeRuntime(config=config) + created.append(runtime) + return runtime + + with ( + patch("owner_voice_pet.cli.build_audio_processing_provider"), + patch("owner_voice_pet.cli.FullDuplexAgentRuntime", side_effect=make_runtime), + ): + with tempfile.TemporaryDirectory() as tmp: + code = main(["--env-file", str(Path(tmp) / ".env"), "run-agent-live", "--once"]) self.assertEqual(code, 0) - self.assertTrue(fake_runtime.once) + self.assertEqual(len(created), 1) + self.assertTrue(created[0].once) + + def test_run_agent_live_fails_when_required_apm_is_unavailable(self) -> None: + with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: + probe.return_value = type( + "Health", + (), + {"available": False, "message": "missing binding"}, + )() + with tempfile.TemporaryDirectory() as tmp: + code, data = self.call("--env-file", str(Path(tmp) / ".env"), "run-agent-live", "--once") + + self.assertEqual(code, 1) + self.assertEqual(data["code"], "AUDIO_APM_UNAVAILABLE") def test_security_check_command_has_no_leaks(self) -> None: code, data = self.call("security-check") diff --git a/tests/test_full_duplex_audio.py b/tests/test_full_duplex_audio.py index f20f0be..b64d246 100644 --- a/tests/test_full_duplex_audio.py +++ b/tests/test_full_duplex_audio.py @@ -5,6 +5,7 @@ from unittest.mock import patch from owner_voice_pet.config import AppConfig from owner_voice_pet.full_duplex_audio import ( + AudioHub, CaptureRingBuffer, FakeWebRtcAudioProcessingProvider, NoopAudioProcessingProvider, @@ -81,6 +82,55 @@ class FullDuplexAudioTests(unittest.TestCase): self.assertTrue(processed.metadata["echo_suppressed"]) self.assertFalse(processed.metadata["speech"]) + def test_audio_hub_fans_out_processed_capture_without_stealing_frames(self) -> None: + hub = AudioHub( + processor=FakeWebRtcAudioProcessingProvider(sample_rate=16000, channels=1), + capture_capacity_ms=100, + ) + vad = hub.subscribe("processed_capture", name="vad") + stt = hub.subscribe("processed_capture", name="stt") + + for idx, timestamp in enumerate([0, 20, 40], start=1): + hub.accept_capture(frame(idx, timestamp, metadata={"speech": True})) + + self.assertEqual([item.frame_id for item in vad.read_available()], [1, 2, 3]) + self.assertEqual([item.frame_id for item in stt.read_available()], [1, 2, 3]) + self.assertEqual(vad.read_available(), ()) + self.assertEqual(stt.read_available(), ()) + + def test_audio_hub_processes_capture_before_processed_subscribers_read_it(self) -> None: + hub = AudioHub( + processor=FakeWebRtcAudioProcessingProvider(sample_rate=16000, channels=1), + capture_capacity_ms=100, + render_capacity_ms=100, + ) + hub.accept_render(frame(1, 0, metadata={"assistant_audio": True})) + subscription = hub.subscribe("processed_capture", name="interrupt") + + processed = hub.accept_capture(frame(2, 20, metadata={"assistant_echo": True, "speech": True})) + + self.assertTrue(processed.metadata["echo_suppressed"]) + self.assertEqual(subscription.read_available(), (processed,)) + self.assertEqual([item.frame_id for item in hub.render_reference.frames()], [1]) + + def test_audio_hub_reports_ring_and_subscriber_overrun(self) -> None: + hub = AudioHub( + processor=FakeWebRtcAudioProcessingProvider(sample_rate=16000, channels=1), + capture_capacity_ms=40, + ) + subscription = hub.subscribe("processed_capture", name="slow-stt") + hub.accept_capture(frame(1, 0)) + self.assertEqual([item.frame_id for item in subscription.read_available()], [1]) + + hub.accept_capture(frame(2, 20)) + hub.accept_capture(frame(3, 40)) + hub.accept_capture(frame(4, 60)) + + self.assertEqual([item.frame_id for item in subscription.read_available()], [3, 4]) + self.assertGreaterEqual(subscription.missed_frames, 1) + self.assertTrue(any(item.code == ErrorCode.AUDIO_BUFFER_OVERRUN for item in hub.diagnostics)) + self.assertTrue(any(item.subscriber == "slow-stt" for item in hub.diagnostics)) + def test_fake_webrtc_apm_rejects_format_mismatch(self) -> None: provider = FakeWebRtcAudioProcessingProvider(sample_rate=16000, channels=1) @@ -131,6 +181,20 @@ class FullDuplexAudioTests(unittest.TestCase): self.assertIsInstance(provider, NoopAudioProcessingProvider) self.assertTrue(provider.health_check().fallback_active) + def test_detected_but_unwired_webrtc_provider_fails_when_required(self) -> None: + with patch("owner_voice_pet.full_duplex_audio.webrtc_apm_probe") as probe: + probe.return_value = type( + "Health", + (), + {"available": True, "message": "found binding"}, + )() + with self.assertRaises(ProviderError) as raised: + build_audio_processing_provider( + AppConfig(audio_apm_provider="webrtc", audio_apm_required=True) + ) + + self.assertEqual(raised.exception.code, ErrorCode.AUDIO_APM_UNAVAILABLE) + def test_webrtc_probe_reports_unavailable_without_binding(self) -> None: with patch("importlib.util.find_spec", return_value=None): health = webrtc_apm_probe()