"""Deterministic application backpressure at the real Starlette ASGI boundary. No socket/ESP32 equivalence is assumed; the one-slot queue is an explicit model. """ import asyncio import json from pathlib import Path from types import SimpleNamespace import pytest from starlette.websockets import WebSocket from device_gateway import app as gateway from device_gateway import speech as speech_module from device_gateway.speech import SpeechManager, CHUNK_BYTES from device_gateway.speech_trace import SpeechTrace class VirtualLoop(asyncio.SelectorEventLoop): def __init__(self): self.now = 100.0 super().__init__() def time(self): return self.now def _run_once(self): # Advance only when runnable tasks have drained. asyncio locks, sleeps and # wait_for run unchanged, including the production 3-second send timeout. if not self._ready and self._scheduled: self.now = max(self.now, self._scheduled[0]._when) super()._run_once() def virtual_run(coro): loop = VirtualLoop() try: return loop.run_until_complete(coro) finally: loop.close() async def experiment(delay_ms, monkeypatch, *, frames=16, competitors=True, burst=False, production_cycle=False): loop = asyncio.get_running_loop() records, received = [], [] trace = SpeechTrace(True, clock=loop.time, sink=records.append) monkeypatch.setattr(gateway, 'speech_trace', trace) monkeypatch.setattr(speech_module, 'speech_trace', trace) queue = asyncio.Queue(maxsize=1) # Finite downstream capacity -> real await backpressure. async def receive(): return {'type': 'websocket.connect'} async def asgi_send(message): if message['type'] == 'websocket.send': await queue.put(json.loads(message['text'])) ws = WebSocket({'type': 'websocket'}, receive, asgi_send) await ws.accept() device = gateway.Device('diagnostic-device', 'test', '0.5.1', ws, capabilities={'speech_playback'}) async def synthesize(*_): return b'\0\0' * (CHUNK_BYTES // 2 * frames) manager = SpeechManager(SimpleNamespace(synthesize=synthesize), None, None, lambda _: 'server', gateway.send_to_device, lambda _: True, lambda _: False) async def consume(): while True: payload = await queue.get() received.append((loop.time(), payload['type'], payload.get('sequence'))) # A 2s stall on just one chunk distinguishes drift from catch-up. delay = (2000 if payload.get('sequence') == 1 else 0) if burst else delay_ms await asyncio.sleep(delay / 1000) queue.task_done() async def periodic(kind, interval): for _ in range(8): await asyncio.sleep(interval) await gateway.send_to_device(device, {'type': kind}) async def system_monitor_cycle(): # Actual production topology: snapshot -> agent targets -> sleep 2s. for _ in range(4): await gateway.send_to_device(device, {'type': 'server_snapshot'}) await gateway.send_to_device(device, {'type': 'agent_state'}) await asyncio.sleep(2) consumer = asyncio.create_task(consume()) manager.start(device, 'screen', 'Fixed public test.') background = ([asyncio.create_task(periodic('server_snapshot', .16)), asyncio.create_task(periodic('agent_state', .24))] if competitors else []) if production_cycle: background.append(asyncio.create_task(system_monitor_cycle())) await asyncio.gather(*list(manager.tasks), *background) await queue.join() consumer.cancel() await asyncio.gather(consumer, return_exceptions=True) await manager.close() return records, received @pytest.mark.parametrize('delay', [0, 100, 300, 1000, 2000]) def test_continuous_speech_with_constrained_receiver(delay, monkeypatch): records, received = virtual_run(experiment(delay, monkeypatch)) sends = [r for r in records if r['event'] == 'send'] chunks = [r for r in sends if r['message_type'] == 'speech_chunk'] assert [r['sequence'] for r in chunks] == list(range(16)) assert all(r['error'] is None for r in sends) assert [s for _, t, s in received if t == 'speech_chunk'] == list(range(16)) assert sum(t == 'speech_end' for _, t, _ in received) == 1 assert all(b['send_attempt_ms'] - a['send_complete_ms'] >= 79.999 for a, b in zip(chunks, chunks[1:])) if delay == 0: assert max(r['device_lock_wait_ms'] for r in sends) == 0 if delay >= 1000: assert max(r['device_lock_wait_ms'] for r in chunks) > 1000 assert max(r['device_lock_wait_ms'] for r in sends if r['message_type'] == 'server_snapshot') > 1000 assert records[-1]['speech_end_sent'] def test_two_second_stall_drifts_without_catchup(monkeypatch): records, _ = virtual_run(experiment(0, monkeypatch, competitors=False, burst=True)) chunks = [r for r in records if r.get('message_type') == 'speech_chunk'] assert max(r['websocket_send_ms'] for r in chunks) > 1800 assert chunks[-1]['drift_from_nominal_ms'] > 1800 assert all(b['send_attempt_ms']-a['send_complete_ms'] >= 79.999 for a, b in zip(chunks, chunks[1:])) def test_timeout_and_cancel_release_lock_without_leaking_payload(): async def check(): loop = asyncio.get_running_loop() records = [] trace = SpeechTrace(True, clock=loop.time, sink=records.append) async def blocked(_): await asyncio.sleep(10) d = gateway.Device('d', 'b', 'v', SimpleNamespace(send_json=blocked)) trace.begin(d, 's', 'screen') payload = {'type': 'speech_chunk', 'speech_id': 's', 'sequence': 0, 'audio': 'PRIVATE_PCM', 'token': 'SECRET'} with pytest.raises(TimeoutError): await trace.send(d, payload, scheduled_elapsed_ms=0, raw_pcm_bytes=2) assert records[-1]['websocket_send_ms'] == 3000 assert not d.send_lock.locked() await d.send_lock.acquire() task = asyncio.create_task(trace.send(d, payload)) await asyncio.sleep(5) task.cancel() await asyncio.gather(task, return_exceptions=True) assert records[-1]['device_lock_wait_ms'] == 5000 assert records[-1]['websocket_send_ms'] is None assert records[-1]['error'] == 'CancelledError' d.send_lock.release() trace.finish(d, 's') assert all(x not in json.dumps(records) for x in ('PRIVATE_PCM', 'SECRET', 'audio', 'token')) virtual_run(check()) def test_default_off_and_sampling(monkeypatch, caplog): monkeypatch.delenv('GATEWAY_SPEECH_TRACE', raising=False) assert not SpeechTrace.from_env().enabled async def check(): records = [] trace = SpeechTrace(True, clock=asyncio.get_running_loop().time, sink=records.append) async def fast(_): pass d = gateway.Device('d', 'b', 'v', SimpleNamespace(send_json=fast)) trace.begin(d, 's', 'screen') for i in range(10): await trace.send(d, {'type': 'speech_chunk', 'speech_id': 's', 'sequence': i}, scheduled_elapsed_ms=0) assert len(records) == 11 assert len(caplog.records) == 3 # begin, 0 and 8 assert records[1]['wire_payload_bytes'] == len('{"type":"speech_chunk","speech_id":"s","sequence":0}') with caplog.at_level('INFO'): virtual_run(check()) if __name__ == '__main__': import sys results = {} for delay in (0, 100, 300, 1000, 2000): with pytest.MonkeyPatch.context() as patch: records, received = virtual_run(experiment(delay, patch)) results[str(delay)] = {'model': 'one-slot ASGI queue; periodic snapshot + agent_state', 'trace': records, 'receive': received} with pytest.MonkeyPatch.context() as patch: records, received = virtual_run(experiment(2000, patch, competitors=False, production_cycle=True)) results['production_cycle_2000'] = {'model': 'one-slot ASGI queue; snapshot -> agent -> sleep 2s', 'trace': records, 'receive': received} Path(sys.argv[1]).write_text(json.dumps(results, indent=2) + '\n') def test_snapshot_timeout_closes_device(monkeypatch): async def check(): loop = asyncio.get_running_loop() records, closes = [], [] trace = SpeechTrace(True, clock=loop.time, sink=records.append) monkeypatch.setattr(gateway, 'speech_trace', trace) async def blocked(_): await asyncio.sleep(4) async def close(**kwargs): closes.append(kwargs) d = gateway.Device('isolated-test', 'b', 'v', SimpleNamespace(send_json=blocked, close=close)) monkeypatch.setattr(gateway, 'devices', {d.device_id: d}) trace.begin(d, 'synthetic', 'screen') await gateway.broadcast_server_snapshot({'type': 'server_snapshot'}) assert not d.online and d.websocket is None and closes == [{'code': 1011}] assert records[-1]['error'] == 'TimeoutError' assert not d.send_lock.locked() virtual_run(check()) def test_actual_sequential_system_monitor_also_contends(monkeypatch): records, received = virtual_run(experiment(2000, monkeypatch, competitors=False, production_cycle=True)) speech = [r for r in records if r.get('message_type') == 'speech_chunk'] snapshots = [r for r in records if r.get('message_type') == 'server_snapshot'] assert max(r['device_lock_wait_ms'] for r in speech) > 1000 assert max(r['websocket_send_ms'] for r in snapshots) == 2000 agents = [r for r in records if r.get('message_type') == 'agent_state'] assert max(r['device_lock_wait_ms'] for r in agents) == 2000 assert any(t == 'speech_end' for _, t, _ in received) def test_disabled_trace_does_not_serialize_or_mutate(monkeypatch): async def check(): observed = [] async def send(payload): observed.append(payload) d = gateway.Device('d', 'b', 'v', SimpleNamespace(send_json=send)) trace = SpeechTrace(False, sink=lambda _: pytest.fail('disabled trace emitted')) trace.begin(d, 's', 'screen') def forbidden(*_, **__): pytest.fail('disabled trace serialized') monkeypatch.setattr('device_gateway.speech_trace.json.dumps', forbidden) payload = {'type': 'speech_chunk', 'audio': 'unchanged'} await trace.send(d, payload) assert observed[0] is payload and not trace.streams virtual_run(check())