import asyncio
import base64
import json
import stat
from types import SimpleNamespace

import httpx
import pytest

from device_gateway.speech import (
    SpeechPreferences, speech_text_cleanup, language_for, CloudflareMeloTTS, SpeechError,
    audio_to_pcm, HaihubSummary, SpeechManager, CHUNK_BYTES, MODES, configured_tts,
)
from device_gateway.voice import wav_pcm
from device_gateway.integrations.pi import PiSessionRegistry, PiSessionEvent
from device_gateway.control.codex import CodexAdapter


def run(coro): return asyncio.run(coro)


def test_preferences_private_atomic_and_per_target(tmp_path):
    path=tmp_path/'config'/'speech.json'; store=SpeechPreferences(path)
    assert store.get('d','agent:pi:a')=='OFF'
    store.set('d','pi:a','AUTO_REPLY');store.set('d','agent:codex:b','AUTO_SUMMARY')
    assert SpeechPreferences(path).get('d','agent:pi:a')=='AUTO_REPLY'
    assert store.get('d','agent:codex:b')=='AUTO_SUMMARY'
    assert store.get('other','agent:pi:a')=='OFF'
    assert stat.S_IMODE(path.stat().st_mode)==0o600
    assert len(list(path.parent.iterdir()))==1
    with pytest.raises(ValueError):store.set('d','server','BAD')
    path.write_text('{malformed'); assert store.get('d','server')=='OFF'


def test_cleanup_language_and_sentence_boundaries():
    value=speech_text_cleanup('## 完成\n**修复** `Codex`。\n```python\nprivate_code()\n```\n文件 /home/lighthouse/foo/bar/baz.py。https://example.com/a   🙂\x00')
    assert 'Codex' in value and '文件 baz.py' in value
    assert all(x not in value for x in ('private_code','https','/home','**','🙂','\x00'))
    text='结果完成。'+'需要详细说明'*140
    limited=speech_text_cleanup(text,600,notice=True)
    assert limited=='结果完成。回答比较长，更多内容请查看屏幕。'
    assert len(speech_text_cleanup('x'*1500,600,notice=True))<=600
    assert language_for('Codex 已完成 pytest')=='zh' and language_for('All tests passed.')=='en'
    assert 'secret_value' not in speech_text_cleanup('api_key=secret_value Bearer private_value')


def mock_http(monkeypatch,handler):
    client=httpx.AsyncClient
    monkeypatch.setattr(httpx,'AsyncClient',lambda **kw: client(**kw,transport=httpx.MockTransport(handler)))


@pytest.mark.parametrize('encoding',['json','binary'])
def test_melo_schema_and_pcm_decode(monkeypatch,encoding):
    audio=wav_pcm(b'\x00\x10'*4410)
    def handler(request):
        assert json.loads(request.content)=={'prompt':'Codex 完成。','lang':'zh'}
        assert request.headers['authorization']=='Bearer private-key'
        if encoding=='json':return httpx.Response(200,json={'success':True,'errors':[],'result':{'audio':base64.b64encode(audio).decode()}})
        return httpx.Response(200,content=audio,headers={'content-type':'audio/mpeg'})
    mock_http(monkeypatch,handler)
    pcm=run(CloudflareMeloTTS('account','private-key').synthesize('Codex 完成。','zh'))
    assert pcm==b'\x00\x10'*4410


@pytest.mark.parametrize('failure',[401,429,500,302,'bad_json','bad_audio','error','network'])
def test_melo_failures_safe(monkeypatch,failure):
    def handler(request):
        if failure=='network':raise httpx.ConnectError('private-key')
        if isinstance(failure,int):return httpx.Response(failure,text='private-key')
        if failure=='bad_json':return httpx.Response(200,text='private-key',headers={'content-type':'application/json'})
        return httpx.Response(200,json={'success':failure!='error','result':{'audio':'%%%'}})
    mock_http(monkeypatch,handler)
    with pytest.raises(SpeechError,match='^Speech unavailable$'):
        run(CloudflareMeloTTS('account','private-key').synthesize('Test.','en'))


def test_ffmpeg_bounds_errors_timeout():
    audio=wav_pcm(b'\0\0'*4410)
    for options in ({'max_input':10},{'max_output':100},{'timeout':.00001}):
        with pytest.raises(SpeechError):run(audio_to_pcm(audio,**options))
    with pytest.raises(SpeechError):run(audio_to_pcm(b'broken compressed audio'))


