import sys,asyncio,threading,json,io,unittest,tempfile,struct
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0,'/tmp/jarvis-pr8-perf-audit/apps/desktop/pipecat/src')
from jarvis_voice_runtime.pocket import PocketDaemon,JarvisPocketTTSService,DaemonError
from jarvis_voice_runtime.runtime import Runtime
class Regressions(unittest.IsolatedAsyncioTestCase):
 async def test_pcm_reaches_service_before_daemon_terminal(self):
  with tempfile.TemporaryDirectory() as d:
   path=Path(d)/'chunk.wav';data=struct.pack('<480f',*([.2]*480));path.write_bytes(struct.pack('<4sI4s4sIHHIIHH4sI',b'RIFF',36+len(data),b'WAVE',b'fmt ',16,3,1,24000,96000,4,32,b'data',len(data))+data)
   release=threading.Event();waiting=threading.Event();request={};calls=0
   class Stdout:
    def readline(self):
     nonlocal calls
     calls+=1
     if calls==1:return json.dumps({'type':'chunk','requestId':request['requestId'],'index':0,'path':str(path)})+'\n'
     waiting.set();release.wait(3)
     return json.dumps({'type':'synthesis-finished','requestId':request['requestId'],'sampleRate':24000})+'\n'
   with patch('jarvis_voice_runtime.pocket.validate_pocket_root'):
    daemon=PocketDaemon(Path(d))
   daemon._send=lambda c:request.update(c)
   daemon._process=type('Process',(),{'stdout':Stdout()})()
   native=type('Native',(),{'daemon':daemon,'sample_rate':24000,'ensure_running':lambda self:None})()
   service=JarvisPocketTTSService(native);stream=service.run_tts('Done.','ctx');first=asyncio.create_task(anext(stream))
   try:
    await asyncio.to_thread(waiting.wait,1)
    try:await asyncio.wait_for(asyncio.shield(first),.1)
    except asyncio.TimeoutError:pass
    self.assertTrue(first.done(),'No PCM delivered while the daemon withholds its terminal event')
   finally:
    release.set();await first
    async for _ in stream:pass
 async def test_abandoned_native_handle_is_closed(self):
  class Native:
   closed=False
   def close(self):self.closed=True
  native=Native()
  def factory(_):
   runtime._shutdown_requested=True
   return native
  runtime=Runtime(Path('/tmp'),pocket_root=Path('/tmp'),tts_factory=factory)
  await runtime._prepare_speech(remote=True)
  self.assertTrue(native.closed,'A created daemon must close when shutdown wins preparation')
 async def test_ready_pipe_on_windows_is_supported(self):
  class Process:
   stdin=io.StringIO();stdout=io.StringIO('{"type":"ready"}\n');stderr=io.StringIO()
   def poll(self):return None
   def kill(self):pass
  with patch('jarvis_voice_runtime.pocket.validate_pocket_root'),patch('jarvis_voice_runtime.pocket.subprocess.Popen',return_value=Process()),patch('jarvis_voice_runtime.pocket.select.select',side_effect=OSError(10038,'not a socket')),patch('jarvis_voice_runtime.pocket.DAEMON_STARTUP_TIMEOUT_SECS',.01):
   daemon=PocketDaemon(Path('/tmp'))
   try:daemon.start()
   except DaemonError as error:self.fail('Ready daemon rejected by Windows pipe handling: '+str(error))
if __name__=='__main__':unittest.main()