diff --git a/services/athena-realtime-voice/README.md b/services/athena-realtime-voice/README.md index 4f86c56..e48e58b 100644 --- a/services/athena-realtime-voice/README.md +++ b/services/athena-realtime-voice/README.md @@ -12,6 +12,13 @@ WebRTC; Athena transcribes it; the service asks OpenClaw to run connection. This is a half-duplex prototype. Barge-in and remote-network TURN support are not implemented. +For replies, the bridge consumes Athena's existing +`/v1/audio/speech/pcm-stream` endpoint (24 kHz mono PCM) and enqueues audio +frames as they arrive. Playback can start before Qwen finishes synthesizing a +long reply. The bounded queue applies backpressure instead of truncating long +audio, and the next turn waits until all frames have played. The two-turn +WebRTC test verifies that audio begins before the fake PCM stream finishes. + The bridge announces each committed user audio item before sending its completed transcript with the same item ID. OpenClaw requires that sequence to persist the transcript; omitting the item caused “Realtime transcript refers to an unknown diff --git a/services/athena-realtime-voice/server.py b/services/athena-realtime-voice/server.py index ae3080c..df73cd6 100644 --- a/services/athena-realtime-voice/server.py +++ b/services/athena-realtime-voice/server.py @@ -137,27 +137,30 @@ class SpeechTrack(MediaStreamTrack): def __init__(self): super().__init__() self._queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=2000) - self._pending = bytearray() + self._frames_enqueued = 0 + self._frames_played = 0 self._pts = 0 self._start = time.monotonic() - def enqueue(self, pcm: bytes) -> None: - for offset in range(0, len(pcm), FRAME_SAMPLES * 2): - try: - self._queue.put_nowait(pcm[offset : offset + FRAME_SAMPLES * 2]) - except asyncio.QueueFull: - break + async def enqueue(self, pcm: bytes) -> int: + if len(pcm) != FRAME_SAMPLES * 2: + raise ValueError("SpeechTrack expects one complete audio frame") + await self._queue.put(pcm) + self._frames_enqueued += 1 + return self._frames_enqueued + + async def wait_played(self, target: int) -> None: + while self._frames_played < target: + await asyncio.sleep(0.02) + await asyncio.sleep(0.2) async def recv(self) -> AudioFrame: await asyncio.sleep(max(0, self._start + self._pts / SAMPLE_RATE - time.monotonic())) - if not self._pending: - try: - self._pending.extend(self._queue.get_nowait()) - except asyncio.QueueEmpty: - pass - chunk = bytes(self._pending[: FRAME_SAMPLES * 2]) - del self._pending[: FRAME_SAMPLES * 2] - chunk = chunk.ljust(FRAME_SAMPLES * 2, b"\x00") + try: + chunk = self._queue.get_nowait() + self._frames_played += 1 + except asyncio.QueueEmpty: + chunk = b"\x00" * FRAME_SAMPLES * 2 frame = AudioFrame.from_ndarray( np.frombuffer(chunk, dtype="= frame_bytes: + target = await self.track.enqueue(bytes(pending[:frame_bytes])) + del pending[:frame_bytes] + if len(pending) % 2: + raise ValueError("Athena TTS returned incomplete PCM samples") + if pending: + target = await self.track.enqueue(bytes(pending).ljust(frame_bytes, b"\0")) + if not target: + raise ValueError("Athena TTS returned no audio") + # Keep half-duplex active until the streamed audio has played. + await self.track.wait_played(target) self.emit({"type": "conversation.item.done", "item": item}) self.emit({"type": "response.done", "response": { "id": response_id, "status": "completed", "output": [item], diff --git a/services/athena-realtime-voice/test_smoke.py b/services/athena-realtime-voice/test_smoke.py index 3d6c271..3db3319 100644 --- a/services/athena-realtime-voice/test_smoke.py +++ b/services/athena-realtime-voice/test_smoke.py @@ -116,12 +116,19 @@ class VoiceSmokeTest(unittest.IsolatedAsyncioTestCase): async def tts(request): body = await request.json() self.assertEqual(body["input"], "Hallo zurück") - return web.Response(body=make_wav((6000).to_bytes(2, "little", signed=True) * SAMPLE_RATE), - content_type="audio/wav") + self.assertEqual(body["chunk_size"], 4) + stream = web.StreamResponse(headers={"Content-Type": "application/octet-stream"}) + await stream.prepare(request) + pcm = (6000).to_bytes(2, "little", signed=True) * SAMPLE_RATE + await stream.write(pcm[:SAMPLE_RATE]) + await asyncio.wait_for(audio_received.wait(), timeout=3) + await stream.write(pcm[SAMPLE_RATE:]) + await stream.write_eof() + return stream fake = web.Application() fake.router.add_post("/audio/transcriptions", stt) - fake.router.add_post("/audio/speech", tts) + fake.router.add_post("/audio/speech/pcm-stream", tts) fake_runner, fake_url = await serve(fake) os.environ.update({ "ATHENA_TALK_REALTIME_SECRET": "test-secret-" * 4,