def test_configured_tts_reuses_cloudflare_key(monkeypatch):
    monkeypatch.setenv('GATEWAY_TTS_PROVIDER','cloudflare-workers-ai')
    monkeypatch.setenv('GATEWAY_CF_ACCOUNT_ID','a');monkeypatch.setenv('GATEWAY_CF_API_TOKEN','same')
    assert configured_tts().api_key=='same'
    monkeypatch.delenv('GATEWAY_CF_API_TOKEN');assert configured_tts() is None


def summary_provider(tmp_path):
    path=tmp_path/'models.json';path.write_text(json.dumps({'providers':{'haihub':{'api':'openai-completions','baseUrl':'https://api.model.haihub.cn/v1','apiKey':'private-key','models':[{'id':'deepseek-flash'}]}}}))
    return HaihubSummary(path)


@pytest.mark.parametrize('status',['success','partial','failed','needs_action'])
def test_summary_structured_and_minimal_input(tmp_path,monkeypatch,status):
    def handler(request):
        payload=json.loads(request.content)
        assert payload['model']=='deepseek-flash' and payload['response_format']=={'type':'json_object'}
        source=json.loads(payload['messages'][1]['content'])
        assert source==dict(agent='codex',session_name='test',terminal_state='DONE',final_response='public result')
        return httpx.Response(200,json={'choices':[{'message':{'content':json.dumps({'status':status,'speech':'任务结果已经返回，请查看屏幕。'})}}]})
    mock_http(monkeypatch,handler)
    result=run(summary_provider(tmp_path).summarize('codex','test','DONE','public result','ignored tools'))
    assert result['status']==status and not result.get('fallback')


@pytest.mark.parametrize('case',['json','network','success_on_failure','long','sentences','empty','bad_status'])
def test_summary_failure_fallback(tmp_path,monkeypatch,case):
    def handler(request):
        if case=='network':raise httpx.ReadTimeout('private-key')
        value={'status':'success' if case=='success_on_failure' else 'failed','speech':'任务没有完成。'}
        if case=='long':value['speech']='很长'*121
        if case=='sentences':value['speech']='失败。'*4
        if case=='empty':value['speech']=''
        if case=='bad_status':value['status']='unknown'
        return httpx.Response(200,json={'choices':[{'message':{'content':'not-json' if case=='json' else json.dumps(value)}}]})
    mock_http(monkeypatch,handler)
    result=run(summary_provider(tmp_path).summarize('codex','test','FAILED','dependency missing'))
    assert result['fallback'] and result['status']=='failed'
    assert result['speech']=='任务没有完成，请查看屏幕上的错误信息。'


class FakeTTS:
    def __init__(self, frames=2):self.calls=[];self.frames=frames
    async def synthesize(self,text,language):
        self.calls.append((text,language));return b'\0'*CHUNK_BYTES*self.frames


class FakeSummary:
    def __init__(self):self.calls=[]
    async def summarize(self,**kw):self.calls.append(kw);return {'status':'success','speech':'任务已经完成。'}


def manager(tmp_path,frames=2):
    device=SimpleNamespace(device_id='d',connection_id='c',capabilities={'speech_playback'},online=True)
    emitted=[]; target=['agent:pi:s']; blocked=[False]
    async def send(d,p):emitted.append((asyncio.get_running_loop().time(),p))
    tts,summary=FakeTTS(frames),FakeSummary()
    m=SpeechManager(tts,summary,SpeechPreferences(tmp_path/'speech.json'),lambda _:target[0],send,lambda d:d.online,lambda _:blocked[0])
    return m,device,emitted,target,blocked


@pytest.mark.parametrize('mode',MODES)
def test_terminal_modes_and_pacing(tmp_path,mode):
    async def check():
        m,d,out,_,_=manager(tmp_path)
        m.preferences.set('d','agent:pi:s',mode)
        m.terminal(d,'pi','s','DONE','Final answer.','test')
        await asyncio.gather(*list(m.tasks))
        if mode in ('OFF','MANUAL'):
            assert not out and not m.provider.calls and not m.summarizer.calls
        else:
            assert [p['type'] for _,p in out]==['speech_start','speech_chunk','speech_chunk','speech_end']
            assert out[2][0]-out[1][0]>=.075
            assert len(json.dumps(out[1][1]).encode())<10000
            assert len(base64.b64decode(out[1][1]['audio']))==7056
            assert bool(m.summarizer.calls)==(mode=='AUTO_SUMMARY')
        await m.close()
    run(check())


