diff --git a/services/athena-realtime-voice/README.md b/services/athena-realtime-voice/README.md index b76992b..4f86c56 100644 --- a/services/athena-realtime-voice/README.md +++ b/services/athena-realtime-voice/README.md @@ -16,6 +16,10 @@ 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 speech item” in the browser. +Spoken assistant replies are also announced as conversation items with matching +transcript IDs and explicit predecessor IDs. Without them, OpenClaw's ordered +transcript storage can wait on a missing assistant item when the next user turn +arrives, leaving later questions unanswered. Required service environment: diff --git a/services/athena-realtime-voice/server.py b/services/athena-realtime-voice/server.py index 3f8830a..ae3080c 100644 --- a/services/athena-realtime-voice/server.py +++ b/services/athena-realtime-voice/server.py @@ -12,6 +12,7 @@ import hashlib import hmac import io import json +import logging import os import time import uuid @@ -21,6 +22,7 @@ from fractions import Fraction import numpy as np from aiohttp import ClientSession, FormData, web from aiortc import MediaStreamTrack, RTCPeerConnection, RTCSessionDescription +from aiortc.mediastreams import MediaStreamError from av import AudioFrame, AudioResampler from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.serialization import load_pem_public_key @@ -34,6 +36,12 @@ USED_TOKENS = web.AppKey("used_tokens", dict) PEERS = web.AppKey("peers", set) HTTP = web.AppKey("http", ClientSession) PUBLIC_KEY_URL = web.AppKey("public_key_url", str) +logger = logging.getLogger("athena_realtime_voice") +logger.setLevel(logging.INFO) +logger.propagate = False +_handler = logging.StreamHandler() +_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s")) +logger.addHandler(_handler) def decode_payload(payload: str, timestamp: int) -> dict: @@ -170,8 +178,10 @@ class RealtimeSession: self.speech = bytearray() self.silence_since: float | None = None self.call_id: str | None = None + self.last_item_id: str | None = None self.pending_reply: str | None = None self.turn_task: asyncio.Task | None = None + self.dropped_speech_while_busy = False self.vad_threshold = float(os.environ.get("VAD_THRESHOLD", "0.018")) self.silence_ms = int(os.environ.get("SILENCE_MS", "750")) @@ -187,6 +197,11 @@ class RealtimeSession: except json.JSONDecodeError: return kind = event.get("type") + if kind in {"conversation.item.create", "response.create", "response.cancel"}: + logger.info("client event=%s item_type=%s call_matches=%s busy=%s pending_reply=%s", + kind, (event.get("item") or {}).get("type"), + (event.get("item") or {}).get("call_id") == self.call_id, + self.busy, self.pending_reply is not None) if kind == "session.update": self.emit({"type": "session.updated", "session": {"id": "athena-local"}}) elif kind == "conversation.item.create": @@ -218,12 +233,21 @@ class RealtimeSession: await self.on_audio(pcm) except asyncio.CancelledError: raise + except MediaStreamError: + return except Exception: + logger.exception("incoming audio stream stopped unexpectedly") return async def on_audio(self, pcm: bytes) -> None: - if self.busy or not pcm: + if not pcm: return + if self.busy: + if not self.dropped_speech_while_busy and pcm_rms(pcm) >= self.vad_threshold: + self.dropped_speech_while_busy = True + logger.info("microphone speech ignored while response is busy") + return + self.dropped_speech_while_busy = False voiced = pcm_rms(pcm) >= self.vad_threshold now = time.monotonic() if voiced and not self.speaking: @@ -231,6 +255,7 @@ class RealtimeSession: self.speech.clear() self.silence_since = None self.emit({"type": "input_audio_buffer.speech_started"}) + logger.info("speech started") if not self.speaking: return self.speech.extend(pcm) @@ -242,6 +267,7 @@ class RealtimeSession: if elapsed_silence >= self.silence_ms or len(self.speech) >= SAMPLE_RATE * 2 * 45: self.speaking = False self.emit({"type": "input_audio_buffer.speech_stopped"}) + logger.info("speech stopped bytes=%s", len(self.speech)) audio = bytes(self.speech) self.speech.clear() if len(audio) >= SAMPLE_RATE * 2 // 4: @@ -265,14 +291,18 @@ class RealtimeSession: return item_id = f"item_{uuid.uuid4().hex}" # OpenClaw tracks transcript items before accepting their text. - self.emit({"type": "input_audio_buffer.committed", "item_id": item_id}) + self.emit({"type": "input_audio_buffer.committed", "item_id": item_id, + "previous_item_id": self.last_item_id}) + self.last_item_id = item_id self.emit({"type": "conversation.item.input_audio_transcription.completed", "item_id": item_id, "transcript": text}) + logger.info("transcript completed item_id=%s", item_id) await self.consult(text) except asyncio.CancelledError: self.busy = False raise except Exception: + logger.exception("speech transcription failed") self.busy = False self.emit({"type": "error", "error": {"message": "Athena STT failed"}}) @@ -289,13 +319,16 @@ class RealtimeSession: "arguments": json.dumps({"prompt": text}, ensure_ascii=False), }], }}) + logger.info("consult requested call_id=%s response_id=%s", self.call_id, response_id) async def speak(self, text: str) -> None: if not text.strip(): self.busy = False return response_id = f"resp_{uuid.uuid4().hex}" + item_id = f"item_{uuid.uuid4().hex}" try: + logger.info("speaking response_id=%s", response_id) self.emit({"type": "response.created", "response": {"id": response_id}}) async with self.http.post( os.environ["ATHENA_API_BASE_URL"].rstrip("/") + "/audio/speech", @@ -306,18 +339,25 @@ class RealtimeSession: response.raise_for_status() pcm = read_wav(await response.read()) self.track.enqueue(pcm) - self.emit({"type": "response.output_audio_transcript.done", "transcript": text}) + item = {"id": item_id, "type": "message", "role": "assistant", + "status": "completed"} + self.emit({"type": "conversation.item.created", "item": item, + "previous_item_id": self.last_item_id}) + self.last_item_id = item_id + self.emit({"type": "response.output_audio_transcript.done", + "item_id": item_id, "transcript": text}) # Keep half-duplex active until the queued audio has actually played. await asyncio.sleep(len(pcm) / (SAMPLE_RATE * 2) + 0.2) + self.emit({"type": "conversation.item.done", "item": item}) self.emit({"type": "response.done", "response": { - "id": response_id, "status": "completed", "output": [ - {"id": f"item_{uuid.uuid4().hex}", "type": "message", "role": "assistant", "status": "completed"} - ], + "id": response_id, "status": "completed", "output": [item], }}) + logger.info("speech response complete response_id=%s", response_id) except asyncio.CancelledError: self.emit({"type": "response.cancelled", "response": {"id": response_id}}) raise except Exception: + logger.exception("speech synthesis failed") self.emit({"type": "error", "error": {"message": "Athena TTS failed"}}) finally: self.busy = False diff --git a/services/athena-realtime-voice/test_smoke.py b/services/athena-realtime-voice/test_smoke.py index 9737260..3d6c271 100644 --- a/services/athena-realtime-voice/test_smoke.py +++ b/services/athena-realtime-voice/test_smoke.py @@ -31,8 +31,8 @@ class Microphone(MediaStreamTrack): async def recv(self): await asyncio.sleep(max(0, self.start + self.count * .02 - time.monotonic())) - # 0.6 s speech followed by silence to trigger server VAD. - if 4 <= self.count < 44: + # Two utterances separated by enough time for the first TTS reply. + if 4 <= self.count < 44 or 750 <= self.count < 790: index = np.arange(FRAME_SAMPLES) + self.count * FRAME_SAMPLES samples = (8000 * np.sin(2 * np.pi * 440 * index / SAMPLE_RATE)).astype(" 100: audio_received.set() @@ -154,11 +158,14 @@ class VoiceSmokeTest(unittest.IsolatedAsyncioTestCase): @channel.on("message") def on_message(raw): + nonlocal completed_responses, function_calls event = json.loads(raw) events.append(event) + event_times.append((round(time.monotonic() - microphone.start, 1), event.get("type"))) if event.get("type") == "response.done": output = event.get("response", {}).get("output", []) if output and output[0].get("type") == "function_call": + function_calls += 1 call = output[0] channel.send(json.dumps({"type": "conversation.item.create", "item": { "type": "function_call_output", "call_id": call["call_id"], @@ -166,7 +173,9 @@ class VoiceSmokeTest(unittest.IsolatedAsyncioTestCase): }})) channel.send(json.dumps({"type": "response.create"})) elif output and output[0].get("type") == "message": - done.set() + completed_responses += 1 + if completed_responses == 2: + done.set() try: await pc.setLocalDescription(await pc.createOffer()) @@ -186,13 +195,29 @@ class VoiceSmokeTest(unittest.IsolatedAsyncioTestCase): answer = await response.text() await pc.setRemoteDescription(RTCSessionDescription(sdp=answer, type="answer")) try: - await asyncio.wait_for(done.wait(), timeout=20) + await asyncio.wait_for(done.wait(), timeout=25) except asyncio.TimeoutError: - self.fail(f"No completed voice response; events: {[e.get('type') for e in events]}") + self.fail(f"No completed voice response; state={pc.connectionState}, " + f"mic_frames={microphone.count}, events: {event_times}") await asyncio.wait_for(audio_received.wait(), timeout=5) + self.assertEqual(function_calls, 2) self.assertTrue(any(event.get("type") == "conversation.item.input_audio_transcription.completed" for event in events)) + user_items = [event for event in events + if event.get("type") == "input_audio_buffer.committed"] + assistant_items = [event for event in events + if event.get("type") == "conversation.item.created" + and event.get("item", {}).get("role") == "assistant"] + assistant_transcripts = [event for event in events + if event.get("type") == "response.output_audio_transcript.done"] + self.assertEqual(len(user_items), 2) + self.assertEqual(len(assistant_items), 2) + self.assertEqual(len(assistant_transcripts), 2) + self.assertEqual(assistant_items[0]["previous_item_id"], user_items[0]["item_id"]) + self.assertEqual(assistant_transcripts[0]["item_id"], assistant_items[0]["item"]["id"]) + self.assertEqual(user_items[1]["previous_item_id"], assistant_items[0]["item"]["id"]) + self.assertEqual(assistant_items[1]["previous_item_id"], user_items[1]["item_id"]) committed = next((index, event) for index, event in enumerate(events) if event.get("type") == "input_audio_buffer.committed") completed = next((index, event) for index, event in enumerate(events)