#!/usr/bin/env python3
"""Real Pi/Codex + real TTS/summary + authenticated TLS synthetic playback.
Only temporary projects/sessions; never uses the production device or gateway.
"""
import asyncio
import base64
import importlib.util
import json
import os
from pathlib import Path
import socket
import ssl
import subprocess
import sys
import tempfile
import time

import httpx
import uvicorn
from websockets.asyncio.client import connect

REPO=Path(__file__).resolve().parents[1]
sys.path.insert(0,str(REPO/'server'))
spec=importlib.util.spec_from_file_location('provider_smoke',REPO/'scripts/smoke-speech.py')
providers=importlib.util.module_from_spec(spec);spec.loader.exec_module(providers)


class SyntheticSpeaker:
    def __init__(self):
        self.queue=asyncio.Queue(maxsize=8);self.sid=None;self.seq=0;self.starts=[];self.ends=[]
        self.max_queue=0;self.frames=0;self.bytes=0;self.cancelled=[];self.errors=[]
        self.state_events=asyncio.Queue();self.wire_frames=[]
    async def consume(self):
        while True:
            sid,pcm=await self.queue.get()
            if sid==self.sid:await asyncio.sleep(len(pcm)/88200)
            self.queue.task_done()
    async def read(self,ws):
        async for wire in ws:
            event=json.loads(wire);kind=event['type']
            self.wire_frames.append((kind,len(wire.encode())))
            if kind=='speech_start':
                assert (event['sample_rate'],event['bits'],event['channels'])==(44100,16,1)
                self.sid=event['speech_id'];self.seq=0;self.starts.append(event)
            elif kind=='speech_chunk':
                assert event['speech_id']==self.sid and event['sequence']==self.seq
                pcm=base64.b64decode(event['audio'],validate=True)
                assert len(pcm)%2==0 and 0<len(pcm)<=7056 and len(wire.encode())<10000
                self.seq+=1;self.frames+=1;self.bytes+=len(pcm)
                self.queue.put_nowait((self.sid,pcm));self.max_queue=max(self.max_queue,self.queue.qsize())
            elif kind=='speech_end':
                assert event['speech_id']==self.sid and event['chunks']==self.seq
                self.ends.append(event)
            elif kind=='speech_cancel':self.cancelled.append(event['speech_id'])
            elif kind=='speech_error':self.errors.append(event)
            await self.state_events.put(event if kind!='speech_chunk' else {'type':kind})
    async def wait(self,kind,timeout=150):
        async with asyncio.timeout(timeout):
            while True:
                event=await self.state_events.get()
                if event['type']==kind:return event
    async def playback(self):
        await self.wait('speech_end');await self.queue.join()
        assert not self.errors


