"""Device lifecycle is independent of Agent execution; reconnect sends latest state."""
import asyncio
import base64
import json

import pytest
from fastapi import WebSocketDisconnect

import device_gateway.app as gateway
from device_gateway.auth import DeviceCredentialStore
from device_gateway.control.models import AgentSession
from device_gateway.integrations.pi import PiSessionEvent, PiSessionRegistry
from device_gateway.state import TargetStore
from device_gateway.voice import VoiceManager
from test_gateway import FakeWebSocket


@pytest.mark.parametrize('terminal', ['DONE', 'FAILED', 'RUNNING'])
def test_agent_continues_and_hello_restores_authoritative_state(tmp_path, monkeypatch, terminal):
    async def scenario():
        credential = tmp_path/'credentials.json'
        credential.write_text(json.dumps(dict(device_id='network-test', device_token='test-only', pi_token='pi-only')))
        registry = PiSessionRegistry()
        targets = TargetStore(tmp_path/'targets.json')
        targets.set('network-test', 'agent:pi:s')
        class Adapter:
            calls = []
            async def get_session(self, sid):
                return AgentSession(agent='pi', session_id=sid, project_id='p', name='Selected Pi',
                    cwd=str(tmp_path), state='IDLE', live=True, updated_at='2026-10-04T00:00:00Z')
            async def send_input(self, *args): self.calls.append(('input', args))
            async def close(self): self.calls.append(('close',))
        class STT:
            calls = 0
            async def transcribe(self, data): self.calls += 1; return 'fixture transcript'
        adapter, stt = Adapter(), STT()
        for key, value in dict(devices={}, credential_store=DeviceCredentialStore(credential),
                              target_store=targets, pi_sessions=registry, adapters={'pi':adapter}).items():
            monkeypatch.setattr(gateway, key, value)
        voices = VoiceManager(stt, lambda did: targets.get(did), lambda _: adapter,
                              gateway.send_to_device, gateway.current_device, lambda: False)
        monkeypatch.setattr(gateway, 'voices', voices)
        async def event(kind, **extra):
            return await gateway.pi_event(PiSessionEvent(event=kind, session_id='s', name='Selected Pi',
                cwd=str(tmp_path), turn_id='turn', **extra), 'pi-only')
        async def receive(ws, kind):
            for _ in range(10):
                message = await asyncio.wait_for(ws.sent.get(), 1)
                if message['type'] == kind: return message
            raise AssertionError(kind)
        async def connect():
            ws = FakeWebSocket()
            await ws.incoming.put(dict(type='hello', device_id='network-test', device_token='test-only',
                board='synthetic', firmware='0.5.1', capabilities=['voice_input']))
            task = asyncio.create_task(gateway.device_websocket(ws))
            await receive(ws, 'hello_ack'); await receive(ws, 'server_snapshot')
            return ws, task, await receive(ws, 'agent_state')
        tasks = []
        try:
            await event('agent_start')
            ws, task, snapshot = await connect(); tasks.append(task)
            assert snapshot['state'] == 'RUNNING' and snapshot['session_id'] == 's'
            first = gateway.devices['network-test']
            # Partial Voice on the broken connection must never reach STT or Agent.
            await ws.incoming.put(dict(type='voice_start', voice_id='partial', sample_rate=44100, bits=16, channels=1))
            assert (await receive(ws, 'voice_state'))['state'] == 'RECORDING'
            await ws.incoming.put(dict(type='voice_chunk', voice_id='partial', sequence=0,
                audio=base64.b64encode(b'\0\0'*1764).decode()))
            await ws.incoming.put(WebSocketDisconnect())
            await asyncio.wait_for(task, 1)
            assert not first.online and first.websocket is None
            assert not voices.requests and stt.calls == 0
            assert targets.get('network-test') == 'agent:pi:s'
            assert registry.get('s').state == 'RUNNING' and adapter.calls == []
            await event('activity', activity_type='tool', activity='Running sleep 15')
            final = '离线期间完成。' if terminal == 'DONE' else '离线期间失败。'
            if terminal != 'RUNNING':
                response = await event('agent_end', failed=terminal == 'FAILED', final_text=final)
                assert response['accepted'] and response['session']['state'] == terminal
            else:
                registry.get('s').run_started_at = registry.get('s').run_started_at.replace(year=2025)
            ws2, task2, restored = await connect(); tasks.append(task2)
            assert restored['state'] == terminal
            assert restored['agent'] == 'pi' and restored['name'] == 'Selected Pi'
            assert restored['target'] == 'agent:pi:s' and restored['session_id'] == 's'
            assert restored['final_text'] == (final if terminal != 'RUNNING' else '')
            assert restored['elapsed_seconds'] >= 0
            if terminal == 'RUNNING': assert restored['elapsed_seconds'] > 10
            assert gateway.devices['network-test'].connection_id != first.connection_id
            assert targets.get('network-test') == 'agent:pi:s' and adapter.calls == []
            # Old connection-bound Voice end/decision cannot submit anything.
            await ws2.incoming.put(dict(type='voice_end', voice_id='partial'))
            await ws2.incoming.put(dict(type='voice_decision', voice_id='partial', decision='send'))
            await ws2.incoming.put(WebSocketDisconnect())
            await asyncio.wait_for(task2, 1)
            assert stt.calls == 0 and adapter.calls == []
        finally:
            for task in tasks:
                if not task.done(): task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            await voices.close()
    asyncio.run(scenario())