"""Bounded, memory-only push-to-talk. Identity is pinned before capture starts.""" from __future__ import annotations import asyncio import base64 import binascii import io import logging import os import re import time import wave from dataclasses import dataclass, field from typing import Protocol import httpx from .control.models import ControlError from .state import split_target SAMPLE_RATE = 44100 MAX_SECONDS = 20 MAX_BYTES = SAMPLE_RATE * 2 * MAX_SECONDS CHUNK_BYTES = 4410 # at most 50 ms, well below an 8192-byte JSON frame MAX_CHUNKS = 1000 CF_STT_MODEL = '@cf/openai/whisper-large-v3-turbo' # Inherit the configured Uvicorn handler/INFO level used by run-server.sh. # The ordinary module logger otherwise inherits the root WARNING level. logger = logging.getLogger('uvicorn.error.device_gateway.voice') def short_id(value): """Bound correlation IDs in lifecycle logs; never log voice contents.""" return value[:16] if isinstance(value, str) else 'unknown' class SpeechToTextProvider(Protocol): async def transcribe(self, wav_bytes: bytes) -> str: ... class TranscriptionError(RuntimeError): def __init__(self): super().__init__('Transcription failed') class _HttpSTTProvider: # Diagnostics for the explicit, sequential smoke script; never contain audio/text. last_request_seconds: float | None = None async def _post(self, url, *, unwrap_result=False, **kwargs): started, status = time.monotonic(), None outcome = 'incomplete' self.last_request_seconds = None try: # A total deadline in addition to HTTPX's per-operation timeout. # No retries and no redirects: one voice goes to one configured provider. async with asyncio.timeout(25): async with httpx.AsyncClient(timeout=25, follow_redirects=False) as client: response = await client.post(url, headers={'Authorization': 'Bearer ' + self.api_key}, **kwargs) status = response.status_code response.raise_for_status() payload = response.json() if unwrap_result: if payload.get('success') is not True or payload.get('errors'): raise ValueError() payload = payload['result'] text = payload['text'] if not isinstance(text, str): raise ValueError() text = ''.join(c for c in text if c in '\n\t' or ord(c) >= 32).strip() if not text or len(text.encode('utf-8')) > 4000: raise ValueError() outcome = 'success' return text except httpx.HTTPStatusError: outcome = 'http_status' raise TranscriptionError() from None except (httpx.TimeoutException, TimeoutError): outcome = 'timeout' raise TranscriptionError() from None except httpx.HTTPError: outcome = 'network' raise TranscriptionError() from None except (ValueError, KeyError, TypeError, AttributeError): # Never expose a response body, Authorization header or exception URL. outcome = 'invalid_response' raise TranscriptionError() from None except asyncio.CancelledError: outcome = 'cancelled' raise finally: self.last_request_seconds = time.monotonic() - started logger.info('STT provider=%s status=%s outcome=%s request_seconds=%.3f', self.provider_name, status, outcome, self.last_request_seconds) class OpenAICompatibleSTT(_HttpSTTProvider): provider_name = 'openai-compatible' def __init__(self, base_url, api_key, model, initial_prompt=None): self.base_url, self.api_key, self.model = base_url.rstrip('/'), api_key, model self.initial_prompt = initial_prompt async def transcribe(self, wav_bytes): data = {'model': self.model} if self.initial_prompt: data['prompt'] = self.initial_prompt return await self._post(self.base_url + '/audio/transcriptions', data=data, files={'file': ('voice.wav', wav_bytes, 'audio/wav')}) class CloudflareWorkersAISttProvider(_HttpSTTProvider): provider_name = 'cloudflare-workers-ai' def __init__(self, account_id, api_key, model=CF_STT_MODEL, initial_prompt=None): self.account_id, self.api_key, self.model = account_id, api_key, model self.initial_prompt = initial_prompt async def transcribe(self, wav_bytes): # Official model schema: base64 of the complete, unchanged WAV container. payload = {'audio': base64.b64encode(wav_bytes).decode('ascii'), 'task': 'transcribe'} if self.initial_prompt: payload['initial_prompt'] = self.initial_prompt # Deliberately omit language: Chinese/English code-switching is expected. return await self._post( f'https://api.cloudflare.com/client/v4/accounts/{self.account_id}/ai/run/{self.model}', unwrap_result=True, json=payload) def configured_stt(): provider = os.environ.get('GATEWAY_STT_PROVIDER') prompt = os.environ.get('GATEWAY_STT_INITIAL_PROMPT') or None if provider == 'cloudflare-workers-ai': account = os.environ.get('GATEWAY_CF_ACCOUNT_ID') token = os.environ.get('GATEWAY_CF_API_TOKEN') model = os.environ.get('GATEWAY_STT_MODEL') or CF_STT_MODEL return CloudflareWorkersAISttProvider(account, token, model, prompt) if account and token else None if provider == 'openai-compatible': values = [os.environ.get('GATEWAY_STT_' + name) for name in ('BASE_URL', 'API_KEY', 'MODEL')] # Reuse an existing Groq key only for an explicitly selected Groq endpoint. if values[0] and values[0].rstrip('/') == 'https://api.groq.com/openai/v1' and not values[1]: values[1] = os.environ.get('GROQ_API_KEY') return OpenAICompatibleSTT(*values, initial_prompt=prompt) if all(values) else None return None def wav_pcm(pcm): output = io.BytesIO() with wave.open(output, 'wb') as stream: stream.setnchannels(1) stream.setsampwidth(2) stream.setframerate(SAMPLE_RATE) stream.writeframes(pcm) return output.getvalue() @dataclass class VoiceRequest: voice_id: str connection: str target: str agent: str session_id: str name: str generation: int = 0 context: dict | None = None state: str = 'RECORDING' pcm: bytearray = field(default_factory=bytearray, repr=False) sequence: int = 0 transcript: str = field(default='', repr=False) task: asyncio.Task | None = field(default=None, repr=False) timer: asyncio.TimerHandle | None = field(default=None, repr=False) class VoiceManager: def __init__(self, provider, target_for, adapter_for, send, is_current, approval_pending, generation_for=lambda did:0, context_prepare=None, context_valid=lambda device,context:True, context_submit=None): self.provider, self.target_for, self.adapter_for = provider, target_for, adapter_for self.send, self.is_current, self.approval_pending = send, is_current, approval_pending self.generation_for=generation_for;self.context_prepare=context_prepare;self.context_valid=context_valid;self.context_submit=context_submit self.requests = {} self.seen = {} self.tasks = set() def spawn(self, coro): task = asyncio.create_task(coro) self.tasks.add(task) task.add_done_callback(self.tasks.discard) return task def valid(self, device, request): return (self.requests.get(device.connection_id) is request and self.is_current(device) and self.target_for(device.device_id) == request.target and self.generation_for(device.device_id)==request.generation and self.context_valid(device,request.context) and not self.approval_pending()) async def emit(self, device, request, kind='voice_state', **extra): if self.is_current(device): await self.send(device, dict(type=kind, voice_id=request.voice_id, state=request.state, agent=request.agent, session_id=request.session_id, name=request.name, **extra)) def cancel(self, device, *, notify=True, forget=False): request = self.requests.pop(device.connection_id, None) if request: if request.timer: request.timer.cancel() if request.task and request.task is not asyncio.current_task(): request.task.cancel() request.state = 'CANCELLED' request.pcm.clear() request.transcript = '' if notify: self.spawn(self.emit_safely(device, request)) if forget: self.seen.pop(device.connection_id, None) async def emit_safely(self, device, request): try: await self.emit(device, request) except Exception: pass async def handle(self, device, event): kind, vid = event.get('type'), event.get('voice_id') if not isinstance(vid, str) or not re.fullmatch(r'[A-Za-z0-9_-]{1,64}', vid): return if not self.is_current(device): return if kind == 'voice_start': logger.info('[VOICE_RX] event=start voice_id=%s connection=%s', short_id(vid), short_id(device.connection_id)) used = self.seen.setdefault(device.connection_id, set()) if vid in used or len(used) >= 256: return used.add(vid) self.cancel(device) target = self.target_for(device.device_id) generation=self.generation_for(device.device_id) prepared=None if self.context_prepare: try:prepared=await self.context_prepare(device,event) except ControlError as exc: logger.info('[VOICE_RX] event=reject voice_id=%s status=%d', short_id(vid), exc.status) await self.send(device,dict(type='voice_error',voice_id=vid,message=exc.detail));return parsed = (prepared['agent'],prepared['session_id']) if prepared else split_target(target) if not parsed: await self.send(device, {'type':'voice_error','voice_id':vid,'message':'NO AGENT SELECTED'}) return if self.approval_pending(): return agent, sid = parsed if any(event.get(k) != v for k, v in [('sample_rate',44100),('bits',16),('channels',1)]): await self.send(device, {'type':'voice_error','voice_id':vid,'message':'Unsupported audio format'}) return # Install identity before the async catalog read, so target switches, # disconnects and approvals can invalidate even an in-flight start. req = VoiceRequest(vid,device.connection_id,target,agent,sid,sid,generation=generation,context=prepared["context"] if prepared else None) self.requests[device.connection_id] = req try: session = None if prepared else await self.adapter_for(agent).get_session(sid) if not self.valid(device, req): self.cancel(device) return req.name = prepared["name"] if prepared else session.name req.timer = asyncio.get_running_loop().call_later(MAX_SECONDS, lambda: self.spawn(self.end(device, req))) await self.emit(device, req) except Exception: await self.fail(device, req, 'Agent unavailable') return req = self.requests.get(device.connection_id) if not req or req.voice_id != vid: if kind=='voice_cancel': used=self.seen.setdefault(device.connection_id,set()) if len(used)<256:used.add(vid) return if not self.valid(device, req): self.cancel(device) return if kind == 'voice_cancel' or (kind == 'voice_decision' and event.get('decision') == 'cancel'): self.cancel(device) elif kind == 'voice_chunk' and req.state == 'RECORDING': try: audio = event.get('audio') if not isinstance(audio, str) or len(audio) > 5880: raise ValueError() pcm = base64.b64decode(audio, validate=True) if (type(event.get('sequence')) is not int or event['sequence'] != req.sequence or not pcm or len(pcm) % 2 or len(pcm) > CHUNK_BYTES or len(req.pcm) + len(pcm) > MAX_BYTES or req.sequence >= MAX_CHUNKS): raise ValueError() req.pcm.extend(pcm) req.sequence += 1 except (ValueError, binascii.Error): logger.info('[VOICE_RX] event=cancel voice_id=%s reason=invalid_chunk', short_id(vid)) self.cancel(device) # never drop a chunk silently elif kind == 'voice_end': logger.info('[VOICE_RX] event=end voice_id=%s connection=%s chunks=%d pcm_bytes=%d', short_id(vid), short_id(device.connection_id), req.sequence, len(req.pcm)) duration=event.get('recorded_duration_ms') if duration is not None and (type(duration) is not int or duration<=1000): logger.info('[VOICE_RX] event=cancel voice_id=%s reason=short_or_invalid_duration recorded_duration_ms=%s', short_id(vid), duration if type(duration) is int else 'invalid') self.cancel(device) else:await self.end(device, req) elif kind == 'voice_decision' and event.get('decision') == 'send' and req.state == 'AWAITING_CONFIRMATION': # Consume the one-shot decision before awaiting adapter preflight. req.state = 'SENDING' transcript, req.transcript = req.transcript, '' context_kind = (req.context or {}).get('kind') logger.info('[VOICE_SEND] event=start voice_id=%s context=%s operation=%s', short_id(vid), context_kind if context_kind in ('message','user_input','new_session') else 'message', short_id((req.context or {}).get('operation_id'))) try: if req.context:await self.context_submit(device,req.context,transcript) else:await self.adapter_for(req.agent).send_input(req.session_id,transcript) req.state = 'SENT' logger.info('[VOICE_SEND] event=complete voice_id=%s success=1', short_id(vid)) await self.emit(device, req) except Exception: logger.info('[VOICE_SEND] event=complete voice_id=%s success=0', short_id(vid)) await self.fail(device, req, 'Input failed; check Agent status before retrying') async def end(self, device, req): if req.state != 'RECORDING' or not self.valid(device, req): return if req.timer: req.timer.cancel() req.state = 'TRANSCRIBING' await self.emit(device, req) req.task = self.spawn(self.transcribe(device, req)) async def transcribe(self, device, req): started = time.monotonic() success = False logger.info('[VOICE_STT] event=start voice_id=%s connection=%s chunks=%d pcm_bytes=%d', short_id(req.voice_id), short_id(req.connection), req.sequence, len(req.pcm)) try: if not self.provider or not req.pcm: raise ValueError() audio = wav_pcm(req.pcm) req.pcm.clear() text = await asyncio.wait_for(self.provider.transcribe(audio), 30) if isinstance(text, str): text = ''.join(c for c in text if c in '\n\t' or ord(c) >= 32).strip() if not isinstance(text, str) or not text or len(text.encode('utf-8')) > 4000: raise ValueError() if not self.valid(device, req): return req.transcript, req.state = text.strip(), 'AWAITING_CONFIRMATION' await self.emit(device, req, 'voice_transcript', transcript=req.transcript) success = True except asyncio.CancelledError: raise except Exception: if self.valid(device, req): await self.fail(device, req, 'Transcription failed') finally: req.pcm.clear() logger.info('[VOICE_STT] event=complete voice_id=%s connection=%s duration_ms=%.1f success=%d', short_id(req.voice_id), short_id(req.connection), (time.monotonic() - started) * 1000, success) async def fail(self, device, req, message): logger.info('[VOICE_ERROR] voice_id=%s phase=%s', short_id(req.voice_id), req.state) req.state, req.transcript = 'FAILED', '' req.pcm.clear() if req.timer: req.timer.cancel() await self.emit(device, req, 'voice_error', message=message) async def close(self): for req in self.requests.values(): if req.timer: req.timer.cancel() req.pcm.clear() req.transcript = '' self.requests.clear() self.seen.clear() for task in self.tasks: task.cancel() await asyncio.gather(*self.tasks, return_exceptions=True)