@pytest.mark.parametrize('reason',['replace','target','disconnect','approval','recording','settings','overflow'])
def test_cancellation_and_stale_ids(tmp_path,reason):
    async def check():
        m,d,out,target,blocked=manager(tmp_path,30)
        assert m.start(d,'screen','A public screen.')
        await asyncio.sleep(.01)
        old=m.active['c'].speech_id
        await m.handle(d,dict(type='speech_stop',speech_id='stale'))
        assert m.active['c'].speech_id==old
        if reason=='target': target[0]='server'
        if reason=='disconnect': d.online=False
        if reason in ('approval','recording'):blocked[0]=True
        if reason=='replace':m.start(d,'reply','Replacement.')
        else:m.cancel(d,notify=reason!='disconnect',forget=reason=='disconnect')
        await asyncio.sleep(.12)
        assert not any(p['type']=='speech_end' and p['speech_id']==old for _,p in out)
        assert sum(p['type']=='speech_chunk' and p['speech_id']==old for _,p in out)==1
        if reason!='disconnect':assert any(p['type']=='speech_cancel' and p['speech_id']==old for _,p in out)
        await m.close()
    run(check())


def test_legacy_and_bounded_manual_request(tmp_path):
    async def check():
        m,d,out,_,_=manager(tmp_path)
        d.capabilities=set()
        assert not m.start(d,'screen','Legacy.')
        await m.settings(d);assert not out
        d.capabilities={'speech_playback'}
        request=dict(type='speech_request',request_id='r',target='agent:pi:s',source='screen',text='x'*1201)
        await m.handle(d,request);assert not m.active
        request.update(request_id='r2',text='Read this screen.')
        await m.handle(d,request);await asyncio.gather(*list(m.tasks))
        await m.handle(d,request);assert len(m.provider.calls)==1
        await m.close()
    run(check())


def test_pi_final_turn_identity_and_dedup():
    registry=PiSessionRegistry()
    def event(kind,**kw):return PiSessionEvent(event=kind,session_id='s',name='test',cwd='/tmp',**kw)
    registry.apply(event('agent_start',turn_id='one'))
    registry.apply(event('activity',activity_type='tool',activity='tool logs'))
    first=registry.apply(event('agent_end',turn_id='one',final_text='Final only.'))
    assert first.final_text=='Final only.' and first.pending_notification=='done'
    assert registry.apply(event('agent_end',turn_id='one',final_text='Duplicate')).pending_notification is None
    registry.apply(event('agent_start',turn_id='two'))
    stale=registry.apply(event('agent_end',turn_id='one',final_text='Stale'))
    assert stale.state=='RUNNING' and stale.final_text==''
    assert registry.apply(event('agent_end',turn_id='two',final_text='Failed reply.',failed=True)).state=='FAILED'


def test_codex_final_message_only_and_terminal_dedup():
    async def check():
        adapter=CodexAdapter(None);calls=[];adapter.on_terminal=lambda *a,**kw:calls.append(a)
        async def event(method,**params):await adapter.event(dict(method=method,params={'threadId':'s',**params}))
        await event('turn/started',turn={'id':'t'})
        await event('item/agentMessage/delta',turnId='t',itemId='a',delta='Intermediate thought.')
        await event('item/completed',turnId='t',item={'id':'a','type':'agentMessage','phase':'commentary','text':'Intermediate thought.'})
        await event('item/agentMessage/delta',turnId='stale',itemId='old',delta='OLD SECRET')
        await event('item/agentMessage/delta',turnId='t',itemId='b',delta='Final ')
        await event('item/agentMessage/delta',turnId='t',itemId='b',delta='answer.')
        assert not calls
        await event('turn/completed',turn={'id':'t','status':'completed'})
        await event('turn/completed',turn={'id':'t','status':'completed'})
        assert len(calls)==1 and calls[0][3]=='Final answer.'
        await event('turn/started',turn={'id':'u'})
        await event('turn/completed',turn={'id':'t','status':'completed'})
        assert len(calls)==1
        await event('turn/completed',turn={'id':'u','status':'failed'})
        assert calls[-1][2:] == ('FAILED','')
    run(check())