async def main():
    providers.load_provider_env()
    root=Path(tempfile.mkdtemp(prefix='spoken-agent-proof-'))
    print('Proof directory:',root,flush=True)
    for name in ('pi','codex'):(root/name).mkdir()
    creds=root/'credentials.json';creds.write_text(json.dumps({'device_id':'synthetic-spoken','device_token':'local-device-only','pi_token':'local-pi-only'}));creds.chmod(0o600)
    cert,key=root/'cert.pem',root/'key.pem'
    subprocess.run(['openssl','req','-x509','-newkey','rsa:2048','-nodes','-days','1','-subj','/CN=localhost',
        '-addext','subjectAltName=DNS:localhost,IP:127.0.0.1','-keyout',str(key),'-out',str(cert)],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
    with socket.socket() as sock:sock.bind(('127.0.0.1',0));port=sock.getsockname()[1]
    os.environ.update(DEVICE_CREDENTIALS_FILE=str(creds),DEVICE_LAB_CREDENTIALS_FILE=str(creds),
        GATEWAY_STATE_FILE=str(root/'targets.json'),GATEWAY_PROJECTS_FILE=str(root/'projects.json'),
        GATEWAY_SPEECH_FILE=str(root/'speech.json'),DEVICE_LAB_CONTROL_DIR=str(root/'runtime'),
        DEVICE_LAB_GATEWAY_URL=f'https://127.0.0.1:{port}',NODE_EXTRA_CA_CERTS=str(cert),
        GATEWAY_CODEX_ENABLED='1',GATEWAY_TTS_PROVIDER='cloudflare-workers-ai')
    import device_gateway.app as gateway
    from device_gateway.control.models import Project
    from device_gateway.speech import HaihubSummary
    from device_gateway.voice import VoiceManager
    for name in ('pi','codex'):gateway.project_registry.add(Project(id=name,name='Public spoken smoke',path=str(root/name)))
    # Deterministic microphone text; real STT is independently checked in smoke-speech.py.
    class STT:
        async def transcribe(self,audio):return '请用两句话解释天空为什么是蓝色。不要调用工具。'
    gateway.voices.provider=STT()
    summary_calls=[]
    class Summary(HaihubSummary):
        async def summarize(self,**kw):
            result=await super().summarize(**kw);assert not result.get('fallback')
            summary_calls.append(result);return result
    gateway.speeches.summarizer=Summary()
    context=ssl.create_default_context(cafile=str(cert))
    server=uvicorn.Server(uvicorn.Config(gateway.app,host='127.0.0.1',port=port,log_level='error',
        ssl_certfile=str(cert),ssl_keyfile=str(key),ws_max_size=8192))
    serving=asyncio.create_task(server.serve());result={};reader=consumer=None
    try:
        async with asyncio.timeout(15):
            while not server.started:await asyncio.sleep(.05)
        async with httpx.AsyncClient(base_url=f'https://127.0.0.1:{port}',verify=context,timeout=45) as client:
            async with connect(f'wss://127.0.0.1:{port}/device/ws',ssl=context) as ws:
                speaker=SyntheticSpeaker();reader=asyncio.create_task(speaker.read(ws));consumer=asyncio.create_task(speaker.consume())
                async def send(value):await ws.send(json.dumps(value))
                async def post(path,data):
                    response=await client.post(path,json=data);response.raise_for_status();return response.json()
                await send(dict(type='hello',device_id='synthetic-spoken',device_token='local-device-only',board='synthetic',firmware='0.5.0',capabilities=['speech_playback']))
                await speaker.wait('hello_ack');await speaker.wait('speech_settings')
                # Explicit read screen under OFF, keeping the target unchanged.
                await send(dict(type='speech_request',request_id='read-screen',source='screen',target='server',text='CPU 18 percent. Memory 52 percent.'))
                await speaker.playback();result['read_screen']='PASS'
                pi=gateway.adapters['pi'];codex=gateway.adapters['codex']
                sid=(await pi.rpc.call('create_smoke',{'cwd':str(root/'pi')}))['session_id']
                await post('/devices/synthetic-spoken/target',{'target':f'agent:pi:{sid}'})
                await post('/api/devices/synthetic-spoken/speech',{'target':f'agent:pi:{sid}','mode':'AUTO_REPLY'})
                before=len(speaker.starts)
                await send(dict(type='voice_start',voice_id='conversation',sample_rate=44100,bits=16,channels=1))
                await speaker.wait('voice_state')
                await send(dict(type='voice_chunk',voice_id='conversation',sequence=0,audio=base64.b64encode(b'\0'*3528).decode()))
                await send(dict(type='voice_end',voice_id='conversation'))
                transcript=await speaker.wait('voice_transcript')
                await send(dict(type='voice_decision',voice_id='conversation',decision='send'))
                await speaker.playback();await asyncio.sleep(.5)
                assert len(speaker.starts)==before+1 and not summary_calls
                final=gateway.pi_sessions.get(sid).final_text
                assert final and speaker.starts[-1]['source']=='reply'
                result['conversation']=dict(status='PASS',stt='FAKE',pi='REAL 0.85.1',tts='REAL',final_text=final,
                    session_id=sid,speeches=1,summaries=0)
                print('Conversation AUTO_REPLY PASS',flush=True)
                await post('/devices/synthetic-spoken/target',{'target':'server'})
                created=await codex.call('thread/start',dict(cwd=str(root/'codex'),approvalPolicy='on-request',sandbox='read-only',
                    baseInstructions='Public connectivity test. Never use tools or access files. Follow the user instruction.'))
                cid=created['thread']['id']
                await codex.call('turn/start',{'threadId':cid,'input':[{'type':'text','text':'Reply exactly READY.'}]})
                async with asyncio.timeout(120):
                    while codex.live.get(cid,{}).get('state') not in ('DONE','FAILED'):await asyncio.sleep(.2)
                assert codex.live[cid]['state']=='DONE'
                await post('/devices/synthetic-spoken/target',{'target':f'agent:codex:{cid}'})
                await post('/api/devices/synthetic-spoken/speech',{'target':f'agent:codex:{cid}','mode':'AUTO_SUMMARY'})
                before=len(speaker.starts)
                await codex.send_input(cid,'执行一个纯文本检查：判断 2+2 是否等于 4，并用中文说明结论。不要使用工具或修改文件。')
                await speaker.playback();await asyncio.sleep(.5)
                assert len(speaker.starts)==before+1 and len(summary_calls)==1
                assert speaker.starts[-1]['source']=='summary'
                result['coding']=dict(status='PASS',codex='REAL',summary=summary_calls[0],tts='REAL',
                    final_text=codex.live[cid]['final_text'],session_id=cid,speeches=1,summaries=1)
                result['transport']=dict(authenticated=True,tls=True,frames=speaker.frames,pcm_bytes=speaker.bytes,
                    queue_high_water=speaker.max_queue,max_wire_bytes=max(size for _,size in speaker.wire_frames))
                print('Coding AUTO_SUMMARY PASS',flush=True)
                await codex.call('thread/unsubscribe',{'threadId':cid})
                await pi.rpc.call('stop',{'session_id':sid})
    finally:
        for task in (reader,consumer):
            if task:task.cancel()
        await asyncio.gather(*(t for t in (reader,consumer) if t),return_exceptions=True)
        server.should_exit=True
        await asyncio.wait_for(serving,30)
        (root/'result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n')
        print(json.dumps(result,ensure_ascii=False,indent=2),flush=True)
    assert result.get('conversation',{}).get('status')=='PASS' and result.get('coding',{}).get('status')=='PASS'

if __name__=='__main__':
    try:asyncio.run(main())
    except Exception as exc:
        print('Software smoke FAILED:',type(exc).__name__,flush=True)
        raise SystemExit(1) from None