"""真实网关连接上的 END 终态确认与结果重交付回归。"""
import asyncio
import json
import threading
import time

import pytest
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, start, chunk


@pytest.mark.parametrize('offline_complete', [False, True])
def test_terminal_reconnect(tmp_path, monkeypatch, offline_complete):
    credential = tmp_path / 'credential.json'
    credential.write_text(json.dumps({'device_id': 'test', 'device_token': 'secret'}))
    monkeypatch.setattr(gateway, 'credential_store', DeviceCredentialStore(credential))
    targets = TargetStore(tmp_path / 'targets.json')
    targets.set('test', 'agent:pi:s')
    monkeypatch.setattr(gateway, 'target_store', targets)
    monkeypatch.setattr(gateway, 'devices', {})
    adapter = FakeAgent('pi', tmp_path)
    monkeypatch.setattr(gateway, 'adapters', {'pi': adapter})
    ready, release = threading.Event(), threading.Event()

    class Provider(FakeSTT):
        calls = 0

        async def transcribe(self, data):
            self.calls += 1
            ready.set()
            while not release.is_set():
                await asyncio.sleep(.005)
            return await super().transcribe(data)

    provider = Provider()
    monkeypatch.setattr(gateway.voices, 'provider', provider)
    monkeypatch.setattr(gateway.voices, 'context_prepare', None)
    hello = dict(type='hello', device_id='test', device_token='secret', board='synthetic', firmware='0.4.0')
    end = dict(type='voice_end', voice_id='v', total_bytes=3528)

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

    with TestClient(gateway.app) as client:
        try:
            with client.websocket_connect('/device/ws') as old:
                old.send_json(hello)
                receive(old, 'hello_ack'); receive(old, 'agent_state')
                old.send_json(start()); receive(old, 'voice_state')
                old.send_json(chunk()); receive(old, 'voice_ack')
                old.send_json(end)
                assert ready.wait(2)
                req = gateway.voices.requests['test']
                # 不消费 END 尾包便断线，模拟远端确认丢失。
            if offline_complete:
                release.set()
                deadline = time.monotonic() + 2
                while req.state != 'AWAITING_CONFIRMATION' and time.monotonic() < deadline:
                    time.sleep(.005)
                assert req.state == 'AWAITING_CONFIRMATION'
            with client.websocket_connect('/device/ws') as new:
                new.send_json(hello)
                receive(new, 'hello_ack'); receive(new, 'agent_state')
                new.send_json(dict(type='voice_status', voice_id='v'))
                ack = receive(new, 'voice_end_ack')
                assert ack == dict(type='voice_end_ack', voice_id='v', state='AWAITING_CONFIRMATION' if offline_complete else 'TRANSCRIBING')
                if offline_complete:
                    receive(new, 'voice_transcript')
                new.send_json(end)
                assert receive(new, 'voice_end_ack')['voice_id'] == 'v'
                release.set()
                transcript = receive(new, 'voice_transcript')
                new.send_json(end)
                assert receive(new, 'voice_end_ack')['state'] == 'AWAITING_CONFIRMATION'
                assert receive(new, 'voice_transcript') == transcript
                new.send_json(dict(type='voice_status', voice_id='v'))
                receive(new, 'voice_end_ack')
                assert receive(new, 'voice_transcript') == transcript
                assert provider.calls == 1 and adapter.inputs == [] and not req.pcm
        finally:
            release.set()