@pytest.fixture
def speech_gateway(tmp_path,monkeypatch):
    from fastapi.testclient import TestClient
    import device_gateway.app as gateway
    from device_gateway.auth import DeviceCredentialStore
    from device_gateway.state import TargetStore
    from test_voice_settings import FakeAgent, FakeSTT
    credential=tmp_path/'device.json'
    credential.write_text(json.dumps(dict(device_id='speech-test',device_token='secret',pi_token='pi-secret')))
    monkeypatch.setattr(gateway,'credential_store',DeviceCredentialStore(credential))
    monkeypatch.setattr(gateway,'target_store',TargetStore(tmp_path/'targets.json'))
    monkeypatch.setattr(gateway,'devices',{})
    monkeypatch.setattr(gateway,'pi_sessions',PiSessionRegistry())
    monkeypatch.setattr(gateway,'adapters',{'pi':FakeAgent('pi',tmp_path)})
    monkeypatch.setattr(gateway.speeches,'preferences',SpeechPreferences(tmp_path/'speech.json'))
    monkeypatch.setattr(gateway.speeches,'provider',FakeTTS(4))
    monkeypatch.setattr(gateway.speeches,'summarizer',FakeSummary())
    monkeypatch.setattr(gateway.voices,'provider',FakeSTT())
    with TestClient(gateway.app) as client:yield gateway,client


def recv(ws,kind):
    for _ in range(100):
        event=ws.receive_json()
        if event['type']==kind:return event
    raise AssertionError(kind)


def hello(ws,legacy=False):
    value=dict(type='hello',device_id='speech-test',device_token='secret',board='fake',firmware='0.4.0' if legacy else '0.5.0')
    if not legacy:value['capabilities']=['speech_playback']
    ws.send_json(value);recv(ws,'hello_ack');recv(ws,'target_state')
    if not legacy:assert recv(ws,'speech_settings')['mode']=='OFF'


def test_authenticated_manual_and_settings(speech_gateway):
    gateway,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        hello(ws)
        ws.send_json(dict(type='speech_settings_set',request_id='one',target='server',mode='MANUAL'))
        assert recv(ws,'speech_settings')['mode']=='MANUAL'
        ws.send_json(dict(type='speech_request',request_id='read',target='server',source='screen',text='CPU 18 percent.'))
        assert recv(ws,'speech_start')['source']=='screen'
        assert recv(ws,'speech_chunk')['sequence']==0
        assert recv(ws,'speech_end')['chunks']==4
        assert gateway.speeches.preferences.get('speech-test','server')=='MANUAL'
        status=client.get('/devices').json()[0]
        assert status['speech_mode']=='MANUAL' and 'CPU 18 percent' not in json.dumps(status)
        assert client.post('/api/devices/speech-test/speech-test',json={'target':'server','text':'arbitrary'}).status_code==422


def test_settings_only_firmware_saves_mode_without_playback(speech_gateway):
    """speech-settings alone selects a mode; it can never enable playback."""
    gateway,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        ws.send_json(dict(type='hello',device_id='speech-test',device_token='secret',board='fake',
                          firmware='0.6.3-dev',capabilities=['speech-settings']))
        recv(ws,'hello_ack');recv(ws,'target_state')
        assert recv(ws,'speech_settings')['mode']=='OFF'
        ws.send_json(dict(type='speech_settings_set',request_id='one',target='server',mode='AUTO_SUMMARY'))
        assert recv(ws,'speech_settings')['mode']=='AUTO_SUMMARY'
        assert gateway.speeches.preferences.get('speech-test','server')=='AUTO_SUMMARY'
        # Playback stays unavailable: no advertisement means no speech_* is sent.
        assert client.post('/api/devices/speech-test/speech-test',json={'target':'server'}).status_code==409
        assert client.post('/api/devices/speech-test/speech',json=dict(target='server',mode='AUTO_REPLY')).status_code==200
        ws.send_json(dict(type='speech_request',request_id='read',target='server',source='screen',text='CPU 18 percent.'))
        client.post('/devices/speech-test/message',json={'title':'sentinel','body':'settings only'})
        while True:
            event=ws.receive_json()
            assert not event['type'].startswith('speech_start')
            assert not event['type'].startswith('speech_chunk')
            assert not event['type'].startswith('speech_end')
            if event['type']=='display_message':break
        assert not gateway.speeches.active


