"""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' logger = logging.getLogger(__name__) 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 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() return text except (httpx.HTTPError, TimeoutError, ValueError, KeyError, TypeError, AttributeError): # Never expose a response body, Authorization header or exception URL. raise TranscriptionError() from None finally: self.last_request_seconds = time.monotonic() - started logger.info('STT provider=%s status=%s request_seconds=%.3f', self.provider_name, status, 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 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): 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.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 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) parsed = 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) self.requests[device.connection_id] = req try: session = await self.adapter_for(agent).get_session(sid) if not self.valid(device, req): self.cancel(device) return req.name = 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: 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): 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)) 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, '' try: await self.adapter_for(req.agent).send_input(req.session_id, transcript) req.state = 'SENT' await self.emit(device, req) except Exception: 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): 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)