"""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 # Recording length cap, independent of how long an interrupted upload may take. 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 # A resumed upload may outlive the 20 s recording; it is still bounded so a # request cannot be pinned open forever by a device that never finishes. MAX_UPLOAD_SECONDS = 600 # Reference framing: the device targets 20 ms (882 byte) PCM chunks after # decoupling capture from transport; the gateway still accepts up to CHUNK_BYTES. 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' device_id: str = '' 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) deadline: asyncio.TimerHandle | None = field(default=None, repr=False) ended: bool = False device: object | 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): # Identity is pinned to the device, not to a single WebSocket, so a # reconnect can resume the same recording from where it stopped. return (self.requests.get(request.device_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()) def current_for(self, device): """The in-flight request for this device, regardless of connection.""" return self.requests.get(device.device_id) 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.device_id, None) if request: for handle in (request.timer, request.deadline): if handle: handle.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.device_id, None) async def emit_safely(self, device, request): try: await self.emit(device, request) except Exception: pass def suspend(self, device, *, reason='transport'): """Keep an unfinished recording across a transport interruption. A weak link must not destroy audio that is already recorded, so the request, its PCM and its acknowledged sequence survive until the device reconnects and resumes, or the upload window expires. """ request = self.requests.get(device.device_id) if not request or request.connection != device.connection_id or request.state in ('TRANSCRIBING', 'AWAITING_CONFIRMATION', 'SENDING', 'SENT', 'FAILED'): # Past recording: there is no upload left to resume. return if request.state == 'SUSPENDED': return logger.info('[VOICE_RX] event=suspend voice_id=%s reason=%s chunks=%d pcm_bytes=%d', short_id(request.voice_id), reason, request.sequence, len(request.pcm)) request.state = 'SUSPENDED' if request.timer: request.timer.cancel() request.timer = None # The upload deadline keeps running so a device that never returns # cannot pin its recording in memory forever. 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.device_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,device_id=device.device_id,device=device) self.requests[device.device_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 # Recording length cap. The device owns capture timing, so the # server must never "finish" a recording on its own: doing so # would transcribe a truncated upload whenever the link is slow. # Reaching the cap suspends the request and preserves PCM until # the device sends a real voice_end or the window expires. req.timer = asyncio.get_running_loop().call_later(MAX_SECONDS, lambda: self.spawn(self.cap_reached(device, req))) # Upload/receive deadline is separate and much looser, so a # reconnect mid-upload is not mistaken for a full recording. req.deadline = asyncio.get_running_loop().call_later(MAX_UPLOAD_SECONDS, lambda: self.spawn(self.expire(device, req))) await self.emit(device, req) except Exception: await self.fail(device, req, 'Agent unavailable') return if kind == 'voice_status': # Reconnect resume: report the authoritative next expected sequence # and take the recording out of its suspended state. pending = self.current_for(device) if pending and pending.voice_id == vid: if not self.valid(device, pending): self.cancel(device) return pending.device = device pending.connection = device.connection_id if pending.state == 'SUSPENDED': pending.state = 'RECORDING' # Adopt the new link so the resumed upload is attributed to it. pending.connection = device.connection_id await self.ack(device, pending) if pending.ended and pending.state in ('TRANSCRIBING', 'AWAITING_CONFIRMATION'): await self.end_ack(device, pending) if pending.state == 'AWAITING_CONFIRMATION': await self.emit(device, pending, 'voice_transcript', transcript=pending.transcript) return if kind == 'voice_decision' and event.get('decision') == 'retry': # A client whose bounded upload window expired asks to resume; the # recording is still retained on the device. Re-arm the receive # deadline and report the breakpoint instead of restarting. pending = self.current_for(device) if pending and pending.voice_id == vid: if not self.valid(device, pending): self.cancel(device) return if pending.deadline: pending.deadline.cancel() pending.deadline = asyncio.get_running_loop().call_later(MAX_UPLOAD_SECONDS, lambda: self.spawn(self.expire(device, pending))) await self.ack(device, pending) return req = self.current_for(device) if not req or req.voice_id != vid: if kind=='voice_cancel': used=self.seen.setdefault(device.device_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) return if req.state == 'SUSPENDED' and req.connection != device.connection_id: # Resume is deliberate: only a voice_status query on the new # connection revives a recording started on a link that dropped. # Any other stale message from a replacement socket is ignored. return if kind == 'voice_cancel' or (kind == 'voice_decision' and event.get('decision') == 'cancel'): self.cancel(device) elif kind == 'voice_chunk' and req.state in ('RECORDING', 'SUSPENDED'): if req.state == 'SUSPENDED': req.state = 'RECORDING' req.connection = device.connection_id try: audio = event.get('audio') sequence = event.get('sequence') if not isinstance(audio, str) or len(audio) > 5880: raise ValueError() pcm = base64.b64decode(audio, validate=True) if (type(sequence) is not int or sequence < 0 or not pcm or len(pcm) % 2 or len(pcm) > CHUNK_BYTES or sequence >= MAX_CHUNKS): raise ValueError() if sequence < req.sequence: # Replay after a reconnect: acknowledge the authoritative # position and do not append the same audio twice. await self.ack(device, req) return if sequence != req.sequence: # A gap means audio was lost; resume instead of splicing a hole. raise ValueError() if len(req.pcm) + len(pcm) > MAX_BYTES: 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 return await self.ack(device, req) 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') total_bytes=event.get('total_bytes') if req.ended: # 重复 END 只重发终态确认和已有结果，绝不重新调用 STT。 if req.state in ('TRANSCRIBING', 'AWAITING_CONFIRMATION'): req.device = device req.connection = device.connection_id await self.end_ack(device, req) if req.state == 'AWAITING_CONFIRMATION': await self.emit(device, req, 'voice_transcript', transcript=req.transcript) return # Never transcribe a partial upload: the device must have sent every # byte it recorded, and the intact PCM must contain exactly those # bytes. A silent short-read would corrupt the transcript instead. if total_bytes is not None and (type(total_bytes) is not int or total_bytes != len(req.pcm)): logger.info('[VOICE_RX] event=cancel voice_id=%s reason=total_bytes_mismatch total_bytes=%s received=%d', short_id(vid), total_bytes if type(total_bytes) is int else 'invalid', len(req.pcm)) self.cancel(device) elif duration is not None and (type(duration) is not int or duration<=1000 or duration > MAX_SECONDS * 1000 + 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 ack(self, device, req): """Authoritative upload position for retry and reconnect resume.""" if self.is_current(device): await self.send(device, dict(type='voice_ack', voice_id=req.voice_id, next_sequence=req.sequence, sequence=req.sequence, state=req.state)) async def end_ack(self, device, req): # END 的确认属于请求；传输失败不能撤销已验证的 END 或阻断 STT。 try: if self.is_current(device): await self.send(device, dict(type='voice_end_ack', voice_id=req.voice_id, state=req.state)) except Exception: pass async def end(self, device, req): if req.state not in ('RECORDING', 'SUSPENDED') or not self.valid(device, req): return for handle in (req.timer, req.deadline): if handle: handle.cancel() req.ended = True req.state = 'TRANSCRIBING' try: await self.end_ack(device, req) await self.emit_safely(device, req) finally: # 即便确认尾包发送失败，也只启动一次已验证的转写。 if self.current_for(device) is req and req.state == 'TRANSCRIBING': req.task = self.spawn(self.transcribe(device, req)) async def cap_reached(self, device, req): """Recording length cap reached without an end from the device. The cap exists to bound memory, not to declare the recording finished. The device owns capture timing, so transcribing here would transcribe a truncated upload whenever the link is slow. Suspend instead: the PCM and the acknowledged sequence survive so a real voice_end can still arrive. """ if self.current_for(device) is not req or req.state != 'RECORDING': return logger.info('[VOICE_RX] event=cap voice_id=%s chunks=%d pcm_bytes=%d action=suspend', short_id(req.voice_id), req.sequence, len(req.pcm)) req.timer = None self.suspend(device, reason='recording-cap') async def expire(self, device, req): """Upload window elapsed without an end; release memory, keep it honest.""" if self.current_for(device) is not req or req.state not in ('RECORDING', 'SUSPENDED', 'TRANSCRIBING'): return logger.info('[VOICE_RX] event=expire voice_id=%s chunks=%d pcm_bytes=%d', short_id(req.voice_id), req.sequence, len(req.pcm)) self.cancel(device) 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() device = req.device or device # STT 结果属于请求而非传输；离线时保留，重连 status 再交付。 if (self.current_for(device) is not req or self.target_for(req.device_id) != req.target or self.generation_for(req.device_id) != req.generation # Navigator context validation also requires a live link. # Offline results stay with the request; voice_status on the # current connection validates context before redelivery. or (self.is_current(device) and not self.context_valid(device, req.context)) or self.approval_pending()): return req.transcript, req.state = text.strip(), 'AWAITING_CONFIRMATION' # 已完成的结果不能因尾包发送失败变为 FAILED；status 可再次交付。 success = True try: await self.emit(device, req, 'voice_transcript', transcript=req.transcript) except Exception: pass except asyncio.CancelledError: raise except Exception: device = req.device or device if self.current_for(device) is 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() for handle in (req.timer, req.deadline): if handle: handle.cancel() await self.emit(device, req, 'voice_error', message=message) async def close(self): for req in self.requests.values(): for handle in (req.timer, req.deadline): if handle: handle.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)