def test_old_firmware_never_receives_speech(speech_gateway):
    _,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        hello(ws,legacy=True)
        assert client.post('/api/devices/speech-test/speech',json=dict(target='server',mode='AUTO_REPLY')).status_code==200
        assert client.post('/api/devices/speech-test/speech-test',json={'target':'server'}).status_code==409
        client.post('/devices/speech-test/message',json={'title':'sentinel','body':'old device works'})
        while True:
            event=ws.receive_json();assert not event['type'].startswith('speech_')
            if event['type']=='display_message':break


@pytest.mark.parametrize('preempt',['voice','approval','target','settings'])
def test_authenticated_preemption(speech_gateway,preempt):
    gateway,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        hello(ws)
        client.post('/devices/speech-test/target',json={'target':'agent:pi:s'})
        recv(ws,'agent_state');recv(ws,'speech_settings')
        assert client.post('/api/devices/speech-test/speech-test',json={'target':'agent:pi:s'}).status_code==200
        old=recv(ws,'speech_start')['speech_id']
        if preempt=='voice':ws.send_json(dict(type='voice_start',voice_id='v',sample_rate=44100,bits=16,channels=1))
        elif preempt=='target':client.post('/devices/speech-test/target',json={'target':'server'})
        elif preempt=='settings':ws.send_json(dict(type='settings_get',request_id='r'))
        else:
            client.portal.call(lambda:gateway.approvals.create(agent='pi',session_id='s',project_id='p',kind='tool',title='Test',detail='Test',owner='test',request_identity='once'))
        assert recv(ws,'speech_cancel')['speech_id']==old
        assert not gateway.speeches.active


def test_authenticated_pi_terminal_final_and_duplicate(speech_gateway):
    gateway,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        hello(ws)
        client.post('/devices/speech-test/target',json={'target':'agent:pi:s'})
        recv(ws,'agent_state');recv(ws,'speech_settings')
        client.post('/api/devices/speech-test/speech',json=dict(target='agent:pi:s',mode='AUTO_REPLY'))
        def post(kind,**extra):
            return client.post('/integrations/pi/events',headers={'x-device-lab-pi-token':'pi-secret'},json=dict(event=kind,session_id='s',name='public test',cwd='/tmp',turn_id='t',**extra))
        assert post('agent_start').status_code==200
        post('activity',activity_type='text',activity='Intermediate text.')
        assert not gateway.speeches.provider.calls
        post('agent_end',final_text='Only the final answer.')
        recv(ws,'speech_end')
        post('agent_end',final_text='Duplicate must not speak.')
        assert gateway.speeches.provider.calls==[('Only the final answer.','en')]
        assert not gateway.speeches.summarizer.calls
        assert client.post('/api/devices/speech-test/speech',json=dict(target='agent:pi:stale',mode='MANUAL')).status_code==409


def test_codex_reconnect_terminal_updates_monitor_without_speaking():
    async def check():
        adapter=CodexAdapter(None);calls=[];adapter.on_terminal=lambda *a,**kw:calls.append(a)
        await adapter.event({'method':'thread/status/changed','params':{'threadId':'s','status':{'type':'active'}}})
        await adapter.event({'method':'turn/completed','params':{'threadId':'s','turn':{'id':'joined-mid-turn','status':'completed'}}})
        assert adapter.live['s']['state']=='DONE' and not calls
    run(check())


def test_large_ffmpeg_output_cancel_reaps_process():
    async def check():
        async with asyncio.timeout(3):
            with pytest.raises(SpeechError):
                await audio_to_pcm(wav_pcm(b'\0\0'*44100*8),max_output=100)
    run(check())


def test_disconnect_clears_active_playback(speech_gateway):
    gateway,client=speech_gateway
    with client.websocket_connect('/device/ws') as ws:
        hello(ws)
        client.post('/api/devices/speech-test/speech-test',json={'target':'server'})
        recv(ws,'speech_start')
    assert not gateway.speeches.active


def test_slow_synthesis_cancel_cannot_start_later(tmp_path):
    async def check():
        m,d,out,_,_=manager(tmp_path)
        cancelled=asyncio.Event()
        class Slow:
            async def synthesize(self,*args):
                try:await asyncio.sleep(60)
                finally:cancelled.set()
        m.provider=Slow();m.start(d,'screen','Screen.')
        await asyncio.sleep(.01);m.cancel(d)
        await asyncio.wait_for(cancelled.wait(),1)
        assert not any(p['type']=='speech_start' for _,p in out)
        await m.close()
    run(check())